Communication between modules (pyobs.comm)

The Comm object is responsible for all communication between modules (see pyobs.modules). The base class for all implementations is Comm.

The user usually only has contact with the Comm object when writing the configuration for an existing module or when developing a new module that needs to communicate with other modules.

In a configuration file, the Comm object is defined at top-level like this:

comm:
    class: pyobs.comm.xmpp.XmppComm

Except for a single parameter defined in Comm’s constructor, all parameters are defined in derived classes.

The most convenient way for getting access to other modules’ method is by using a Proxy object, which can easily be obtained by using the proxy() method or the [] operator like this (if the module named ‘camera’ implements the ICamera interface):

camera = comm['camera']
camera.expose().wait()

Note that camera is now not of type ICamera.

Each Module that was configured with a Comm object (see modules) has an attribute comm for easy access.

There is currently one one implementation of the Comm interface:

  • XmppComm uses the XMPP protocol for communication.

See also

Module modules

Description for modules, to which Comm objects are usually assigned.

Comm

class Comm(cache_proxies: bool = True)

Base class for all Comm modules in pyobs.

Creates a comm module.

cast_to_real_post(value: Any, annotation: Any | None = None) Tuple[bool, Any][source]

Special treatment of single parameters when converting them after being sent via Comm.

Parameters:
  • value – Value to be treated.

  • annotation – Annotation for value.

Returns:

A tuple containing a tuple that indicates whether this value should be further processed and a new value.

cast_to_real_pre(value: Any, annotation: Any | None = None) Tuple[bool, Any][source]

Special treatment of single parameters when converting them after being sent via Comm.

Parameters:
  • value – Value to be treated.

  • annotation – Annotation for value.

Returns:

A tuple containing a tuple that indicates whether this value should be further processed and a new value.

cast_to_simple_post(value: Any, annotation: Any | None = None) Tuple[bool, Any][source]

Special treatment of single parameters when converting them to be sent via Comm.

Parameters:
  • value – Value to be treated.

  • annotation – Annotation for value.

Returns:

A tuple containing a tuple that indicates whether this value should be further processed and a new value.

cast_to_simple_pre(value: Any, annotation: Any | None = None) Tuple[bool, Any][source]

Special treatment of single parameters when converting them to be sent via Comm.

Parameters:
  • value – Value to be treated.

  • annotation – Annotation for value.

Returns:

A tuple containing a tuple that indicates whether this value should be further processed and a new value.

property clients: List[str]

Returns list of currently connected clients.

Returns:

(list) List of currently connected clients.

async clients_with_interface(interface: Type[Interface]) List[str][source]

Returns list of currently connected clients that implement the given interface.

Parameters:

interface – Interface to search for.

Returns:

(list) List of currently connected clients that implement the given interface.

async close() None[source]

Close module.

async execute(client: str, method: str, annotation: Dict[str, Any], *args: Any) Any[source]

Execute a given method on a remote client.

Parameters:
  • client (str) – ID of client.

  • method (str) – Method to call.

  • annotation – Method annotation.

  • *args – List of parameters for given method.

Returns:

Passes through return from method call.

async get_interfaces(client: str) List[Type[Interface]][source]

Returns list of interfaces for given client.

Parameters:

client – Name of client.

Returns:

List of supported interfaces.

Raises:

IndexError – If client cannot be found.

log_message(entry: LogEvent) None[source]

Send a log message to other clients.

Parameters:

entry (LogEvent) – Log event to send.

property module: Module | None

The module that this Comm object is attached to.

property name: str | None

Name of this client.

async open() None[source]

Open module.

async proxy(name_or_object: str | object, obj_type: Type[ProxyType] | None = None) Any | ProxyType[source]

Returns object directly if it is of given type. Otherwise get proxy of client with given name and check type.

If name_or_object is an object:
  • If it is of type (or derived), return object.

  • Otherwise raise exception.

If name_name_or_object is string:
  • Create proxy from name and raise exception, if it doesn’t exist.

  • Check type and raise exception if wrong.

  • Return object.

Parameters:
  • name_or_object – Name of object or object itself.

  • obj_type – Expected class of object.

Returns:

Object or proxy to object.

Raises:

ValueError – If proxy does not exist or wrong type.

async register_event(event_class: Type[Event], handler: Callable[[Event, str], Coroutine[Any, Any, bool]] | None = None) None[source]

Register an event type. If a handler is given, we also receive those events, otherwise we just send them.

Parameters:
  • event_class – Class of event to register.

  • handler – Event handler method.

async safe_proxy(name_or_object: str | object, obj_type: Type[ProxyType] | None = None) Any | ProxyType | None[source]

Calls proxy() in a safe way and returns None instead of raising an exception.

async send_event(event: Event) None[source]

Send an event to other clients.

Parameters:

