Objects (pyobs.object)

Object is the base for almost all classes in pyobs. It adds some convenience methods and helper methods for creating other Objects.

There are a few convenience functions:

• create_object() creates objects from dictionaries.

• get_object() is a wrapper around pyobs.object.create_object() that can do further checks.

• get_safe_object() is a wrapper around get_object() that never raises exceptions.

The Object class is the base for almost everything in pyobs. Understanding it is the key to understanding how the framework is configured and how objects are created from YAML files.

The class: key and YAML instantiation

Every pyobs object can be described in a YAML configuration block by specifying its fully qualified class name under the class: key, followed by any constructor parameters:

class: pyobs.modules.camera.DummyCamera
exposure_time: 2.0
vfs:
  class: pyobs.vfs.VirtualFileSystem
  roots:
    cache:
      class: pyobs.vfs.LocalFile
      root: /data

When pyobs encounters such a block, it strips the class: key and passes the remaining keys as keyword arguments to the constructor. This means constructor parameters map directly to YAML keys, and nested objects (like vfs above) are instantiated recursively in the same way.

The functions create_object() and get_object() implement this mechanism. You will typically use get_object() or add_child_object() inside your own classes rather than calling create_object() directly.

Lifecycle: __init__, open(), and close()

Every Object follows a strict two-phase lifecycle:

1. Construction (__init__): Store parameters, create child objects via add_child_object(), and register background tasks via add_background_task(). Do not connect to hardware or external services here.

2. Opening (open()): Connect to hardware, subscribe to events, and start background tasks. This is where side effects should happen.

Always pair open() with a corresponding close(), which stops background tasks and closes all child objects.

Important

add_background_task() and add_child_object() must be called in __init__, before open() is called.

Runtime context: comm, vfs, observer, location, timezone

Object provides several shared runtime resources as properties:

• comm — the Comm object for communicating with other modules.

• vfs — the VirtualFileSystem for file access.

• observer — an astroplan.Observer built from the configured location and timezone.

• location — the observatory’s EarthLocation.

• timezone — the local timezone as a datetime.tzinfo.

These are automatically propagated to child objects created via add_child_object(). They can be configured in the YAML block (see Overview for examples) or inherited from a parent object.

Background tasks

Background tasks are async coroutines that run concurrently alongside the main module. Register them in __init__ using add_background_task():

class MyObject(Object):
    def __init__(self, interval: int = 10, **kwargs):
        Object.__init__(self, **kwargs)
        self._interval = interval
        self.add_background_task(self._run)

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

By default, background tasks are restarted automatically if they raise an unhandled exception. Pass restart=False to disable this. Pass autostart=False to prevent the task from starting when open() is called (you can then start it manually).

Child objects

add_child_object() creates a child object from a config dict or existing instance, automatically copies the runtime context (comm, vfs, observer, etc.) into it, and registers it for automatic open()/close() calls:

class MyObject(Object):
    def __init__(self, camera: dict | ICamera, **kwargs):
        Object.__init__(self, **kwargs)
        self._camera = self.add_child_object(camera, ICamera)

When MyObject.open() is called, it will also call self._camera.open() automatically. The same applies to close().

API reference

class Object(vfs: VirtualFileSystem | dict[str, Any] | None = None, comm: Comm | dict[str, Any] | None = None, timezone: str | datetime.tzinfo | None = 'utc', location: str | dict[str, Any] | EarthLocation | None = None, observer: Observer | None = None, **kwargs: Any)

Base class for all objects in pyobs.

Note

Objects must always be opened and closed using open() and close(), respectively.

This class provides a VirtualFileSystem, a timezone and a location. From the latter two, an observer object is automatically created.

Object also adds support for easily adding threads using the add_background_task() method as well as a watchdog thread that automatically restarts threads, if requested.

Using add_child_object(), other objects can be (created an) attached to this object, which then automatically handles calls to open() and close() on those objects.

Parameters:

• vfs – VFS to use (either object or config)

• comm – Comm object to use

• timezone – Timezone at observatory.

• location – Location of observatory, either a name or a dict containing latitude, longitude, and elevation.

add_background_task(func: Callable[[...], Coroutine[Any, Any, None]], restart: bool = True, autostart: bool = True) → BackgroundTask

Add a new function that should be run in the background.

MUST be called in constructor of derived class or at least before calling open() on the object.

Parameters:

• func – Func to add.

• restart – Whether to restart this function.

• autostart – Whether to start this function when the module is opened

Returns:

Background task

