Scripts (pyobs.robotic.scripts)

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

Writing a script

Subclass Script and implement two async methods:

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

if TYPE_CHECKING:
    from pyobs.robotic.task import TaskData

log = logging.getLogger(__name__)


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

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

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

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

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

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

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

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

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

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

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

Runtime context

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

self.comm — Comm for calling other modules
self.vfs — VirtualFileSystem for file I/O
self.observer — Observer with the observatory location
self.location — EarthLocation
self.timezone — tzinfo

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

TaskData

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

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

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

Script base class

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

Bases: PolymorphicBaseModel

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

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

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

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

Checks whether this script could run now.

Returns:: True, if the script can run now.

cant_run_reason() → str | None[source]: Returns reason why script cannot run, or None if it can.

estimate_duration(data: TaskData | None = None, time: Time | None = None) → float[source]: Estimate duration of this script in seconds.

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

Returns FITS header for the current status of this module.

Parameters:: namespaces – If given, only return FITS headers for the given namespaces.
Returns:: Dictionary containing FITS headers.

model_config = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

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

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

Parameters:

self – The BaseModel instance.
context – The context.

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

Run script.

Raises:: InterruptedError – If interrupted

Built-in scripts

Observing

class ImagingScript(*, exptime_done: float = 0.0, configuration: Configuration = <factory>, camera: str, telescope: str | None = None, filters: str | None = None, autoguider: str | None = None, acquisition: str | None = None)[source]

Bases: Script

Default script for imaging configs.

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

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

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

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

Whether this config can currently run.

Returns:: True, if the script can run now

estimate_duration(data: TaskData | None = None, time: Time | None = None) → float[source]: Estimate the duration of this script in seconds.

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

Returns FITS header for the current status of this module.

Parameters:: namespaces – If given, only return FITS headers for the given namespaces.
Returns:: Dictionary containing FITS headers.

model_config = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

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

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

Parameters:

self – The BaseModel instance.
context – The context.

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

Run script.

Raises:: InterruptedError – If interrupted

The default script for science exposures: moves to the target, optionally acquires and guides on it, then works through one or more instrument_configs (binning, window, exposure time, filter, image type), each repeated count times, for repeats full passes.

class TransitImagingScript(*, exptime_done: float = 0.0, configuration: Configuration = <factory>, camera: str, telescope: str | None = None, filters: str | None = None, autoguider: str | None = None, acquisition: str | None = None)[source]

Bases: ImagingScript

Imaging script that runs until the end of a transit window.

Requires a TransitMerit on the task. Overrides _run_configurations() to loop instrument configurations until transit_time + duration/2 + ingress.

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

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

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

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

Whether this script can currently run.

In addition to ImagingScript checks, requires a TransitMerit on the task.

Returns:: True if the script can run now.

estimate_duration(data: TaskData | None = None, time: Time | None = None) → float[source]

Estimate duration of the transit observation.

Parameters:

data – Task data containing the TransitMerit.
time – If given, return remaining time until end of the next transit window that starts at or after time. If None, return the full observable window (ingress + duration + ingress).

Returns:

Estimated duration in seconds.

model_config = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

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

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

Parameters:

self – The BaseModel instance.
context – The context.

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

Run script.

Raises:

InterruptedError – If interrupted.
ValueError – If no TransitMerit found.

An ImagingScript subclass that repeats its instrument configurations for as long as a transit window is open, instead of a fixed number of times. Requires a TransitMerit on the task.

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

Bases: Script

Script for running autofocus series.

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

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

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

async can_run(data: TaskData | None) → bool[source]: Whether this config can currently run. :returns: True if script can run now.

estimate_duration(data: TaskData | None = None, time: Time | None = None) → float[source]: Estimate duration of the autofocus run.

model_config = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

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

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

Parameters:

self – The BaseModel instance.
context – The context.

async run(data: TaskData | None) → None[source]: Run script. :raises InterruptedError: If interrupted

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

Bases: Script

Script for running darks or biases.

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

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

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

