Scheduling (pyobs.robotic.scheduler)

This page documents the data model and scheduling logic of the robotic system. For a conceptual overview see Robotic mode (pyobs.robotic). For a worked setup example see Setting up a minimal robotic observation system.

Task and Observation

Task is the fundamental unit of work. It is a pydantic model, so it is fully described by its YAML representation and can be validated, serialised, and round-tripped without any custom code:

name: Observe M51
duration: 120.0
priority: 1.0

target:
  class: pyobs.robotic.scheduler.targets.SiderealTarget
  ra: 202.47
  dec: 47.20

constraints:
  - class: pyobs.robotic.scheduler.constraints.AirmassConstraint
    max_airmass: 2.0

merits:
  - class: pyobs.robotic.scheduler.merits.TransitMerit
    jd0: 2459000.5
    period: 3.14
    duration: 7200

script:
  class: myobs.scripts.ObserveScript
  camera: camera
  telescope: telescope

An Observation is a scheduled instance of a task — it adds a concrete start and end time, a priority score, and an ObservationState. The scheduler produces Observation objects; the mastermind consumes them.

class Task(*, id: Any | None = None, name: str = '', project: str = '', duration: Annotated[float, ~annotated_types.Ge(ge=0.0), ~annotated_types.Le(le=84000.0)] = 0.0, priority: Annotated[float | None, ~annotated_types.Ge(ge=0.0), ~annotated_types.Le(le=9999.0)] = 1.0, constraints: list[Constraint] = <factory>, merits: list[Merit] = <factory>, target: Target | None = None, script: dict[str, ~typing.Any]=<factory>, active: bool = True)

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

Checks whether this task could run now.

Returns:

True, if the task can run now.

property can_start_late: bool

Whether this tasks is allowed to start later than the user-set time, e.g. for flatfields.

Returns:

True, if task can start late.

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

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.

is_finished() → bool

Whether task is finished.