event (Event) – Event to send

Proxy

class Proxy(comm: Comm, client: str, interfaces: List[Type[Interface]])

A proxy for remote pyobs modules.

Creates a new proxy.

Parameters:
  • comm – Comm object to use for connection.

  • client – Name of client to connect to.

  • interfaces – List of interfaces supported by client.

async execute(method: str, *args: Any, **kwargs: Any) Any[source]

Execute a method on the remote client.

Parameters:
  • method – Name of method to call.

  • *args – Parameters for method call.

  • **kwargs – Parameters for method call.

Returns:

Result of method call.

interface_method(method: str) Any[source]

Returns the method of the given name from the interface and not from the object itself.

Parameters:

method – Name of method.

Returns:

The interface method.

property interfaces: List[Type[Interface]]

List of interfaces.

property method_names: List[str]

List of method names.

property name: str

Name of the client.

signature(method: str) Signature[source]

Returns the signature of a given method.

Parameters:

method – Name of the method.

Returns:

Signature of the given method.

XmppComm (based on XMPP)

class XmppComm(jid: str | None = None, user: str | None = None, domain: str | None = None, resource: str = 'pyobs', password: str = '', server: str | None = None, use_tls: bool = False, *args: Any, **kwargs: Any)

A Comm class using XMPP.

This Comm class uses an XMPP server (e.g. ejabberd) for communication between modules. Essentially required for a connection to the server is a JID, a JabberID. It can be specified in the configuration like this:

comm:
    class: pyobs.comm.xmpp.XmppComm
    jid:  someuser@example.com/pyobs

Using this, pyobs tries to connect to example.com as user someuser with resource pyobs. Since pyobs is the default resource, it can be omitted:

jid:  someuser@example.com

Alternatively, one can split the user, domain, and resource (if required) into three different parameters:

user: someuser
domain: example.com

This comes in handy, if one wants to put the basic Comm configuration into a separate file. Imagine a _comm.yaml in the same directory as the module config:

comm_cfg: &comm
    class: pyobs.comm.sleekxmpp.XmppComm
    domain: example.com

Now in the module configuration, one can simply do this:

{include _comm.yaml}

comm:
    <<: *comm
    user: someuser
    password: supersecret

This allows for a super easy change of the domain for all configurations, which especially makes developing on different machines a lot easier.

The server parameter can be used, when the server’s hostname is different from the XMPP domain. This might, e.g., be the case, when connecting to a server via SSH port forwarding:

jid:  someuser@example.com/pyobs
server: localhost:52222

Finally, always make sure that use_tls is set according to the server’s settings, i.e. if it uses TLS, this parameter must be True, and False otherwise. Cryptic error messages will follow, if one does not set this properly.

Create a new XMPP Comm module.

Either a fill JID needs to be provided, or a set of user/domian/resource, from which a JID is built.

Parameters:
  • jid – JID to connect as.

  • user – Username part of the JID.

  • domain – Domain part of the JID.

  • resource – Resource part of the JID.

  • password – Password for given JID.

  • server – Server to connect to. If not given, domain from JID is used.

  • use_tls – Whether or not to use TLS.

cast_to_real_post(value: Any, annotation: Any | None = None) Tuple[bool, Any][source]

Special treatment of single parameters when converting them after being sent via Comm.

Parameters:
  • value – Value to be treated.

  • annotation – Annotation for value.

Returns:

A tuple containing a tuple that indicates whether this value should be further processed and a new value.

cast_to_simple_pre(value: Any, annotation: Any | None = None) Tuple[bool, Any][source]

Special treatment of single parameters when converting them to be sent via Comm.

Parameters:
  • value – Value to be treated.

  • annotation – Annotation for value.

Returns:

A tuple containing a tuple that indicates whether this value should be further processed and a new value.

property client: XmppClient

Returns the XMPP client.

Returns:

The XMPP client.

property clients: List[str]

Returns list of currently connected clients.

Returns:

(list) List of currently connected clients.

async close() None[source]

Close connection.

async execute(client: str, method: str, annotation: Dict[str, Any], *args: Any) Any[source]

Execute a given method on a remote client.

Parameters:
  • client (str) – ID of client.

  • method (str) – Method to call.

  • annotation – Method annotation.

  • *args – List of parameters for given method.

Returns:

Passes through return from method call.

async get_interfaces(client: str) List[Type[Interface]][source]

Returns list of interfaces for given client.

Parameters:

client – Name of client.

Returns:

List of supported interfaces.

Raises:

IndexError – If client cannot be found.

property name: str | None

Name of this client.

async open() None[source]

Open the connection to the XMPP server.

Returns:

Whether opening was successful.

async send_event(event: Event) None[source]

Send an event to other clients.

Parameters:

event (Event) – Event to send

DbusComm (based on D-Bus)