ubii.framework.services module¶

class ubii.framework.services.ServiceConnection¶

Bases: ABC

A ServiceConnection is a two-way request-reply connection. If the connection is present the only defined public API method is supposed to send a message in the ServiceRequest format, await a response from the master node and return it to the caller.

abstract async send(request: ServiceRequest) → ServiceReply¶

Send a ServiceRequest message, and receive a ServiceReply, according to the protobuf specifications. Can be implemented with any communication transports the master node supports.

Parameters:: request – ServiceRequest protobuf message (wrapper)

class ubii.framework.services.ServiceCallFactory(*args, **kwargs)¶

Bases: Protocol[T_Service_Cov]

__call__(mapping: Service) → T_Service_Cov¶

ServiceCallFactory objects need to have this call signature

Parameters:: mapping – instance of a ubii.proto.Service wrapper for a Service message
Returns:: A type of ServiceCall object

class ubii.framework.services.ServiceMap(mapping=None, *, service_call_factory: ServiceCallFactory[T_Service], **kwargs)¶

Bases: ServiceList, Mapping[str, T_Service], Generic[T_Service]

A ServiceMap is a wrapper around a ServiceList proto message which provides a mapping for the Service messages by topic.

An adapter like this is needed because the master node advertises its services in a ubii.proto.ServiceList message, but the semantics are to access them by topic – typically with the default topics provided by the master node.

Additionally, a ServiceMap converts the respective ubii.proto.Service for a topic to a ServiceCall (the service_call_factory argument can be supplied during creation, it has to be a callable with signature matching a ServiceCallFactory, and will be called with the Service message as only argument). The results are cached.

If you change the protobuf contents of a ServiceMap object after creation (e.g. when you get a new ServiceList from the master node), make sure to invalidate the cache by calling cache_clear(). This is automatically done if you assign to the elements attribute directly, but if you e.g. copy the contents of a ServiceList message into a ServiceMap by means of copy_from(), you need to invalidate the cache manually.

Until the message format changes and the master node advertises its services in a mapping, use this adapter.

Example

Create a ServiceMap – the example sets the transport for the ServiceCall to None, real code needs to use a valid ServiceConnection

>>> from ubii.framework.services import ServiceMap, ServiceCall
>>> from functools import partial
>>> from ubii.proto import Service
>>> services = [Service(topic=topic) for topic in ['foo', 'bar', 'foobar']]
>>> service_map = ServiceMap(service_call_factory=partial(ServiceCall, transport=None), elements=services)

When accessing keys that have corresponding services in the elements of the map, a ServiceCall is returned

>>> ...
>>> service = service_map['foo']
>>> type(service)
<class 'ubii.framework.services.ServiceCall'>
>>> service.topic
'foo'

Services that are not defined in the elements raise a KeyError as expected

>>> ...
>>> service_map['thing']
Traceback <...>
KeyError: 'found no services for topic thing'

The created ServiceCall will be cached

>>> ...
>>> service is service_map['foo']
True

If you assign a Service list to elements the cache will be cleared implictily and a new but equivalent ServiceCall is created for the topic

>>> ...
>>> service_map.elements = services
>>> service is service_map['foo']
False
>>> service == service_map['foo']
True

The elements can change implicitly if the internal protobuf message changes. In this case the code also needs to call cache_clear() explicitly

>>> ...
>>> service_map.elements = services
>>> from ubii.proto import ServiceList
>>> service_list = ServiceList(elements=[service for service in services if not service.topic == 'foo'])
>>> type(service_list).copy_from(service_map, service_list)
>>> service is service_map['foo']
True
>>> service_map.elements
[
topic: "bar",
topic: "foobar"
]

As you can see the foo service is cached, although no foo topic is in the elements anymore. After cache invalidation, accessing 'foo' topic raises a KeyError as expected

>>> ...
>>> service_map.elements = services
>>> service_map.cache_clear()
>>> service is service_map['foo']
Traceback <...>
KeyError: 'found no services for topic foo'

elements¶

RepeatedField of type Service – inherited from ServiceList

Type:: proto.fields.RepeatedField

__deepcopy__(memo)¶: Not provided by the proto plus base class. Needed because proto plus base implements __getattr__, so when you try to deepcopy a ServiceMap (e.g. dataclasses.dataclass.replace() with a dataclass that contains a ServiceMap) you get infinite recursion if you don’t also implement __deepcopy__

__init__(mapping=None, *, service_call_factory: ServiceCallFactory[T_Service], **kwargs)¶

Parameters:

mapping – used to initialize the protobuf wrapper, can also itself be a ServiceList wrapper
service_call_factory – used to convert the protobuf messages to ServiceCall objects
**kwargs – passed to protobuf wrapper initialization

cache_clear()¶: Clear cached service calls. Implicitly used if a new list of Services is assigned to the elements field, need to call manually if assignment is done implicitly.

See also

ServiceMap code example – shows caching behaviour in detail

elements: MutableSequence[Service]¶

class ubii.framework.services.DefaultServiceMap(*args, defaults: MutableMapping[str, str] | None = None, **kwargs)¶