model_config = {'arbitrary_types_allowed': True, 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': 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 resolve_target(time: Time, task: Task, data: DataProvider) → bool

Resolve dynamic target. Returns False if no valid target found.

async run(data: TaskData | None) → None

Run a task

set_resolved_target(target: Target | None) → None

Set the resolved target if not already set, e.g. when restoring from an observation.

property target: Target | None

The resolved target, or the static target if not dynamic.

class Project(*, code: str, name: str = '', priority: Annotated[float | None, Ge(ge=0.0), Le(le=9999.0)] = 1.0)

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 TaskData(task: 'Task', observation_archive: 'ObservationArchive | None' = None, task_archive: 'TaskArchive | None' = None)

class Observation(*, id: Any | None = None, task: Task | Any, start: Annotated[Time, _AstropyTimePydanticTypeAnnotation], end: Annotated[Time, _AstropyTimePydanticTypeAnnotation], state: ObservationState = ObservationState.PENDING, priority: float | None = None, target: Target | None = None)

A scheduled task.

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 fetch_task(task_archive: TaskArchive) → None

Fetch a task from the task archive.

model_config = {'arbitrary_types_allowed': True}

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

model_dump(use_task_id: bool = False, **kwargs: Any) → dict[str, Any]

!!! abstract “Usage Documentation”

[model_dump](../concepts/serialization.md#python-mode)

Generate a dictionary representation of the model, optionally specifying which fields to include or exclude.

Parameters:

• mode – The mode in which to_python should run. If mode is ‘json’, the output will only contain JSON serializable types. If mode is ‘python’, the output may contain non-JSON-serializable Python objects.

• include – A set of fields to include in the output.

• exclude – A set of fields to exclude from the output.

• context – Additional context to pass to the serializer.

• by_alias – Whether to use the field’s alias in the dictionary key if defined.

• exclude_unset – Whether to exclude fields that have not been explicitly set.

• exclude_defaults – Whether to exclude fields that are set to their default value.

• exclude_none – Whether to exclude fields that have a value of None.

• exclude_computed_fields – Whether to exclude computed fields. While this can be useful for round-tripping, it is usually recommended to use the dedicated round_trip parameter instead.

• round_trip – If True, dumped values should be valid as input for non-idempotent types such as Json[T].

• warnings – How to handle serialization errors. False/”none” ignores them, True/”warn” logs errors, “error” raises a [PydanticSerializationError][pydantic_core.PydanticSerializationError].

• fallback – A function to call when an unknown value is encountered. If not provided, a [PydanticSerializationError][pydantic_core.PydanticSerializationError] error is raised.

• serialize_as_any – Whether to serialize fields with duck-typing serialization behavior.

• polymorphic_serialization – Whether to use model and dataclass polymorphic serialization for this call.

Returns:

A dictionary representation of the model.

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 ObservationState(value)

Targets

A Target defines where to point the telescope. It is a PolymorphicBaseModel, so any subclass can appear in the task YAML via the class: key.

The coordinates(time) method accepts a Time and returns a SkyCoord, which lets non-sidereal targets compute their position on the fly:

target:
  class: pyobs.robotic.scheduler.targets.SiderealTarget
  name: M51
  ra: 202.47
  dec: 47.20

class Target(*, name: str)

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.

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 resolve(time: Time, task: Task, data: DataProvider) → Target | None

For dynamic targets. Pick the best available target given current conditions.

class SiderealTarget(*, name: str, ra: float, dec: float)

Bases: Target

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.

Constraints

A Constraint answers a binary question: may this task run at this time? If any constraint returns False, the task is excluded from scheduling entirely for that slot.

Constraints appear in the task YAML under the constraints key (per-task) or under the OnDemandScheduler.constraints key (global, applied to every task).

To write a custom constraint, subclass Constraint and implement __call__ and to_astroplan:

from pyobs.robotic.scheduler.constraints import Constraint

class MyConstraint(Constraint):
    min_elevation: float = 30.0

    def to_astroplan(self):
        return astroplan.AltitudeConstraint(min=self.min_elevation * u.deg)

    async def __call__(self, time, task, data) -> bool:
        if task.target is None:
            return False
        alt = data.observer.altaz(time, task.target).alt.deg
        return alt >= self.min_elevation

class Constraint(*, cost: float = 1.0, target_dependent: bool = False)

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.

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 AirmassConstraint(*, cost: float = 2.0, target_dependent: bool = True, max_airmass: Annotated[float, Ge(ge=1.0), Le(le=9.9)] = 1.3)

Bases: Constraint

Airmass constraint.

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 SolarElevationConstraint(*, cost: float = 1.0, target_dependent: bool = False, min_elevation: Annotated[float, Ge(ge=-90), Le(le=90)] = -90.0, max_elevation: Annotated[float, Ge(ge=-90), Le(le=90)] = -18.0, direction: Literal['rising', 'setting', 'both'] = 'both')

Bases: Constraint

Solar elevation constraint.

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 MoonIlluminationConstraint(*, cost: float = 3.0, target_dependent: bool = False, max_phase: Annotated[float, Ge(ge=0.0), Le(le=1.0)] = 0.0)

Bases: Constraint

Moon illumination constraint.

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 MoonSeparationConstraint(*, cost: float = 3.0, target_dependent: bool = True, min_distance: Annotated[float, Ge(ge=0.0), Le(le=180.0)] = 30.0)

Bases: Constraint

Moon separation constraint.

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 TimeConstraint(*, cost: float = 1.0, target_dependent: bool = False, start: Time, ~astropydantic.time._AstropyTimePydanticTypeAnnotation]=<factory>, end: Time, ~astropydantic.time._AstropyTimePydanticTypeAnnotation]=<factory>)

Bases: Constraint

Time constraint.

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.

Merits

A Merit answers a continuous question: how desirable is it to run this task right now? It returns a float in the range [0, N]. All merit values for a task are multiplied together along with the task’s priority and the project’s priority to produce a single score. The task with the highest score wins each time slot.

A merit returning 0.0 has the same effect as a constraint returning False — the task is excluded from that slot. This lets merits double as soft constraints when needed.

Every merit’s __call__ receives three arguments: time, task, and data. The data argument is a DataProvider which gives access to:

• data.observer — the Observer for the site

• data.last_sunset(time) / data.last_sunrise(time) — cached sunrise/sunset times

• data.night(time) — the calendar date of the observing night

• data.archive — an ObservationArchiveEvolution that provides historical and simulated-future observations

To query past observations from within a merit, use data.archive.get_observations() rather than accessing the archive directly. This ensures the lookahead cache is used during scheduling:

