Mixins (pyobs.mixins)

Mixins are classes that can be inherited from to automatically add some functionality to a module.

CameraSettingsMixin

class CameraSettingsMixin(filters: str | IFilters | None = None, filter_name: str | None = None, binning: int | None = None, **kwargs: Any)

Mixin for a device that should be able to set camera settings.

Initializes the mixin.

Parameters:

• filters – Filter wheel module.

• filter – Filter to set.

• binning – Binning to set.

async _do_camera_settings(camera: Module | IData | IFilters | IBinning | IWindow) → None

Do camera settings for given camera.

FitsHeaderMixin

class FitsHeaderMixin(fits_namespaces: list[str] | None = None, fits_headers: dict[str, Any] | None = None, filenames: str = '/cache/pyobs-{DAY-OBS|date:}-{FRAMENUM|string:04d}.fits', frame_number: bool = True, night_obs: bool = True, fits_header_timeout: float = 15.0, **kwargs: Any)

Helper methods for all modules that implement IImageGrabber.

Initialise the mixin.

Parameters:

• fits_namespaces – List of namespaces for FITS headers that this camera should request.

• fits_headers – Additional FITS headers.

• filename – Filename pattern for FITS images.

• frame_number – Whether to add frame number to FITS file header.

• night_obs – If True, DAY-OBS will contain the night of observation, not the calendar day.

• fits_header_timeout – Maximum seconds to wait for a peer’s FITS headers before skipping them. A peer that never answers (e.g. a laptop put to sleep without closing its client) would otherwise stall the frame for the full XMPP IQ timeout.

_fitsheadermixin_add_fits_headers(image: Image | PrimaryHDU) → None

Add FITS header keywords to the given FITS header.

Parameters:

image – Image with header to add to.

async _fitsheadermixin_add_framenum(image: Image | PrimaryHDU) → None

Add FRAMENUM keyword to header

Parameters:

image – Image with header to add to.

async add_fits_headers(image: Image | PrimaryHDU) → None

Add requested FITS headers to header of given image.

Parameters:

image – Image with header to add to.

add_local_fits_headers(image: Image | PrimaryHDU) → None

Add the cheap, local FITS headers to the given image (no I/O, no comm).

This is the always-on half of add_fits_headers(): the configured static fits_headers dict plus the computed fields from _fitsheadermixin_add_fits_headers(). The frame-sequence-number step is deliberately kept out (it does a VFS read+write per call), so video-rate callers can reuse this without hitting the VFS on every frame.

Parameters:

image – Image with header to add to.

async add_requested_fits_headers(image: Image | PrimaryHDU, futures: dict[str, Task[Any]]) → None

Add requested FITS headers to header of given image.

Parameters:

• image – Image with header to add to.

• futures – Futures to get headers from.

format_filename(image: Image | PrimaryHDU) → str | None

Format filename according to given pattern and store in header of image.

Parameters:

image – Image with header to add to.

async request_fits_headers(before: bool = True) → dict[str, Task[Any]]

Request FITS headers from other modules.

Returns:

Futures from all modules.

FitsNamespaceMixin

class FitsNamespaceMixin(fits_namespaces: dict[str, list[str]] | None = None, **kwargs: Any)

Mixin for IFitsHeaderProvider modules that filters FITS headers by namespace.

__add_namespace(name: str, keywords: list[str], hdr: dict[str, Any]) → None

Add FITS header keywords from namespace to list of valid keywords

Parameters:

• name – Name of namespace

• keywords – List of valid keywords, which will be added to

• hdr – Full unfiltered header

_filter_fits_namespace(hdr: dict[str, FitsHeaderEntry], sender: str, namespaces: list[str] | None = None, **kwargs: Any) → dict[str, FitsHeaderEntry]

Filter FITS header keywords by given namespaces. If no namespaces are given, let all through. Always let keywords with this module’s name as namespace pass.

Parameters:

• hdr – Input header to filter

• namespaces – Requested namespaces

• sender – Name of module that requested headers

Returns:

Filtered FITS header

FollowMixin

class FollowMixin(device: str | None, mode: type[IPointingAltAz | IPointingRaDec], interval: float = 10, tolerance: float = 1, only_follow_when_ready: bool = True, follow_max_age: float | None = None, *args: Any, **kwargs: Any)

Mixin for a device that should follow the motion of another.

Initializes the mixin.

Parameters:

• device – Name of device to follow

• interval – Interval in seconds between position checks.

• tolerance – Tolerance in degrees between both devices to trigger new movement.

• mode – Set to “altaz” to follow Alt/Az coordinates or “radec” to follow RA/Dec.

• only_follow_when_ready – Only follow if is_ready() is True.

• follow_max_age – Treat the followed device’s cached position as unavailable once it’s older than this many seconds – guards against silently following a frozen position if the followed device’s own position-publish loop dies while its connection stays up. Defaults to 3x interval (loose enough to tolerate a slow tick, tight enough to notice a dead feed within a few position checks).

async __update_follow() → None

Update function.

property is_following: bool

Returns True, if we’re following another device.

