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.

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

For runtime target selection instead of a fixed coordinate, use DynamicTarget together with a Picker. resolve() is called once per scheduling attempt and delegates to the configured picker to choose a concrete target; coordinates() then delegates to whatever was picked:

target:
  class: pyobs.robotic.scheduler.targets.DynamicTarget
  picker:
    class: pyobs.robotic.scheduler.targets.picker.CsvPicker
    csv: /robotic/stars/hipparcos_8mag.csv
    name_col: HIP
    ra_col: RAICRS
    dec_col: DEICRS

CsvPicker picks a random target from a CSV catalogue that survives the task’s own target-dependent constraints (e.g. an altitude constraint, evaluated at the time being scheduled) — useful for filler targets drawn from a large catalogue rather than one fixed coordinate per task.

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

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

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.

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 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, time: Time | None = None) → Observation | None

Returns the currently running observation.

Parameters:

• task_archive – Task archive to get task from.

• time – Time to check for. Defaults to now.

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(time: Time | None = None) → ObservationList

Fetch schedule.

Parameters:

time – Time to fetch schedule for. Defaults to now.

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.

TaskRunner is the abstract base for the object that actually executes a task’s Script once the mastermind hands it over.

Archive implementations

Filesystem (pyobs.robotic.storage.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. FileSystemTaskArchive/ FileSystemObservationArchive are abstract intermediate bases shared by any filesystem-backed archive; YamlTaskArchive/YamlObservationArchive are the concrete YAML implementation.

Portal (pyobs.robotic.storage.portal)

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

Las Cumbres Observatory (pyobs.robotic.storage.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.

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.

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.