from pyobs.robotic.scheduler.merits import Merit
from pyobs.robotic.observation import ObservationState

class MyMerit(Merit):
    min_days: float = 7.0

    async def __call__(self, time, task, data) -> float:
        from astropy.time import TimeDelta
        import astropy.units as u

        observations = await data.archive.get_observations(
            task=task,
            state=ObservationState.COMPLETED,
            start_after=time - TimeDelta(self.min_days * u.day),
        )
        return 0.0 if len(observations) > 0 else 1.0

Class	Returns 1.0 when…
ConstantMerit	Always. Useful as a baseline or placeholder.
AfterTimeMerit	The current time is after the configured time.
BeforeTimeMerit	The current time is before the configured time.
TimeWindowMerit	The current time falls within one of the configured windows.
TransitMerit	A transit is imminent (based on period and JD of first transit).
IntervalMerit	Enough time has passed since the last observation of this task.
PerNightMerit	The task has not yet reached its maximum observations-per-night count.
FollowMerit	A specified other task has already completed this night.
RandomMerit	Always (but with added Gaussian noise — useful for breaking ties randomly).

class Merit

Bases: PolymorphicBaseModel

Merit class.

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 ConstantMerit(*, merit: Annotated[float, Ge(ge=0.0), Le(le=100.0)] = 1.0)

Bases: Merit

Merit function that returns a constant value.

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 AfterTimeMerit(*, time: Time, ~astropydantic.time._AstropyTimePydanticTypeAnnotation]=<factory>)

Bases: Merit

Merit function that gives 1 after a given time.

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 BeforeTimeMerit(*, time: Time, ~astropydantic.time._AstropyTimePydanticTypeAnnotation]=<factory>)

Bases: Merit

Merit function that gives 1 before a given time.

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 TimeWindowMerit(*, windows: list[TimeWindow], inverse: bool = False)

Bases: Merit

Merit function that uses time windows.

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 TransitMerit(*, jd0: Annotated[float, Ge(ge=2400000), Le(le=2499999)] = 2450000, period: Annotated[float, Ge(ge=0.01), Le(le=9999)] = 1.0, duration: Annotated[int, Ge(ge=1), Le(le=99999)] = 1, ingress: Annotated[float, Ge(ge=0), Le(le=5)] = 0.2, over: Annotated[float, Ge(ge=0), Le(le=5)] = 0.0)

Bases: Merit

Merit function for observing transits.

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 IntervalMerit(*, interval: Annotated[float, Ge(ge=0.0), Le(le=31536000.0)] = 0.0, unit: Literal['s', 'min', 'h', 'hr', 'd', 'wk', 'yr'] = 's')

Bases: Merit

Merit function that enforces an interval between observations.

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 PerNightMerit(*, count: Annotated[int, Ge(ge=0), Le(le=999)] = 0)

Bases: Merit

Merit functions for defining a max number of observations per night.

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 FollowMerit(*, task_id: Any = <factory>)

Bases: Merit

Merit functions that only returns after another given task has run this night.

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 RandomMerit(*, std: Annotated[float, Ge(ge=0.0), Le(le=999.0)] = 1.0)

Bases: Merit

Merit functions for a random normal-distributed number.

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.

Scheduler

pyobs-core ships two TaskScheduler implementations. Both are configured as a nested object inside the Scheduler module:

scheduler:
  class: pyobs.robotic.scheduler.OnDemandScheduler  # or AstroplanScheduler
  twilight: astronomical
  constraints:
    - class: pyobs.robotic.scheduler.constraints.SolarElevationConstraint
      max_solar_elevation: -12.0

The constraints block defines global constraints applied to every task in addition to each task’s own constraints. Note that global constraints are only supported by OnDemandScheduler — AstroplanScheduler applies only per-task constraints.

Class	Strategy
OnDemandScheduler	Greedy, on-demand scheduling. Evaluates constraints and merits at each time step and picks the highest-scoring task. Robust to interruptions — re-runs from the current moment. Supports merits, global constraints, and lookahead to avoid missing higher-priority tasks.
AstroplanScheduler	Full-night planning via astroplan’s PriorityScheduler. Computes a fixed schedule for the entire night in one pass, running the heavy computation in a separate process to avoid blocking the event loop. Only supports SiderealTarget and per-task constraints (not merits). Use when you need a committed nightly plan rather than rolling on-demand decisions.