ImageFitsHeaderMixin

class ImageFitsHeaderMixin(centre: tuple[float, float] | None = None, rotation: float | None = None, **kwargs: Any)

Helper methods for all modules that need FITS headers for an image.

Initialise the mixin.

Parameters:

• fits_namespaces – List of namespaces for FITS headers that this camera should request.

• fits_headers – Additional FITS headers.

• centre – (x, y) tuple of camera centre.

• rotation – Rotation east of north.

• filename – Filename pattern for FITS images.

_fitsheadermixin_add_fits_headers(image: Image | PrimaryHDU) → None

Add FITS header keywords to the given FITS header.

Parameters:

image – Image with header to add to.

MotionStatusMixin

class MotionStatusMixin(motion_status_interfaces: list[str] | None = None, **kwargs: Any)

Mixin for IMotion devices for handling status.

Initializes the mixin.

Parameters:

interfaces – List of interfaces to handle or None

async _change_motion_status(status: MotionStatus, interface: str | None = None) → None

Change motion status and send event,

Parameters:

• status – New motion status

• interface – Interface to set motion status for

_combine_motion_status() → MotionStatus

Method for combining motion statuses for individual interfaces into the global one. Can be overriden.

_is_ready() → bool

Whether the device is ready, for publishing as IReady state. Override for device-specific criteria.

motion_status(device: str | None = None) → MotionStatus

Returns current motion status (synchronous, for internal use).

Parameters:

device – Name of device to get status for, or None.

Returns:

Current motion status.

PipelineMixin

class PipelineMixin(steps: list[dict[str, Any] | ImageProcessor] | None = None, archive: dict[str, Any] | Archive | None = None, **kwargs: Any)

Mixin for a module that needs to implement an image pipeline.

Initializes the mixin.

Parameters:

• steps – Pipeline steps to run on images.

• archive – Default archive config/object for steps that accept one (e.g. Calibration) and don’t already specify their own – lets a pipeline’s steps share the archive it was itself given, instead of repeating the same archive config in every step that needs one. Only injected into a step’s config if that step’s class actually declares an archive parameter (checked via signature inspection); steps that don’t declare one never receive it, rather than receiving-then-silently-dropping it.

static _accepts_archive(class_name: str) → bool

Whether the given class declares an archive parameter anywhere in its __init__ MRO.

async reset_pipeline() → None

Resets all previous state of the involved image processors.

async run_pipeline(image: Image) → Image

Run the pipeline on the given image.

Each step is run, and an ImageError it raises is handled according to the step’s on_error setting (default “raise”): re-raised, dispatched to the step’s handle_error(), logged, or ignored. Non-ImageError exceptions always propagate.

Parameters:

image – Image to run pipeline on.

Returns:

Image after pipeline run.

SpectrumFitsHeaderMixin

class SpectrumFitsHeaderMixin(fits_namespaces: list[str] | None = None, fits_headers: dict[str, Any] | None = None, filenames: str = '/cache/pyobs-{DAY-OBS|date:}-{FRAMENUM|string:04d}.fits', frame_number: bool = True, night_obs: bool = True, fits_header_timeout: float = 15.0, **kwargs: Any)

Helper methods for all modules that need FITS headers for an image.

Initialise the mixin.

Parameters:

• fits_namespaces – List of namespaces for FITS headers that this camera should request.

• fits_headers – Additional FITS headers.

• filename – Filename pattern for FITS images.

• frame_number – Whether to add frame number to FITS file header.

• night_obs – If True, DAY-OBS will contain the night of observation, not the calendar day.

• fits_header_timeout – Maximum seconds to wait for a peer’s FITS headers before skipping them. A peer that never answers (e.g. a laptop put to sleep without closing its client) would otherwise stall the frame for the full XMPP IQ timeout.

WaitForMotionMixin

class WaitForMotionMixin(wait_for_modules: list[str] | None = None, wait_for_states: list[MotionStatus | str] | None = None, wait_for_timeout: float = 0, **kwargs: Any)

Mixin for a device that should wait for the motion status of another device.

Initializes the mixin.

Parameters:

• wait_for_modules – One or more modules to wait for.

• wait_for_states – List of states to wait for.

• wait_for_timeout – Wait timeout in seconds.

async _wait_for_motion(abort: Event) → None

Wait until all devices are in one of the given motion states.

Parameters:

abort – Abort event.

Raises:

TimeoutError – If wait timed out.

WeatherAwareMixin

class WeatherAwareMixin(weather: str | IWeather | None = None, weather_max_age: float = 120.0, **kwargs: Any)

Mixin for IMotion devices that should park(), when weather gets bad.

async __on_bad_weather(event: Event, sender: str) → bool

Abort exposure if a bad weather event occurs.

Parameters:

• event – The bad weather event.

• sender – Who sent it.

async __weather_check() → None

Thread for continuously checking for good weather

async _on_good_weather(event: Event, sender: str) → bool

Change status of weather.

Parameters:

• event – The good weather event.

• sender – Who sent it.

async open() → None

Open mixin.