async can_run(data: TaskData | None) → bool[source]: Whether this config can currently run. :returns: True if script can run now.

estimate_duration(data: TaskData | None = None, time: Time | None = None) → float[source]: Estimate duration of the dark/bias series.

model_config = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

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

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

Parameters:

self – The BaseModel instance.
context – The context.

async run(data: TaskData | None) → None[source]: Run script. :raises InterruptedError: If interrupted

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

Bases: Script

Script for scheduling and running skyflats using an IFlatField module.

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

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

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

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

Whether this config can currently run.

Returns:: True if script can run now.

estimate_duration(data: TaskData | None = None, time: Time | None = None) → float[source]

Estimate duration of the sky flats.

The actual schedule depends on sky conditions that can only be evaluated at runtime, so this returns the configured timespan as a conservative upper bound.

model_config = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

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

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

Parameters:

self – The BaseModel instance.
context – The context.

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

Run script.

Raises:: InterruptedError – If interrupted

class PointingScript(*, exptime_done: float = 0.0, telescope: str, pointing: SkyFlatsBasePointing)[source]

Bases: Script

Script for pointing the telescope for flats.

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

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

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

async can_run(data: TaskData | None) → bool[source]: Whether this config can currently run. :returns: True if script can run now.

estimate_duration(data: TaskData | None = None, time: Time | None = None) → float[source]: Estimate duration of slewing to the flat-field pointing.

model_config = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

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

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

Parameters:

self – The BaseModel instance.
context – The context.

async run(data: TaskData | None) → None[source]: Run script. :raises InterruptedError: If interrupted

Points the telescope at the sky position configured for flat-fielding (via a SkyFlatsBasePointing\ ), without taking any exposures itself — typically run just before a SkyFlatsScript.

Control flow

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

class SequentialRunner(*, exptime_done: float = 0.0, scripts: list[Script], check_all_can_run: bool = True)[source]

Bases: Script

Script for running a sequence of other scripts.

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

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

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

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

Checks whether this script could run now.

Returns:: True, if the script can run now.

estimate_duration(data: TaskData | None = None, time: Time | None = None) → float[source]: Estimate duration as the sum of the durations of all sub-scripts.

model_config = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

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

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

Parameters:

self – The BaseModel instance.
context – The context.

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

Run script.

Raises:: InterruptedError – If interrupted

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

class ParallelRunner(*, exptime_done: float = 0.0, scripts: list[Script], check_all_can_run: bool = True)[source]

Bases: Script

Script for running other scripts in parallel.

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

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

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

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

Checks whether this script could run now.

Returns:: True, if the script can run now.

estimate_duration(data: TaskData | None = None, time: Time | None = None) → float[source]: Estimate duration as the longest duration of all sub-scripts, since they run in parallel.

model_config = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

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

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

Parameters:

self – The BaseModel instance.
context – The context.

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

Run script.

Raises:: InterruptedError – If interrupted

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

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

Bases: Script

Script for running an if condition.

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

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

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

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

Checks whether this script could run now.

Returns:: True, if the script can run now.

estimate_duration(data: TaskData | None = None, time: Time | None = None) → float[source]: Estimate duration of the branch that would be run for the current condition.

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

Returns FITS header for the current status of this module.

Parameters:: namespaces – If given, only return FITS headers for the given namespaces.
Returns:: Dictionary containing FITS headers.

model_config = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

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

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

Parameters:

self – The BaseModel instance.
context – The context.

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

Run script.

Raises:: InterruptedError – If interrupted

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

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

Bases: Script

Script for distinguishing cases.

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

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

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

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

Checks whether this script could run now.

Returns:: True, if the script can run now.

estimate_duration(data: TaskData | None = None, time: Time | None = None) → float[source]: Estimate duration of the script for the current case.

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

Returns FITS header for the current status of this module.

Parameters:: namespaces – If given, only return FITS headers for the given namespaces.
Returns:: Dictionary containing FITS headers.

model_config = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

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

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

Parameters:

