Service

To implement your own Service, just inherit from the Service class, create the methods/attributes that you want your Service to provide and then run your script.

Warning

Your subclass cannot have method/attribute names that are the same as those found in the Link class, otherwise a request will not be sent to your Service because the Link instance will handle the method call.

import asyncio
from msl.network import Service


class Camera(Service):
    def __init__(self) -> None:
        """Allows for a camera to be controllable via the network."""
        super().__init__()
        self._camera = ...  # (1)!
        self.version: str = "1.3"  # (2)!

    def get_resolution(self) -> tuple[int, int]:
        """Get the image resolution."""
        return self._camera.resolution

    def set_resolution(self, width: int, height: int) -> None:
        """Set the image resolution."""
        self._camera.resolution = (width, height)

    async def stream(self) -> None:
        """This coroutine is also run in the event loop."""
        while True:
            self.publish(self._camera.capture())
            await asyncio.sleep(1)


if __name__ == "__main__":
    c = Camera()
    c.add_tasks(c.stream())  # (3)!
    c.connect()  # (4)!

1. Access the camera interface.
2. A Client can also send a request for an attribute value, not just callable methods.
3. Add the stream method to the event loop of the Service to publish images every second.
4. Connect to the Broker. Runs the event loop forever.

Service

Service(
    *,
    name: str | None = None,
    host: str = "127.0.0.1",
    port: int = BROKER_PORT,
    flag: Flag = PICKLE,
    curve: AuthCurve | None = None,
    plain: AuthPlain | None = None,
    xsub_port: int | None = None,
    ignore_attributes: str | Iterable[str] | None = None
)

Base class for a Service.

Parameters:

Name	Type	Description	Default
name	str | None	The name of the Service that a Client will use to link with. If not specified, the class name is used.	None
host	str	The hostname (or IP address) that the Broker is running on.	'127.0.0.1'
port	int	The network port that the Broker is running on.	BROKER_PORT
flag	Flag	The serialisation and compression algorithms to apply to a reply before sending the message.	PICKLE
curve	AuthCurve | None	The CURVE authentication to use.	None
plain	AuthPlain | None	The PLAIN authentication to use.	None
xsub_port	int | None	The port on the Broker that is subscribed to publications. Typically, this value is port + 2 and does not need to be specified.	None
ignore_attributes	str | Iterable[str] | None	The names of the attributes to not include in the signatures map. See ignore_attributes for more details.	None

Source code in src/msl/network/service.py

def __init__(  # noqa: PLR0913
    self,
    *,
    name: str | None = None,
    host: str = "127.0.0.1",
    port: int = BROKER_PORT,
    flag: Flag = Flag.PICKLE,
    curve: AuthCurve | None = None,
    plain: AuthPlain | None = None,
    xsub_port: int | None = None,
    ignore_attributes: str | Iterable[str] | None = None,
) -> None:
    """Base class for a Service.

    Args:
        name: The name of the Service that a [Client][msl.network.client.Client] will
            use to [link][msl.network.client.Client.link] with. If not specified,
            the class name is used.
        host: The hostname (or IP address) that the [Broker][] is running on.
        port: The network port that the [Broker][] is running on.
        flag: The serialisation and compression algorithms to apply to a reply before
            sending the message.
        curve: The [CURVE](https://rfc.zeromq.org/spec/25/) authentication to use.
        plain: The [PLAIN](https://rfc.zeromq.org/spec/24/) authentication to use.
        xsub_port: The port on the [Broker][] that is subscribed to publications.
            Typically, this value is `port + 2` and does not need to be specified.
        ignore_attributes: The names of the attributes to not include in the
            [signatures][..signatures] map. See [ignore_attributes][..ignore_attributes]
            for more details.
    """
    self.flag: Flag = flag
    """The serialisation and compression algorithms to apply to a reply or publication
    before sending the message."""

    self._service_id: bytes = f"Service[{os.urandom(8).hex()}]".encode()
    self._service_name: str = name or self.__class__.__name__
    self._host_port: tuple[str, int] = (host, port)
    self._context: Context = Context()
    self._poller: Poller = Poller()
    self._interrupter: Interrupter | None = None
    self._dealer: Socket | None = None
    self._monitor: Socket | None = None
    self._curve: AuthCurve | None = curve
    self._plain: AuthPlain | None = plain
    self._tasks: list[Awaitable[None]] = []
    self._loop_thread_id: int = -1
    self._xsub_port: int = xsub_port or port + 2  # Service connects with PUBlish: PUB -> XSUB
    self._pub_queue: asyncio.Queue[bytes] | None = None

    # Just define type annotations
    self._loop: asyncio.AbstractEventLoop

    # Python 3.8 and 3.9 require an asyncio event loop to be running to create an asyncio.Event instance
    self.connected: asyncio.Event
    """An [Event][asyncio.Event] object that represents whether the Service is connected to the [Broker][]."""

    if curve is not None and plain is not None:
        msg = "Cannot use both PLAIN and CURVE authentication, select only one authentication mechanism"
        raise ValueError(msg)

    self._ignore_attributes: set[str] = {
        "add_tasks",
        "before_disconnect",
        "connect",
        "connected",
        "disconnect",
        "flag",
        "flag_at",
        "ignore_attributes",
        "publish",
        "signatures",
    }

    if ignore_attributes is not None:
        if isinstance(ignore_attributes, str):
            self.ignore_attributes(ignore_attributes)
        else:
            self.ignore_attributes(*ignore_attributes)

