Scripts (pyobs.robotic.scripts)

A Script is the leaf of the robotic system — it contains the actual observing logic. Scripts are pydantic models (not Module subclasses), so they have no async lifecycle of their own. Instead, they receive runtime context (comm, vfs, observer) injected at instantiation time, and they are created fresh for each task execution.

Writing a script

Subclass Script and implement two async methods:

import logging
from typing import TYPE_CHECKING
from pyobs.interfaces import ICamera, IPointingRaDec
from pyobs.robotic.scripts import Script

if TYPE_CHECKING:
    from pyobs.robotic.task import TaskData

log = logging.getLogger(__name__)


class ObserveScript(Script):
    camera: str = "camera"
    telescope: str = "telescope"
    exposure_time: float = 30.0
    num_exposures: int = 1

    async def can_run(self, data: TaskData | None) -> bool:
        try:
            await self.comm.proxy(self.camera, ICamera)
            await self.comm.proxy(self.telescope, IPointingRaDec)
        except ValueError:
            return False
        return True

    async def run(self, data: TaskData | None) -> None:
        if data is None or data.task.target is None:
            raise ValueError("No target.")

        camera = await self.comm.proxy(self.camera, ICamera)
        telescope = await self.comm.proxy(self.telescope, IPointingRaDec)

        from pyobs.utils.time import Time
        target = data.task.target.coordinates(Time.now())

        log.info("Moving telescope to %s...", data.task.target.name)
        await telescope.move_radec(target.ra.deg, target.dec.deg)

        for i in range(self.num_exposures):
            log.info("Taking exposure %d/%d...", i + 1, self.num_exposures)
            await camera.set_exposure_time(self.exposure_time)
            await camera.grab_data(broadcast=True)

``can_run(data)`` is called by the scheduler before each scheduling cycle. Return False if required hardware is offline or conditions are not met. The scheduler will exclude tasks whose script returns False from the current slot.

``run(data)`` is called by the mastermind when the task’s scheduled time arrives. The TaskData argument gives access to the current task, the ObservationArchive, and the TaskArchive. Raise InterruptedError to signal that the script was aborted cleanly.

The script is configured in the task YAML under the script key:

script:
  class: myobs.scripts.ObserveScript
  camera: camera
  telescope: telescope
  exposure_time: 60.0
  num_exposures: 3

Runtime context

Scripts have access to the same runtime properties as Object via PrivateAttrMixin:

  • self.comm — Comm for calling other modules

  • self.vfs — VirtualFileSystem for file I/O

  • self.observer — Observer with the observatory location

  • self.location — EarthLocation

  • self.timezone — tzinfo

These are injected automatically when the script is created via pyobs_model_validate(). They are never set during __init__ or pydantic validation — they are only available when the script is instantiated at runtime from within an Object context.

TaskData

TaskData is passed to both can_run and run. It is a simple dataclass that bundles references to the relevant parts of the robotic system:

@dataclass
class TaskData:
    task: Task
    observation_archive: ObservationArchive | None = None
    task_archive: TaskArchive | None = None

Most scripts only need data.task (for the target and duration). Scripts that need to record results or look up task history can use data.observation_archive.

Script base class

class Script(*, exptime_done: float = 0.0)[source]

Bases: PolymorphicBaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

async can_run(data: TaskData | None) → bool[source]

Checks whether this script could run now.

Returns:

True, if the script can run now.

estimate_duration() → float[source]

Estimate duration of this script in seconds.

get_fits_headers(namespaces: list[str] | None = None) → dict[str, Any][source]

Returns FITS header for the current status of this module.

Parameters:

namespaces – If given, only return FITS headers for the given namespaces.

Returns:

Dictionary containing FITS headers.

model_config = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

  • self – The BaseModel instance.

  • context – The context.

async run(data: TaskData | None) → None[source]

Run script.

Raises:

InterruptedError – If interrupted

Built-in scripts

Observing

class AutoFocus(*, exptime_done: float = 0.0, autofocus: str = 'autofocus', telescope: str = 'telescope', count: int = 5, step: float = 0.1, exposure_time: float = 2.0)[source]

Bases: Script

Script for running autofocus series.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

async can_run(data: TaskData | None) → bool[source]

Whether this config can currently run. :returns: True if script can run now.

model_config = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

  • self – The BaseModel instance.

  • context – The context.

async run(data: TaskData | None) → None[source]

Run script. :raises InterruptedError: If interrupted