self – The BaseModel instance.
context – The context.

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

Run script.

Raises:: InterruptedError – If interrupted

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

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

Bases: Script

Script for running Mode Selection.

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

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

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

async can_run(data: TaskData | None) → bool[source]: Whether this config can currently run. :returns: True if script can run now.

estimate_duration(data: TaskData | None = None, time: Time | None = None) → float[source]: Estimate duration of the mode change.

model_config = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

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

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

Parameters:

self – The BaseModel instance.
context – The context.

async run(data: TaskData | None) → None[source]: Run script. :raises InterruptedError: If interrupted

Switch a module implementing IMode to a specified mode.

class CallModuleScript(*, exptime_done: float = 0.0, module: str, interface: str, method: str, params: dict[str, str | int | float]=<factory>)[source]

Bases: Script

Script for calling a method on a module.

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

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

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

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

Checks whether this script could run now.

Returns:: True, if the script can run now.

model_config = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

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

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

Parameters:

self – The BaseModel instance.
context – The context.

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

Run script.

Raises:: InterruptedError – If interrupted

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

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

Bases: Script

Script for logging something.

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

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

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

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

Checks whether this script could run now.

Returns:: True, if the script can run now.

model_config = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

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

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

Parameters:

self – The BaseModel instance.
context – The context.

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

Run script.

Raises:: InterruptedError – If interrupted

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

class DebugTriggerScript(*, exptime_done: float = 0.0, triggered: bool = False)[source]

Bases: Script

Script for a debug trigger.

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

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

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

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

Checks whether this script could run now.

Returns:: True, if the script can run now.

model_config = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

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

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

Parameters:

self – The BaseModel instance.
context – The context.

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

Run script.

Raises:: InterruptedError – If interrupted

Sets its own triggered flag to True when run and does nothing else. Useful as a minimal, dependency-free script for testing scheduling and task-runner behavior without any real hardware.

Sky flat utilities

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

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

Bases: Object

Automatized flat-fielding.

Initialize a new flat fielder.

Parameters:

functions – Function f(h) for each filter to describe ideal exposure time as a function of solar elevation h, i.e. something like exp(-0.9*(h+3.9)). See ExpTimeEval for details.
target_count – Count rate to aim for.
min_exptime – Minimum exposure time.
max_exptime – Maximum exposure time.
test_frame – Tupel (left, top, width, height) in percent that describe the frame for on-sky testing.
counts_frame – Tupel (left, top, width, height) in percent that describe the frame for calculating mean count rate.
allowed_offset_frac – Offset from target_count (given in fraction of it) that’s still allowed for good flat-field
min_counts – Minimum counts in frames.
observer – Observer to use.
vfs – VFS to use.
callback – Callback function for statistics.

class State(value)[source]: Bases: Enum

class Twilight(value)[source]: Bases: Enum

async abort() → None[source]: Abort current actions.

property has_filters: bool: Returns True, if functions are based on filters.

async reset() → None[source]: Reset flat fielder

class SkyFlatsBasePointing

Bases: PolymorphicBaseModel

Base class for flat pointings.

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

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

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

model_config = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

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

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

Parameters:

self – The BaseModel instance.
context – The context.

async reset() → None[source]: Reset pointing.

class SkyFlatsStaticPointing

Bases: SkyFlatsBasePointing

Static flat pointing.

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

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

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

model_config = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

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

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

Parameters:

self – The BaseModel instance.
context – The context.

async reset() → None[source]: Reset pointing.

class SkyflatPriorities

Bases: PolymorphicBaseModel

Base class for sky flat priorities.

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

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

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

model_config = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

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

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

Parameters:

self – The BaseModel instance.
context – The context.

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

Bases: SkyflatPriorities

Constant flat priorities.

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

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

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

model_config = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

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

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

Parameters:

self – The BaseModel instance.
context – The context.

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

Bases: SkyflatPriorities

Calculate flat priorities from an archive.

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

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

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

model_config = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

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

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

Parameters:

self – The BaseModel instance.
context – The context.