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.

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:

camera = await self.proxy("camera", ICamera)
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)
    telescope = await self.proxy("telescope", ITelescope)
    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(name: str | None = None, label: str | None = None, own_comm: bool = True, additional_config_variables: list[str] | None = None, **kwargs: Any)

Bases: Object, IModule, IConfig

Base class for all pyobs modules.

Parameters:

• name – Name of module. If None, ID from comm object is used.

• 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.

async close() → None

Close module.

async execute(method: str, *args: Any, **kwargs: Any) → Any

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]]

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) → Any

Returns current value of config item with given name.

Parameters:

name – Name of config item.

Returns:

Current value.

Raises:

ValueError – If config item of given name does not exist.

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

Returns possible values for config item with given name.

Parameters:

name – Name of config item.

Returns:

Possible values.

Raises:

ValueError – If config item of given name does not exist.

async get_error_string(**kwargs: Any) → str

Returns description of error, if any.

async get_label(**kwargs: Any) → str

Returns label of module.

async get_state(**kwargs: Any) → ModuleState

Returns current state of module.

async get_version(**kwargs: Any) → str

Returns pyobs version of module.

property interfaces: list[type[Interface]]

List of implemented interfaces.

async main() → None

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

Open module.

quit() → None

Quit module.

async reset_error(**kwargs: Any) → bool

Reset error of module, if any. Should be overwritten by derived class to handle error resolution.

async set_config_value(name: str, value: Any, **kwargs: Any) → None

Sets value of config item with given name.

Parameters:

• name – Name of config item.

• value – New value.

Raises:

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

set_error_string(error: str = '') → None

Set error string.

async set_state(state: ModuleState, error_string: str | None = None) → None

Set state of module.

Parameters:

• state – New state to set.

• error_string – If given, set error string.

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.