class DarkBias(*, exptime_done: float = 0.0, camera: str, count: int = 20, exptime: float = 0, binning: tuple[int, int] = (1, 1))[source]

Bases: Script

Script for running darks or biases.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

async can_run(data: TaskData | None) → bool[source]

Whether this config can currently run. :returns: True if script can run now.

model_config = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

  • self – The BaseModel instance.

  • context – The context.

async run(data: TaskData | None) → None[source]

Run script. :raises InterruptedError: If interrupted

class SkyFlats(*, exptime_done: float = 0.0, roof: str, telescope: str, flatfield: str, functions: dict[str, Any], priorities: dict[str, Any], min_exptime: float = 0.5, max_exptime: float = 5, timespan: float = 7200, filter_change: float = 30, count: int = 20, readout: dict[str, Any] | None = None)[source]

Bases: Script

Script for scheduling and running skyflats using an IFlatField module.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

async can_run(data: TaskData | None) → bool[source]

Whether this config can currently run.

Returns:

True if script can run now.

model_config = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

  • self – The BaseModel instance.

  • context – The context.

async run(data: TaskData | None) → None[source]

Run script.

Raises:

InterruptedError – If interrupted

Control flow

These scripts do not perform observations themselves — they compose other scripts into more complex execution patterns. They can be nested arbitrarily.

class SequentialRunner(*, exptime_done: float = 0.0, scripts: list[dict[str, Any]], check_all_can_run: bool = True)[source]

Bases: Script

Script for running a sequence of other scripts.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

async can_run(data: TaskData | None) → bool[source]

Checks whether this script could run now.

Returns:

True, if the script can run now.

model_config = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

  • self – The BaseModel instance.

  • context – The context.

async run(data: TaskData | None) → None[source]

Run script.

Raises:

InterruptedError – If interrupted

Run a list of scripts one after the other. By default, checks that all scripts can run before starting. Set ``check_all_can_run: false`` to only check the first.

class ParallelRunner(*, exptime_done: float = 0.0, scripts: list[dict[str, Any]], check_all_can_run: bool = True)[source]

Bases: Script

Script for running other scripts in parallel.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

async can_run(data: TaskData | None) → bool[source]

Checks whether this script could run now.

Returns:

True, if the script can run now.

model_config = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

  • self – The BaseModel instance.

  • context – The context.

async run(data: TaskData | None) → None[source]

Run script.

Raises:

InterruptedError – If interrupted

Run a list of scripts concurrently using asyncio.gather. Useful for simultaneously operating two independent hardware systems.

class ConditionalRunner(*, exptime_done: float = 0.0, condition: str, true: dict[str, Any], false: dict[str, Any] | None = None)[source]

Bases: Script

Script for running an if condition.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

async can_run(data: TaskData | None) → bool[source]

Checks whether this script could run now.

Returns:

True, if the script can run now.

get_fits_headers(namespaces: list[str] | None = None) → dict[str, Any][source]

Returns FITS header for the current status of this module.

Parameters:

namespaces – If given, only return FITS headers for the given namespaces.

Returns:

Dictionary containing FITS headers.

model_config = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

  • self – The BaseModel instance.

  • context – The context.

async run(data: TaskData | None) → None[source]

Run script.

Raises:

InterruptedError – If interrupted

Evaluate a Python expression and run either a ``true`` or ``false`` sub-script. The expression context provides ``now`` as a UTC datetime.

class CasesRunner(*, exptime_done: float = 0.0, expression: str, cases: dict[str | int | float, Any])[source]

Bases: Script

Script for distinguishing cases.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

async can_run(data: TaskData | None) → bool[source]

Checks whether this script could run now.

Returns:

True, if the script can run now.

get_fits_headers(namespaces: list[str] | None = None) → dict[str, Any][source]

Returns FITS header for the current status of this module.

Parameters:

namespaces – If given, only return FITS headers for the given namespaces.

Returns:

Dictionary containing FITS headers.

model_config = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

  • self – The BaseModel instance.

  • context – The context.

async run(data: TaskData | None) → None[source]

Run script.

Raises:

InterruptedError – If interrupted

Evaluate an expression and select a sub-script from a dict of cases. Supports an ``else`` key for a default.

class SelectorScript(*, exptime_done: float = 0.0, mode: str, selector: str)[source]

Bases: Script

Script for running Mode Selection.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

async can_run(data: TaskData | None) → bool[source]

Whether this config can currently run. :returns: True if script can run now.