add_child_object(config_or_object: dict[str, Any] | ObjectClass | type[ObjectClass] | Any, object_class: type[ObjectClass], copy_comm: bool = True, **kwargs: Any) → ObjectClass

add_child_object(config_or_object: ObjectClass, **kwargs: Any) → ObjectClass

add_child_object(config_or_object: dict[str, Any] | ObjectClass | type[ObjectClass] | Any, object_class: Literal[None], copy_comm: bool = True, **kwargs: Any) → Any

Create a new sub-module, which will automatically be opened and closed.

Parameters:

• config_or_object – Module definition

• object_class – Class for new module

• copy_comm – Copy comm from this object to the new one.

Returns:

The created module.

async close() → None

Close module.

static config_or_object_get_param(config_or_object: dict[str, Any] | Any, param: str) → Any

Checks, whether a config_or_object has the given parameter.

Parameters:

• config_or_object – Dict config or object.

• param – Parameter name to check.

Returns:

get_object(config_or_object: dict[str, Any] | ObjectClass | type[ObjectClass], object_class: type[ObjectClass], copy_comm: bool = True, **kwargs: Any) → ObjectClass

get_object(config_or_object: dict[str, Any], object_class: Literal[None], copy_comm: bool = True, **kwargs: Any) → Any

get_object(config_or_object: ObjectClass | type[ObjectClass], object_class: Literal[None], copy_comm: bool = True, **kwargs: Any) → ObjectClass

get_object(config_or_object: dict[str, Any] | ObjectClass | type[ObjectClass], object_class: type[ObjectClass] | None = None, copy_comm: bool = True, **kwargs: Any) → ObjectClass | Any

Creates object from config or returns object directly, both optionally after check of type.

Parameters:

• config_or_object – A configuration dict or an object itself to create/check. If a dict with a class key is given, a new object is created.

• object_class – Class to check object against.

• copy_comm – Copy comm from this object to the new one.

Returns:

(New) object (created from config) that optionally passed class check.

Raises:

TypeError – If the object does not match the given class.

get_safe_object(config_or_object: dict[str, Any] | ObjectClass | type[ObjectClass] | Any, object_class: Type[ObjectClass], copy_comm: bool = True, **kwargs: Any) → ObjectClass | None

get_safe_object(config_or_object: dict[str, Any] | ObjectClass | type[ObjectClass] | Any, object_class: None, copy_comm: bool = True, **kwargs: Any) → Any | None

Calls get_object in a safe way and returns None, if an exceptions thrown.

async open() → None

Open module.

property opened: bool

Whether object has been opened.

async proxy(name_or_object: str | object, obj_type: type[ProxyType]) → ProxyType

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

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.

quit() → None

Can be overloaded to quit program.

get_object(config_or_object: dict[str, Any] | ObjectClass | type[ObjectClass], object_class: type[ObjectClass], **kwargs: Any) → ObjectClass

get_object(config_or_object: dict[str, Any], object_class: Literal[None], **kwargs: Any) → Any

get_object(config_or_object: ObjectClass | type[ObjectClass], object_class: Literal[None], **kwargs: Any) → ObjectClass

get_object(config_or_object: dict[str, Any] | ObjectClass | type[ObjectClass], object_class: type[ObjectClass] | None = None, **kwargs: Any) → ObjectClass | Any

Creates object from config or returns object directly, both optionally after check of type.

Parameters:

• config_or_object – A configuration dict or an object itself to create/check. If a dict with a class key is given, a new object is created.

• object_class – Class to check object against.

Returns:

(New) object (created from config) that optionally passed class check.

Raises:

TypeError – If the object does not match the given class.

get_safe_object(config_or_object: ObjectClass | dict[str, Any], object_class: type[ObjectClass], **kwargs: Any) → ObjectClass

get_safe_object(config_or_object: ObjectClass | Any, object_class: None, **kwargs: Any) → Any | None

Calls get_object in a safe way and returns None, if an exceptions thrown.

Parameters:

• config_or_object – A configuration dict or an object itself to create/check. If a dict with a class key is given, a new object is created.

• object_class – Class to check object against.

Returns:

(New) object (created from config) that optionally passed class check or None.

create_object(config: dict[str, Any], *args: Any, **kwargs: Any) → Any

Create object from dict config.

Parameters:

• config – Config to create object from

• *args – Parameters to be passed to object.

• **kwargs – Parameters to be passed to object.

Returns:

Created object.

get_class_from_string(class_name: str) → Any

Get class from a given string.

Parameters:

class_name – Name of class as string.

Returns:

Actual class.