class TaskScheduler(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)

Bases: Object

Abstract base class for tasks scheduler.

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.

class OnDemandScheduler(twilight: str = 'astronomical', observation_archive: ObservationArchive | dict[str, Any] | None = None, constraints: list[Constraint] | list[dict[str, Any]] | None = None, **kwargs: Any)

Bases: TaskScheduler

Scheduler based on merits.

Initialize a new scheduler.

Parameters:

twilight – astronomical or nautical

async evaluate_constraints(task: Task, start: Time, end: Time, data: DataProvider) → bool

Loops all constraints. If any evaluates to False, return False. Otherwise, return True.

Parameters:

• task – Task to evaluate.

• start – Start time.

• end – End time.

• data – Data provider.

Returns:

True if all constraints evaluate True, False otherwise.

async evaluate_merits(task: Task, start: Time, end: Time, data: DataProvider) → float

Loop all merits, evaluate them and multiply the results. If any evaluates to 0, abort and return 0.

Parameters:

• task – Task to evaluate.

• start – Start time.

• end – End time.

• data – Data provider.

Returns:

The final merit for this task.

class AstroplanScheduler(twilight: str = 'astronomical', **kwargs: Any)

Bases: TaskScheduler

Scheduler based on astroplan.

Initialize a new scheduler.

Parameters:

twilight – astronomical or nautical

Task and Observation archives

TaskArchive and ObservationArchive are abstract base classes defining the interface that the Scheduler and Mastermind modules depend on. pyobs-core ships three concrete implementations of each — see Archive implementations below.

class TaskArchive(on_tasks_changed: Callable[[], Coroutine[Any, Any, None]] | None = None, **kwargs: Any)

Bases: Object

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.

abstractmethod async get_projects() → list[Project]

Returns list of projects.

Returns:

List of projects.

abstractmethod async get_schedulable_tasks() → list[Task]

Returns list of schedulable tasks.

Returns:

List of schedulable tasks

abstractmethod async get_task(id: Any) → Task | None

Returns the task with the given ID.

Returns:

Task with given ID.

abstractmethod async last_changed() → Time | None

Returns time when last time any tasks changed.

class ObservationArchive(**kwargs: Any)

Bases: Object

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.

abstractmethod async add_observations(tasks: ObservationList) → None

Add the list of scheduled tasks to the schedule.

Parameters:

tasks – Scheduled tasks.

abstractmethod async clear_schedule(start_time: Time) → None

Clear schedule after given start time.

Parameters:

start_time – Start time to clear from.

abstractmethod async get_current_observation(task_archive: TaskArchive | None = None) → Observation | None

Returns the currently running observation.

Parameters:

task_archive – Task archive to get task from.

Returns:

Currently running observation.

abstractmethod async get_next_observation(time: Time, task_archive: TaskArchive | None = None) → Observation | None

Returns the active scheduled task at the given time.

Parameters:

• time – Time to return task for.

• task_archive – Task archive to get task from.

Returns:

Scheduled task at the given time.

abstractmethod async get_observations(task: Task | None = None, state: ObservationState | None = None, start_before: Time | None = None, start_after: Time | None = None, end_before: Time | None = None, end_after: Time | None = None) → ObservationList

Returns a list of observations matching the given filters.

Parameters:

• task – If given, only return observations for this task.

• state – If given, only return observations in this state.

• start_before – If given, only return observations that start before this time.

• start_after – If given, only return observations that start after this time.

• end_before – If given, only return observations that end before this time.

• end_after – If given, only return observations that end after this time.

Returns:

List of matching observations.

abstractmethod async get_schedule() → ObservationList

Fetch schedule from portal.

Returns:

Dictionary with tasks.

Raises:

• Timeout – If request timed out.

• ValueError – If something goes wrong.

abstractmethod async update_observation(observation: Observation) → None

Updates observation. :Parameters: observation – Observation to update.

Archive implementations

Filesystem (pyobs.robotic.filesystem)

Tasks are YAML files in a directory; observations are YAML files named by night. No external services required — the simplest setup for a single telescope.

Backend (pyobs.robotic.backend)

Tasks and observations are managed by the pyobs-robotic-backend HTTP service. Enables multi-telescope coordination, a web UI for queue management, and centralised logging.