model_config = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

  • self – The BaseModel instance.

  • context – The context.

async run(data: TaskData | None) → None[source]

Run script. :raises InterruptedError: If interrupted

Switch a module implementing IMode to a specified mode.

class CallModule(*, exptime_done: float = 0.0, module: str, method: str, params: list[Any] = <factory>)[source]

Bases: Script

Script for calling method on a module.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

async can_run(data: TaskData | None) → bool[source]

Checks whether this script could run now.

Returns:

True, if the script can run now.

model_config = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

  • self – The BaseModel instance.

  • context – The context.

async run(data: TaskData | None) → None[source]

Run script.

Raises:

InterruptedError – If interrupted

Call an arbitrary method on any module by name. Useful for one-off actions without writing a full script class.

class LogRunner(*, exptime_done: float = 0.0, expression: str)[source]

Bases: Script

Script for logging something.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

async can_run(data: TaskData | None) → bool[source]

Checks whether this script could run now.

Returns:

True, if the script can run now.

model_config = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

  • self – The BaseModel instance.

  • context – The context.

async run(data: TaskData | None) → None[source]

Run script.

Raises:

InterruptedError – If interrupted

Evaluate a Python expression and log the result. Useful for debugging.

TargetPicker

TargetPicker is a helper used by some scripts (e.g. AutoFocus) to select a suitable target from a CSV catalogue at runtime. It filters the catalogue by current altitude and picks a target in the observable window:

target:
  class: pyobs.robotic.utils.TargetPicker
  csv: /robotic/stars/hipparcos_8mag.csv
  name_col: HIP
  ra_col: RAICRS
  dec_col: DEICRS
  min_alt: 30
  max_alt: 75

Sky flat utilities

These classes support the SkyFlats script and are configured as nested objects within it.

class FlatFielder(functions: str | Dict[str, str | Dict[str, str]], target_count: float = 30000, min_exptime: float = 0.5, max_exptime: float = 5, test_frame: Tuple[float, float, float, float] | None = None, counts_frame: Tuple[float, float, float, float] | None = None, allowed_offset_frac: float = 0.2, min_counts: int = 100, pointing: Dict[str, Any] | SkyFlatsBasePointing | None = None, callback: Callable[[...], Coroutine[Any, Any, None]] | None = None, **kwargs: Any)

Bases: Object

Automatized flat-fielding.

Initialize a new flat fielder.

Parameters:

  • functions – Function f(h) for each filter to describe ideal exposure time as a function of solar elevation h, i.e. something like exp(-0.9*(h+3.9)). See ExpTimeEval for details.

  • target_count – Count rate to aim for.

  • min_exptime – Minimum exposure time.

  • max_exptime – Maximum exposure time.

  • test_frame – Tupel (left, top, width, height) in percent that describe the frame for on-sky testing.

  • counts_frame – Tupel (left, top, width, height) in percent that describe the frame for calculating mean count rate.

  • allowed_offset_frac – Offset from target_count (given in fraction of it) that’s still allowed for good flat-field

  • min_counts – Minimum counts in frames.

  • observer – Observer to use.

  • vfs – VFS to use.

  • callback – Callback function for statistics.

class State(value)[source]

Bases: Enum

class Twilight(value)[source]

Bases: Enum

async abort() → None[source]

Abort current actions.

property has_filters: bool

Returns True, if functions are based on filters.

async reset() → None[source]

Reset flat fielder

class SkyFlatsBasePointing

Bases: PolymorphicBaseModel

Base class for flat pointings.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

  • self – The BaseModel instance.

  • context – The context.

async reset() → None[source]

Reset pointing.

class SkyFlatsStaticPointing

Bases: SkyFlatsBasePointing

Static flat pointing.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

  • self – The BaseModel instance.

  • context – The context.

async reset() → None[source]

Reset pointing.

class SkyflatPriorities

Bases: PolymorphicBaseModel

Base class for sky flat priorities.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

  • self – The BaseModel instance.

  • context – The context.

class ConstSkyflatPriorities(*, priorities: dict[tuple[str, tuple[int, int]], float])

Bases: SkyflatPriorities

Constant flat priorities.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

  • self – The BaseModel instance.

  • context – The context.

class ArchiveSkyflatPriorities(*, archive: Archive, site: str, instrument: str, filter_names: list[str], binnings: list[int])

Bases: SkyflatPriorities

Calculate flat priorities from an archive.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

  • self – The BaseModel instance.

  • context – The context.