Modules (pyobs.modules)

A Module is the building block of a pyobs system. Each module represents one component of the observatory — a camera, a telescope, a scheduler, a weather monitor, and so on — and runs as its own process, configured from a YAML file.

Module inherits from Object and adds the communication layer that allows modules to call each other’s methods across a network.

Writing a minimal module

A module is a class that inherits from Module (plus any interfaces it implements):

import asyncio
import logging
from typing import Any
from pyobs.modules import Module

log = logging.getLogger(__name__)


class MyModule(Module):
    """A minimal example module."""

    def __init__(self, interval: int = 10, **kwargs: Any):
        Module.__init__(self, **kwargs)
        self._interval = interval
        self.add_background_task(self._run)

    async def open(self) -> None:
        await Module.open(self)
        # connect to hardware or subscribe to events here

    async def _run(self) -> None:
        while True:
            log.info("Running...")
            await asyncio.sleep(self._interval)

The matching YAML configuration:

class: mypackage.MyModule
interval: 5

comm:
  class: pyobs.comm.xmpp.XmppComm
  jid: mymodule@my.domain.com

timezone: UTC
location:
  longitude: 10.0
  latitude: 51.0
  elevation: 200.0

vfs:
  class: pyobs.vfs.VirtualFileSystem
  roots:
    cache:
      class: pyobs.vfs.LocalFile
      root: /data

Note

Always forward **kwargs to Module.__init__. This is how comm, vfs, timezone, and location are passed down from the YAML configuration.

Startup and `ModuleState.STARTING`

A module starts in STARTING and stays there for the whole open() override chain — including everything a subclass’s own open() does after calling await Module.open(self), e.g. connecting to hardware. While STARTING, execute() rejects any call from another module except get_permitted_methods/reset_error, raising ModuleStartingError, and the module stays invisible to XMPP peer discovery (see Modules reject RPC calls until fully started). This matters if your own open() calls into another module via self.proxy(...) — that module may itself still be STARTING.

A module doesn’t call startup() on itself; whatever launches it (Application, the normal pyobs/pyobsd entry point, or MultiModule) calls startup(), which runs open() and then transitions the module to READY. You only need to call startup() yourself when opening a module outside of those two (a test, a standalone script) — open() alone leaves it in STARTING indefinitely. It’s named startup() rather than start() because start() is already IStartStop’s abstract RPC method — a plain start() here would be silently shadowed by any module implementing that interface.

Interfaces

The functionality a module exposes for remote calls is defined by the interfaces it declares. Interfaces are abstract base classes (defined in pyobs.interfaces) that specify method signatures. A module implementing ICamera, for example, advertises that it can take images:

from pyobs.interfaces import ICamera
from pyobs.utils.enums import ImageType

class MyCamera(Module, ICamera):
    async def grab_data(self, broadcast: bool = True, **kwargs: Any) -> str:
        ...

Other modules can then obtain a proxy to MyCamera and call grab_data remotely, without knowing which machine the camera is running on:

async with self.proxy("camera", ICamera) as camera:
    filename = await camera.grab_data()

See Interfaces (pyobs.interfaces) for the full list of available interfaces.

Communicating between modules

Modules communicate via the comm property, which provides access to the Comm object. The most common use is obtaining a proxy to another module:

async def open(self) -> None:
    await Module.open(self)
    async with self.proxy("telescope", ITelescope) as telescope:
        await telescope.move_radec(ra=83.8, dec=-5.4)

Modules can also subscribe to and emit Events (pyobs.events):

async def open(self) -> None:
    await Module.open(self)
    await self.comm.register_event(NewImageEvent, self._on_new_image)

async def _on_new_image(self, event: NewImageEvent, sender: str) -> bool:
    log.info("New image from %s: %s", sender, event.filename)
    return True

The `@timeout` decorator

Methods exposed via an interface should declare an expected timeout, so that the comm layer can raise a helpful error if a call takes too long. Use the timeout() decorator:

from pyobs.modules import timeout

class MyCamera(Module, ICamera):
    @timeout(30)                   # fixed 30 second timeout
    async def grab_data(self, broadcast: bool = True, **kwargs: Any) -> str:
        ...

    @timeout("exposure_time + 10") # expression using method parameters
    async def expose(self, exposure_time: float, **kwargs: Any) -> str:
        ...

The expression form is evaluated with the method’s keyword arguments as variables.

API reference

class Module(label: str | None = None, own_comm: bool = True, additional_config_variables: list[str] | None = None, acl: dict[str, Any] | None = None, **kwargs: Any)

Bases: Object, IModule, IConfig

Base class for all pyobs modules.

Parameters:

