Link

Link

Link(
    *,
    client: Client,
    curve: AuthCurve | None,
    host: str,
    plain: AuthPlain | None,
    service_name: str,
    xpub_port: int
)

A link with a Service.

Warning

Do not instantiate directly. Use the link method to create a Link instance.

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

def __init__(  # noqa: PLR0913
    self,
    *,
    client: Client,
    curve: AuthCurve | None,
    host: str,
    plain: AuthPlain | None,
    service_name: str,
    xpub_port: int,
) -> None:
    """A link with a [Service][].

    !!! warning
        Do not instantiate directly. Use the [link][Client.link] method
        to create a [Link][] instance.
    """
    self.flag_at: Callable[[Flag], AbstractContextManager[None]] = client.flag_at
    """Reference to [flag_at][Client.flag_at]."""

    self.service_name: str = service_name
    """[str][] — The name of the [Service][] that the link is with."""

    self.timeout: float | None = None
    """[float][] | `None` — The number of seconds to wait for a reply from a *synchronous* request.

    The value is initially `None` for new links, which means that there is no limit on the wait time.
    """

    self._request: Callable[..., Future[Any]] = client._request  # pyright: ignore[reportPrivateUsage]  # noqa: SLF001
    self._link_subscriber: _LinkSubscriber = _LinkSubscriber(
        service_name=service_name,
        host=host,
        curve=curve,
        plain=plain,
        xpub_port=xpub_port,
    )
    self._client: Client = client

    self._thread: Thread = Thread(
        target=run_event_loop, daemon=True, args=(_create_async_subscriber(self._link_subscriber),)
    )
    self._thread.start()

    while self._link_subscriber.sub_socket is None:
        continue

flag_at instance-attribute

flag_at: Callable[[Flag], AbstractContextManager[None]]

Reference to flag_at.

service_name instance-attribute

service_name: str

str — The name of the Service that the link is with.

timeout instance-attribute

timeout: float | None

float | None — The number of seconds to wait for a reply from a synchronous request.

The value is initially None for new links, which means that there is no limit on the wait time.

__getattr__

__getattr__(attr: str) -> FutureOrResult

All undefined attributes are sent to the linked Service to process.

The internal wrapper function that is returned is essentially:

def wrapper(*args: Any, sync: bool = True, **kwargs: Any) -> Any | Future[Any]:
    future = request(self.service_name, attr, *args, **kwargs)
    if sync:
        return future.result(self.timeout)
    return future
return wrapper

Warning

Your Service 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.

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

def __getattr__(self, attr: str) -> FutureOrResult:
    """All undefined attributes are sent to the linked [Service][] to process.

    The internal wrapper function that is returned is essentially:

    ```python
    def wrapper(*args: Any, sync: bool = True, **kwargs: Any) -> Any | Future[Any]:
        future = request(self.service_name, attr, *args, **kwargs)
        if sync:
            return future.result(self.timeout)
        return future
    return wrapper
    ```

    !!! warning
        Your [Service][] 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.
    """

    def wrapper(*args: Any, sync: bool = True, **kwargs: Any) -> Any | Future[Any]:
        """Returns the result, if `sync=True`, otherwise a future that will eventually contain the result."""
        future = self._request(self.service_name, attr, *args, **kwargs)
        if sync:
            return future.result(self.timeout)
        return future

    return wrapper

signatures

signatures(timeout: float | None = None) -> dict[str, str]

Get the function signatures that the linked Service provides.

For example,

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

Parameters:

Name	Type	Description	Default
timeout	float | None	The maximum number of seconds to wait for the reply. If None, then the value of timeout is used.	None

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/client.py

def signatures(self, timeout: float | None = None) -> dict[str, str]:
    """Get the function signatures that the linked [Service][] provides.

    For example,

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

    Args:
        timeout: The maximum number of seconds to wait for the reply.
            If `None`, then the value of [timeout][..timeout] is used.

    Returns:
        A mapping between the function (attribute) name and the
            function signature (attribute type).
    """
    t = self.timeout if timeout is None else timeout
    future = self._request(self.service_name, "signatures")
    signatures: dict[str, str] = future.result(t)
    return signatures

subscribe

subscribe(callback: Callable[[Any], None]) -> None

Subscribe to publications from the linked Service.

Parameters:

Name	Type	Description	Default
callback	Callable[[Any], None]	A callback function to receive the published result. The callback receives a single argument, the published result, and the returned value is ignored.	required

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

def subscribe(self, callback: Callable[[Any], None]) -> None:
    """Subscribe to publications from the linked [Service][].

    Args:
        callback: A callback function to receive the published *result*. The callback
            receives a single argument, the published *result*, and the returned value is ignored.
    """
    if self._link_subscriber.sub_socket is None:
        msg = f"Cannot subscribe to {self.service_name!r}, unlinked from the service"
        raise RuntimeError(msg)

    self._link_subscriber.sub_socket.setsockopt(zmq.SUBSCRIBE, self.service_name.encode())
    self._link_subscriber.callback = callback

timeout_at

timeout_at(timeout: float | None) -> Generator[None, None, None]

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

Example

from msl.network import Client

with Client() as c:
    camera = c.link("Camera")
    camera.timeout = 5  # (1)!

    # uses a timeout of 5 seconds
    resolution = camera.get_resolution()

    with camera.timeout_at(None):  # (2)!
        # no timeout
        images = camera.capture(n=100)

    # uses a timeout of 5 seconds
    shutter_speed = camera.get_shutter_speed()

1. Use a timeout value of 5 seconds for each synchronous request.
2. Temporarily disable the timeout to capture 100 images.

Parameters:

Name	Type	Description	Default
timeout	float | None	The temporary timeout value to use for synchronous requests while within the context. Once the context exits, the value is reset to the original value.	required

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

@contextmanager
def timeout_at(self, timeout: float | None) -> Generator[None, None, None]:
    """Use as a context manager to temporarily change the [timeout][..timeout] value.

    !!! example
        ```python hl_lines="10-12"
        from msl.network import Client

        with Client() as c:
            camera = c.link("Camera")
            camera.timeout = 5  # (1)!

            # uses a timeout of 5 seconds
            resolution = camera.get_resolution()

            with camera.timeout_at(None):  # (2)!
                # no timeout
                images = camera.capture(n=100)

            # uses a timeout of 5 seconds
            shutter_speed = camera.get_shutter_speed()
        ```

        1. Use a timeout value of 5 seconds for each synchronous request.
        2. Temporarily disable the timeout to capture 100 images.

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

unlink

unlink() -> None

Unlink from the Service.

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

def unlink(self) -> None:
    """Unlink from the [Service][]."""
    if self._link_subscriber.interrupter is not None:
        self.unsubscribe()
        self._link_subscriber.interrupter()
        self._client._links.remove(self)  # pyright: ignore[reportPrivateUsage]  # noqa: SLF001
        self._thread.join()
        logger.debug("%s unlinked", self)

unsubscribe

unsubscribe() -> None

Unsubscribe from receiving publications from the linked Service.

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

def unsubscribe(self) -> None:
    """Unsubscribe from receiving publications from the linked [Service][]."""
    self._link_subscriber.callback = None
    if self._link_subscriber.sub_socket is not None:
        self._link_subscriber.sub_socket.setsockopt(zmq.UNSUBSCRIBE, self.service_name.encode())