Bases: ServiceMap[T_Service]

Automatically creates Services for missing topics (like a defaultdict) Takes and optional mapping \(name \rightarrow topic\), which can be accessed as attributes of the ServiceMap.

Example

Make a map with defaults

>>> from ubii.framework.services import ServiceCall, DefaultServiceMap
>>> from functools import partial
>>> service_map = DefaultServiceMap(
...     service_call_factory=lambda service: ServiceCall(topic=service.topic, transport=None),
...     defaults = {'foo': 'services/foo', 'bar': 'services/bar'},
... )

Now the services for which there are defined defaults can be accessed as an attribute, and will be automatically added to the elements

>>> ...
>>> service_map.elements
[]
>>> service = service_map.foo
>>> type(service)
<class 'ubii.framework.services.ServiceCall'>
>>> service.topic
'services/foo'
>>> service_map.elements
[topic: "services/foo"]

If no default is set, an AttributeError is raised as expected

>>> ...
>>> service_map.boo
Traceback <...>
AttributeError: Unknown field for DefaultServiceMap: boo

If the framework is used in debug mode, the AttributeError contains information about possible matches in the defaults

>>> ...
>>> from ubii.framework import debug
>>> debug(True)
True
>>> service_map.boo
Traceback <...>
AttributeError: Unknown field for DefaultServiceMap: boo

The above exception was the direct cause of the following exception:
Traceback <...>
AttributeError: DefaultServiceMap has no attribute 'boo'. Best match[es] in default topics: 'foo'

services are only automatically created when the key is present as a value in defaults

>>> ...
>>> service_map['bar']
Traceback <...>
KeyError: 'found no services for topic boo'
>>> assert service_map['services/bar']
True

elements¶

RepeatedField of type Service – inherited from ServiceList – inherited from ServiceMap

Type:: proto.fields.RepeatedField

property defaults: MutableMapping[str, str]¶

Reference to the mapping that was passed as defaults during initialization.

Basically, attribute access is equivalent to item access for the corresponding value inside this mapping (if present)

service_map = DefaultServiceMap(...)
assert 'some_attr' in service_map.defaults
service_map.some_attr == service_map[service_map.defaults['some_attr']]

elements: MutableSequence[Service]¶

class ubii.framework.services.ServiceCall(mapping=None, *, transport: ServiceConnection, **kwargs)¶

Bases: Service

A ServiceCall is a callable that can be represented as a ubii.proto.Service protobuf message.

topic¶

Field of type STRING – inherited from Service

Type:: proto.fields.Field

request_message_format¶

Field of type STRING – inherited from Service

Type:: proto.fields.Field

response_message_format¶

Field of type STRING – inherited from Service

Type:: proto.fields.Field

tags¶

RepeatedField of type STRING – inherited from Service

Type:: proto.fields.RepeatedField

description¶

Field of type STRING – inherited from Service

Type:: proto.fields.Field

__init__(mapping=None, *, transport: ServiceConnection, **kwargs)¶

Parameters:

mapping – for ubii.proto.Service initialization
transport – connection used to transfer messages
**kwargs – for ubii.proto.Service initialization

property transport¶: Reference to the connection used

__call__(**payload) → Awaitable[ubii.proto.ServiceReply]¶

send the ubii.proto.ServiceRequest defined by the payload keyword arguments and the topic of this ServiceCall using the transport.

The reply is awaited and error handling is applied.

Example

You can get ServiceCall objects from a running client’s Services behaviour

>>> from ubii.node import connect_client
>>> from ubii.framework.client import Services
>>> import asyncio
>>> service_call = None
>>> reply = None
>>> async def main():
...     global service_call
...     global reply
...     async with connect_client() as client:
...             assert client.implements(Services)
...             service_call = client[Services].service_map.server_config
...             reply = await service_call()
...
>>> asyncio.run(main())
>>> service_call
topic: "/services/server_configuration"
response_message_format: "ubii.servers.Server"
>>> reply
server {
  id: "f491e47e-e591-4961-b4ca-8e1142175ae8"
  name: "master-node"
  ...
}

Parameters:: **payload – will be converted to a ubii.proto.ServiceRequest message with topic defined by this calls topic, i.e. typically only one keyword argument will be used, to define the field for the type oneof group. Values can be mappings or protobuf wrappers.
Returns:: the reply from the master node
Raises:: ubii.framework.errors.UbiiError – if master node replies with an error message

This callable had the hook decorator applied. Original signature: async def __call__(self, **payload) -> 'ubii.proto.ServiceReply'

classmethod register_decorator(decorator, instance: object | None = None) → None¶

Since __call__() is a hook, you can easily decorate it. There are two ways to do that – either directly use the fact that __call__() is a hook

from ubii.framework.services import ServiceCall
ServiceCall.__call__.register_decorator(decorator)

Or use this method for convenience

from ubii.framework.services import ServiceCall
ServiceCall.register_decorator(decorator)

Parameters:

decorator – Some callable to decorate __call__()
instance – only register decorator for this instance – optional

ubii.framework.services module¶

Navigation

Related Topics