label – Label for module. If None, name is used.
own_comm – If True, module owns comm and opens/closes it.
additional_config_variables – List of additional variable names available to remote config getter/setter.
acl – Access control config, with either an “allow” or a “deny” key (mutually exclusive), and an optional “mode” key (“enforce”, the default, or “log”). No acl block means fully open access.

async close() → None[source]: Close module.

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

Execute a local method safely with type conversion

All incoming variables in args and kwargs must be of simple type (i.e. int, float, str, bool, tuple) and will be converted to the requested type automatically. All outgoing variables are converted to simple types automatically as well.

Parameters:

method – Name of method to execute.
*args – Parameters for method.
**kwargs – Parameters for method.

Returns:

Response from method call.

Raises:

KeyError – If method does not exist.

async get_config_caps(**kwargs: Any) → dict[str, tuple[bool, bool, bool]][source]

Returns dict of all config capabilities. First value is whether it has a getter, second is for the setter, third is for a list of possible options..

Returns:: Dict with config caps

async get_config_value(name: str, **kwargs: Any) → bool | int | float | str | list[bool | int | float | str] | dict[str, bool | int | float | str][source]

Returns current value of config item with given name.

Parameters:: name – Name of config item.
Returns:: Current value.
Raises:: InvalidArgumentError – If config item of given name does not exist.

async get_config_value_options(name: str, **kwargs: Any) → list[str][source]

Returns possible values for config item with given name.

Parameters:: name – Name of config item.
Returns:: Possible values.
Raises:: InvalidArgumentError – If config item of given name does not exist.

async get_error_string(**kwargs: Any) → str[source]: Returns description of error, if any.

async get_label(**kwargs: Any) → str[source]: Returns label of module.

async get_permitted_methods(**kwargs: Any) → list[str][source]: Returns names of all methods the calling module is allowed to invoke on this module.

async get_state(**kwargs: Any) → ModuleState[source]: Returns current state of module.

async get_version(**kwargs: Any) → str[source]: Returns pyobs version of module.

property interfaces: list[type[Interface]]: List of implemented interfaces.

async main() → None[source]: Main loop for application.

property methods: dict[str, tuple[Callable[[...], Any], Signature, dict[Any, Any]]]: List of methods.

property name: str: Returns name of module.

async open() → None[source]: Open module.

quit() → None[source]: Quit module.

async reset_error(**kwargs: Any) → bool[source]: Reset error of module, if any. Should be overwritten by derived class to handle error resolution.

async set_config_value(name: str, value: bool | int | float | str | list[bool | int | float | str] | dict[str, bool | int | float | str], **kwargs: Any) → None[source]

Sets value of config item with given name.

Parameters:

name – Name of config item.
value – New value.

Raises:

InvalidArgumentError – If config item of given name does not exist.
ValueError – If value is invalid.

set_error_string(error: str = '') → None[source]: Set error string.

async set_state(state: ModuleState, error_string: str | None = None) → None[source]

Set state of module.

Parameters:

state – New state to set.
error_string – If given, set error string.

async startup() → None[source]

Open the module and mark it ready for RPC dispatch.

Runs the full open() override chain (base Module setup plus every subclass’s own setup) and only then transitions ModuleState.STARTING -> READY, so Module.execute() starts accepting non-whitelisted calls exactly once startup has actually finished. Every caller that opens a module standalone (Application, MultiModule) should call this instead of open() directly – otherwise the module stays in STARTING forever. Callers that need finer-grained control (e.g. tests exercising STARTING behavior directly) can call open() and set_state() separately instead.

Named startup() rather than start() because start() is already IStartStop’s abstract RPC method – a plain start() here would be silently shadowed by any module implementing that interface (guiding, mastermind, scheduler, weather, kiosk, …), which would then never leave STARTING under Application/MultiModule.

class MultiModule(modules: dict[str, Module | dict[str, Any]], shared: dict[str, Any | dict[str, Any]] | None = None, **kwargs: Any)

Bases: Module

Wrapper for running multiple modules in a single process.

Parameters:

modules – Dictionary with modules.
shared – Shared objects between modules.

async close() → None[source]: Cancel sub-module tasks and close shared objects.

async main() → None[source]: Wait until all sub-module tasks have finished.

async open() → None[source]

Open MultiModule.

Shared/non-module child objects are opened normally. Each sub-module is spawned as its own asyncio task so that the module name context var — set at the top of each task — is inherited by all background tasks that module creates, giving correct PYOBS_MODULE attribution in log output.

quit() → None[source]: Quit all sub-modules.

Modules (pyobs.modules)

Writing a minimal module

Startup and ModuleState.STARTING

Interfaces

Communicating between modules

The @timeout decorator

API reference

Startup and `ModuleState.STARTING`

The `@timeout` decorator