class BackendTaskArchive(url: str, token: str, auto_update: bool = True, **kwargs: Any)

Bases: TaskArchive

Task archive based on pyobs-robotic-backend.

Creates a new task archive.

Parameters:

• url – URL of pyobs-robotic-backend.

• token – Auth token.

async close() → None

Closes the backend observation archive.

async get_projects() → list[Project]

Returns list of projects.

Returns:

List of projects.

async get_schedulable_tasks() → list[Task]

Returns list of schedulable tasks.

Returns:

List of schedulable tasks

async get_task(id: Any) → Task | None

Returns the task with the given ID.

Returns:

Task with given ID.

async last_changed() → Time | None

Returns time when last time any tasks changed.

async last_update_time() → Time

Fetches last schedule update time.

async open() → None

Opens the backend task archive.

class BackendObservationArchive(url: str, token: str, mode: Literal['day', 'night'] = 'night', auto_update: bool = True, **kwargs: Any)

Bases: ObservationArchive

Observation archive based on pyobs-robotic-backend.

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.

async add_observations(tasks: ObservationList) → None

Add the list of scheduled tasks to the schedule.

Parameters:

tasks – Scheduled tasks.

async clear_schedule(start_time: Time) → None

Clear schedule after given start time.

Parameters:

start_time – Start time to clear from.

async close() → None

Closes the backend observation archive.

async get_current_observation(task_archive: TaskArchive | None = None) → Observation | None

Returns the currently running observation.

Parameters:

task_archive – Task archive to get task from.

Returns:

Currently running observation.

async get_next_observation(time: Time, task_archive: TaskArchive | None = None) → Observation | None

Returns the active scheduled task at the given time.

Parameters:

• time – Time to return task for.

• task_archive – Task archive to get task from.

Returns:

Scheduled task at the given time.

async get_observations(task: Task | None = None, state: ObservationState | None = None, start_before: Time | None = None, start_after: Time | None = None, end_before: Time | None = None, end_after: Time | None = None) → ObservationList

Returns a list of observations matching the given filters.

Parameters:

• task – If given, only return observations for this task.

• state – If given, only return observations in this state.

• start_before – If given, only return observations that start before this time.

• start_after – If given, only return observations that start after this time.

• end_before – If given, only return observations that end before this time.

• end_after – If given, only return observations that end after this time.

Returns:

List of matching observations.

async get_schedule() → ObservationList

Fetch schedule from portal.

Returns:

Dictionary with tasks.

Raises:

• Timeout – If request timed out.

• ValueError – If something goes wrong.

async last_update_time() → Time

Fetches last schedule update time.

async open() → None

Opens the backend observation archive.

async update_observation(observation: Observation) → None

Updates observation. :Parameters: observation – Observation to update.

Las Cumbres Observatory (pyobs.robotic.lco)

Integration with the Las Cumbres Observatory observation portal. Tasks are fetched from the LCO portal API using an instrument type and authorisation token; observations are read from and written back to the LCO schedule. Also includes LcoTaskRunner, which maps LCO request configurations to the appropriate Script subclass based on a configurable scripts map.

class LcoTaskArchive(url: str, token: str, instrument_type: str | list[str], **kwargs: Any)

Bases: TaskArchive

Scheduler for using the LCO portal

Creates a new LCO scheduler.

Parameters:

• url – URL to portal

• token – Authorization token for portal

• instrument_type – Type of instrument to use.

• scripts – External scripts

async get_projects() → list[Project]

Returns list of projects from the LCO portal.

async get_schedulable_tasks() → list[Task]

Returns a list of schedulable tasks.

Returns:

List of schedulable tasks

async get_task(id: Any) → Task | None

Returns the task with the given ID.

async last_changed() → Time | None

Returns time when last time any blocks changed.

class LcoObservationArchive(url: str, configdb: str, site: str, token: str, enclosure: str, telescope: str, period: int = 24, mode: Literal['read', 'write', 'readwrite'] = 'readwrite', **kwargs: Any)

Bases: ObservationArchive

Scheduler for using the LCO portal

Creates a new LCO scheduler.

Parameters:

• url – URL to portal

• configdb – URL to configdb

• site – Site filter for fetching requests

• token – Authorization token for portal

• enclosure – Enclosure for new schedules.

• telescope – Telescope for new schedules.