connected instance-attribute

connected: Event

An Event object that represents whether the Service is connected to the Broker.

flag instance-attribute

flag: Flag = flag

The serialisation and compression algorithms to apply to a reply or publication before sending the message.

add_tasks

add_tasks(*aws: Awaitable[None]) -> None

Add tasks to run in the event loop.

Parameters:

Name	Type	Description	Default
aws	Awaitable[None]	Awaitables that will run in the event loop.	()

Source code in src/msl/network/service.py

def add_tasks(self, *aws: Awaitable[None]) -> None:
    """Add tasks to run in the [event loop][asyncio-event-loop].

    Args:
        aws: Awaitables that will run in the [event loop][asyncio-event-loop].
    """
    self._tasks.extend(aws)

before_disconnect

before_disconnect() -> None

Called immediately before the Service disconnects from the Broker.

Override this method to, for example, clean up resources that are used by the Service.

Source code in src/msl/network/service.py

def before_disconnect(self) -> None:
    """Called immediately before the Service disconnects from the [Broker][].

    Override this method to, for example, clean up resources that are used by the Service.
    """

connect

connect() -> None

Connect (or reconnect) to the Broker.

Calling this method will run the event loop until all tasks have completed.

Source code in src/msl/network/service.py

def connect(self) -> None:
    """Connect (or reconnect) to the [Broker][].

    Calling this method will run the [event loop][asyncio-event-loop] until all tasks have completed.
    """
    try:
        run_event_loop(self._gather())
    except KeyboardInterrupt:  # pragma: no cover
        pass
    finally:
        run_event_loop(self._handle_disconnect())
        self.before_disconnect()
        self.disconnect()
        logger.debug("%s event loop closed", self._service_name)

disconnect

disconnect() -> None

Disconnect from the Broker.

Source code in src/msl/network/service.py

def disconnect(self) -> None:
    """Disconnect from the [Broker][]."""
    if self._pub_queue is not None and self._interrupter is not None:
        self._interrupter()

    if self._interrupter is None or self._dealer is None or self._monitor is None:
        return

    self.connected.clear()
    self._poller.unregister(self._dealer)
    self._poller.unregister(self._interrupter.receiver)
    self._poller.unregister(self._monitor)
    self._interrupter.close()
    self._dealer.disable_monitor()
    self._dealer.close(linger=0)
    self._interrupter = None
    self._dealer = None
    self._monitor = None
    logger.debug("%s disconnected", self._service_name)

flag_at

flag_at(flag: Flag) -> Generator[None, None, None]

Use as a context manager to temporarily change the flag value.

Example

from msl.network import Flag, Service

class Camera(Service):

    def __init__(self) -> None:
        """By default, use JSON to serialise all replies (no compression)."""
        super().__init__(flag=Flag.JSON)
        self._camera = ...  # Access the camera interface

    def get_resolution(self) -> tuple[int, int]:
        """Returns the (width, height) of a captured image.

        The reply is serialised using JSON without compression.
        """
        return self._camera.resolution

    def capture(self) -> bytes:
        """Capture an image and return the image bytes.

        For this method, compressed bytes without serialisation is returned.
        """
        with self.flag_at(Flag.ZLIB):  # (1)!
            return self._camera.capture()

1. Temporarily change the serialisation and compression algorithms to send the reply.

Parameters:

Name	Type	Description	Default
flag	Flag	The temporary flag to use while within the context. Once the context exits, the value is reset to the original value.	required

Source code in src/msl/network/service.py

@contextmanager
def flag_at(self, flag: Flag) -> Generator[None, None, None]:
    """Use as a context manager to temporarily change the [flag][..flag] value.

    !!! example
        ```python hl_lines="22 23"
        from msl.network import Flag, Service

        class Camera(Service):

            def __init__(self) -> None:
                \"\"\"By default, use JSON to serialise all replies (no compression).\"\"\"
                super().__init__(flag=Flag.JSON)
                self._camera = ...  # Access the camera interface

            def get_resolution(self) -> tuple[int, int]:
                \"\"\"Returns the (width, height) of a captured image.

                The reply is serialised using JSON without compression.
                \"\"\"
                return self._camera.resolution

            def capture(self) -> bytes:
                \"\"\"Capture an image and return the image bytes.

                For this method, compressed bytes without serialisation is returned.
                \"\"\"
                with self.flag_at(Flag.ZLIB):  # (1)!
                    return self._camera.capture()
        ```

        1. Temporarily change the serialisation and compression algorithms to send the reply.

    Args:
        flag: The temporary flag to use while within the context. Once the
            context exits, the value is reset to the original value.
    """
    original = self.flag
    self.flag = flag
    try:
        yield
    finally:
        self.flag = original

