Robotic mode (pyobs.modules.robotic)

Modules for robotic mode. TODO: write doc

Mastermind

class Mastermind(schedule: TaskSchedule | dict[str, Any], runner: TaskRunner | dict[str, Any], allowed_late_start: int = 300, allowed_overrun: int = 300, after_task_sleep: int = 0, **kwargs: Any)

Bases: Module, IAutonomous, IFitsHeaderBefore

Mastermind for a full robotic mode.

Initialize a new auto focus system.

Parameters:

• schedule – Object that can return schedule.

• allowed_late_start – Allowed seconds to start late.

• allowed_overrun – Allowed time for a task to exceed it’s window in seconds

async get_fits_header_before(namespaces: list[str] | None = None, **kwargs: Any) → 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.

async is_running(**kwargs: Any) → bool

Whether a service is running.

async open() → None

Open module.

async start(**kwargs: Any) → None

Starts a service.

async stop(**kwargs: Any) → None

Stops a service.

PointingSeries

class PointingSeries(grid: list[Grid | GridFilter | dict[str, Any]], finish: int = 90, acquisition: str | None = None, telescope: str = 'telescope', **kwargs: Any)

Bases: Module, IAutonomous

Module for running pointing series.

Initialize a new pointing series.

Parameters:

• grid – Grid to use for pointing series.

• finish – When this number in percent of points have been finished, terminate mastermind.

• acquisition – IAcquisition unit to use.

• telescope – ITelescope unit to use.

async is_running(**kwargs: Any) → bool

Whether a service is running.

async start(**kwargs: Any) → None

Starts a service.

async stop(**kwargs: Any) → None

Stops a service.

Scheduler

class Scheduler(scheduler: dict[str, Any] | TaskScheduler, tasks: Dict[str, Any] | TaskArchive, schedule: Dict[str, Any] | TaskSchedule, trigger_on_task_started: bool = False, trigger_on_task_finished: bool = False, schedule_range: float = 24.0, safety_time: float = 300, **kwargs: Any)

Bases: Module, IStartStop, IRunnable

Scheduler.

Initialize a new scheduler.

Parameters:

• scheduler – Scheduler to use.

• tasks – Task archive to use.

• schedule – Task schedule to use.

• trigger_on_task_started – Whether to trigger a re-calculation of schedule, when task has started.

• trigger_on_task_finishes – Whether to trigger a re-calculation of schedule, when task has finished.

• schedule_range – Number of hours to schedule into the future

• safety_time – If no ETA for next task to start exists (from current task, weather became good, etc), use this time in seconds to make sure that we don’t schedule for a time when the scheduler is still running

async abort(**kwargs: Any) → None

Abort current actions.

async is_running(**kwargs: Any) → bool

Whether scheduler is running.

async open() → None

Open module.

async run(**kwargs: Any) → None

Trigger a re-schedule.

async start(**kwargs: Any) → None

Start scheduler.

async stop(**kwargs: Any) → None

Stop scheduler.

ScriptRunner

class ScriptRunner(script: Dict[str, Any], run_once: bool = False, timeout: int = 10, **kwargs: Any)

Bases: Module, IRunnable

Module for running a script.

Initialize a new script runner.

Parameters:

script – Config for script to run.

async abort(**kwargs: Any) → None

Abort current actions.

async run(**kwargs: Any) → None

Run script.