• instrument – Instrument for new schedules.

• period – Period to schedule in hours

async add_observations(tasks: ObservationList) → None

Add the list of scheduled tasks to the schedule.

Parameters:

tasks – Scheduled tasks.

async clear_schedule(start_time: Time) → None

Clear schedule after given start time.

Parameters:

start_time – Start time to clear from.

async get_current_observation(task_archive: TaskArchive | None = None) → Observation | None

Returns the currently running observation.

async get_next_observation(time: Time, task_archive: TaskArchive | None = None) → Observation | None

Returns the active scheduled task at the given time.

Parameters:

time – Time to return an observation for.

Returns:

Scheduled task at the given time.

async get_observations(task: Task | None = None, state: ObservationState | None = None, start_before: Time | None = None, start_after: Time | None = None, end_before: Time | None = None, end_after: Time | None = None) → ObservationList

Returns a list of observations matching the given filters.

The LCO portal requires a request id, so a task is mandatory for this archive.

Parameters:

• task – Task to get observations for (required for the LCO archive).

• state – If given, only return observations in this state.

• start_before – If given, only return observations that start before this time.

• start_after – If given, only return observations that start after this time.

• end_before – If given, only return observations that end before this time.

• end_after – If given, only return observations that end after this time.

Returns:

List of matching observations.

async get_schedule() → ObservationList

Fetch schedule from the portal.

Returns:

Dictionary with tasks.

Raises:

• Timeout – If request timed out.

• ValueError – If something goes wrong.

async observations_for_night(date: date) → ObservationList

Returns a list of observations for the given task.

Parameters:

date – Date of night to get observations for.

Returns:

List of observations for the given task.

async send_update(status_id: int | None, status: dict[str, Any]) → None

Send a report to the LCO portal

Parameters:

• status_id – id of config status

• status – Status dictionary

async update_observation(observation: Observation) → None

Updates observation state in the portal.

class LcoTaskRunner(scripts: dict[str, Any], **kwargs: Any)

Bases: TaskRunner

Creates a new LCO task runner.

Parameters:

scripts – External scripts

async can_run(task: Task) → bool

Checks whether this task could run now.

Parameters:

task – Task to run

Returns:

True, if the task can run now.

async run_task(task: Task) → bool

Run a task.

Parameters:

task – Task to run

Returns:

Success or not

Image archives

Archive is the base class used by ArchiveSkyflatPriorities to query historical observations when calculating flat-field priorities. Concrete implementations are configured via the class: key like any other polymorphic model.

class Archive

Bases: PolymorphicBaseModel

Base class for image archives.

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 PyobsArchive(*, url: str, token: str, proxies: dict[str, str] | None = None)

Bases: Archive

Connector class to running pyobs-archive instance.

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(_PyobsArchive__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 LocalArchive(*, root: str)

Bases: Archive

Connector class to a local image 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(_LocalArchive__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.

Scheduling internals

These classes are used internally by the scheduler and are relevant mainly when writing custom merit functions.

DataProvider is passed to every Merit and Constraint during a scheduling run. It provides cached access to site geometry (sunrise, sunset, night boundaries) and to the observation history via its archive attribute.

ObservationArchiveEvolution wraps the real ObservationArchive with two additions:

• Caching — observations for each task are fetched from the archive once per scheduling run and cached in memory, avoiding repeated HTTP requests during evaluation of many time slots.

• Lookahead simulation — as the scheduler plans ahead and tentatively assigns tasks to future slots, it calls evolve() to record those assignments. Subsequent merit evaluations for the same task then see those simulated observations, so IntervalMerit and PerNightMerit correctly prevent the same task from being scheduled twice in one run.

class DataProvider(observer: Observer, archive: ObservationArchiveEvolution | None = None)

Bases: object

Data provider for Merit classes.

last_sunrise(time: Time) → Time

Returns the time of the last sunrise.

last_sunset(time: Time) → Time

Returns the time of the last sunset.

night(time: Time) → date

Returns the time of the last sunset.

class ObservationArchiveEvolution(observer: Observer, obs_archive: ObservationArchive | None = None)

Bases: object

async observations_for_night(date: datetime.date) → ObservationList

Returns list of observations for the given task.

Parameters:

date – Date of night to get observations for.

Returns:

List of observations for the given task.