ignore_attributes

ignore_attributes(*names: str) -> None

Ignore attributes from being added to the signatures map.

There are a few reasons why you may want to call this method:

• If you do not want an attribute to be made publicly known that it exists; however, a Client can still receive a reply for an ignored attribute that it requested.
• If warnings are logged indicating that the signature of an attribute cannot be found and you prefer not to see the warnings (this could result from a Service using multiple inheritance).

Private attributes (i.e., attribute names that start with an underscore) are automatically ignored and if a Client sends a request for a private attribute it will receive an error message as the reply.

Parameters:

Name	Type	Description	Default
names	str	The names of the attributes to exclude from the signatures map.	()

Source code in src/msl/network/service.py

def ignore_attributes(self, *names: str) -> None:
    """Ignore attributes from being added to the [signatures][..signatures] map.

    There are a few reasons why you may want to call this method:

    * If you do not want an attribute to be made publicly known that it exists; however,
      a [Client][msl.network.client.Client] can still receive a reply for an ignored attribute
      that it requested.
    * If warnings are logged indicating that the signature of an attribute cannot be found and you
      prefer not to see the warnings (this could result from a Service using multiple inheritance).

    Private attributes (i.e., attribute names that start with an underscore) are automatically
    ignored and if a [Client][msl.network.client.Client] sends a request for a private attribute
    it will receive an error message as the reply.

    Args:
        names: The names of the attributes to exclude from the [signatures][..signatures] map.
    """
    self._ignore_attributes.update(names)

publish

publish(value: Any, flag: Flag | None = None) -> None

Publish a value to all subscribers.

Parameters:

Name	Type	Description	Default
value	Any	The value to publish.	required
flag	Flag | None	The flag to use to convert the value to bytes before sending the message. If None, the flag defined by flag is used.	None

Source code in src/msl/network/service.py

def publish(self, value: Any, flag: Flag | None = None) -> None:
    """Publish a value to all subscribers.

    Args:
        value: The value to publish.
        flag: The flag to use to convert the `value` to bytes before sending the message.
            If `None`, the flag defined by [flag][..flag] is used.
    """
    if self._pub_queue is None:
        msg = "Event loop not running, cannot publish value"
        raise RuntimeError(msg)

    data = Reply(id=0, ok=True, result=value).to_bytes(flag or self.flag)
    if get_ident() == self._loop_thread_id:
        self._pub_queue.put_nowait(data)
    else:
        _ = self._loop.call_soon_threadsafe(self._pub_queue.put_nowait, data)

signatures

signatures() -> dict[str, str]

Returns the function signatures that the Service provides.

For example,

{
    "get_resolution": "() -> tuple[int, int]",
    "set_resolution": "(width: int, height: int) -> None",
    "stream": "() -> None",
    "version": "() -> str"
}

Returns:

Type	Description
dict[str, str]	A mapping between the function (attribute) name and the function signature (attribute type).

Source code in src/msl/network/service.py

def signatures(self) -> dict[str, str]:
    """Returns the function signatures that the Service provides.

    For example,

    ```json
    {
        "get_resolution": "() -> tuple[int, int]",
        "set_resolution": "(width: int, height: int) -> None",
        "stream": "() -> None",
        "version": "() -> str"
    }
    ```

    Returns:
        A mapping between the function (attribute) name and the
            function signature (attribute type).
    """
    signature_map: dict[str, str] = {}
    for name in dir(self):
        if name.startswith("_") or (name in self._ignore_attributes):
            continue

        try:
            attrib = getattr(self, name)
        except Exception as e:  # noqa: BLE001
            # This can happen if the Service is also a subclass of
            # another class (e.g., the PiCamera class) and the other
            # class defines some of its attributes using the builtin
            # property function, e.g., property(fget, fset, fdel, doc),
            # and defines fget=None or if the getattr() function
            # executes code, like PiCamera.frame does, which raises
            # a custom exception if the camera is not running.
            logger.warning("%s [attribute=%r]", e, name)
            continue

        try:
            signature_map[name] = str(inspect.signature(attrib)).replace("'", "")
        except TypeError:
            # Then the attribute is not a callable object
            signature_map[name] = f"() -> {attrib.__class__.__name__}"
        except ValueError as e:
            # Cannot get the signature of the callable object.
            # This can happen if the Service is also a subclass of
            # some other object, for example a Qt class.
            logger.warning("%s [attribute=%r]", e, name)

    return signature_map