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)[source]

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

cant_run_reason() → str | None[source]

Returns reason why task cannot run, or None if it can.

get_fits_headers(namespaces: list[str] | None = None) → dict[str, tuple[Any, str]][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.

is_finished() → bool[source]

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.

reset_resolved_target() → None[source]

Reset resolved target.

async resolve_target(time: Time, task: Task, data: DataProvider) → bool[source]

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

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

Run a task

set_resolved_target(target: Target | None) → None[source]

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)[source]

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)[source]
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)[source]

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

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][source]
!!! 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)[source]

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

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

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

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

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)[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 filter_skycoord(time: Time, coords: SkyCoord, data: DataProvider) → np.ndarray[source]

Returns a boolean mask of candidates passing this constraint.

Default implementation passes all candidates. Override in target-dependent subclasses to vectorise the constraint evaluation across a SkyCoord array.

Parameters:

  • time – Time to evaluate constraint at.

  • coords – Array of candidate coordinates.

  • data – Data provider.

Returns:

Boolean numpy array, True for candidates that pass the constraint.

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)[source]

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.

async filter_skycoord(time: Time, coords: SkyCoord, data: DataProvider) → np.ndarray[source]

Returns a boolean mask of candidates passing this constraint.

Default implementation passes all candidates. Override in target-dependent subclasses to vectorise the constraint evaluation across a SkyCoord array.

Parameters:

  • time – Time to evaluate constraint at.

  • coords – Array of candidate coordinates.

  • data – Data provider.

Returns:

Boolean numpy array, True for candidates that pass the constraint.

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')[source]

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)[source]

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)[source]

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.

async filter_skycoord(time: Time, coords: SkyCoord, data: DataProvider) → np.ndarray[source]

Returns a boolean mask of candidates passing this constraint.

Default implementation passes all candidates. Override in target-dependent subclasses to vectorise the constraint evaluation across a SkyCoord array.

Parameters:

  • time – Time to evaluate constraint at.

  • coords – Array of candidate coordinates.

  • data – Data provider.

Returns:

Boolean numpy array, True for candidates that pass the constraint.

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>)[source]

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

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)[source]

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>)[source]

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>)[source]

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] = <factory>, inverse: bool = False)[source]

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, min_window: Annotated[float, Ge(ge=0), Le(le=60)] = 5.0)[source]

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.

end_time() → Time[source]

Returns the time until which observations should run: mid-transit + duration/2 + ingress buffer.

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.

transit_time() → Time[source]

Returns the time of the next mid-transit.

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')[source]

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)[source]

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>)[source]

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)[source]

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)[source]

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

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

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)[source]

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.

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.

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.

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

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

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)[source]

Bases: object

Data provider for Merit classes.

last_sunrise(time: Time) → Time[source]

Returns the time of the last sunrise.

last_sunset(time: Time) → Time[source]

Returns the time of the last sunset.

night(time: Time) → date[source]

Returns the time of the last sunset.

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

Bases: object

async observations_for_night(date: datetime.date) → ObservationList[source]

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.