Image operations (pyobs.modules.image)

Modules for image operations. TODO: write doc

ImageWatcher

class ImageWatcher(watchpath: str, destinations: list[str] | None = None, poll: bool = False, poll_interval: int = 5, wait_time: int = 10, pattern: str = '*', flatten: bool = True, *, error_after: int = 1800, backoff_cap: int = 3600, **kwargs: Any)

Bases: Module

Watch for new files and write them to all given destinations.

Watches a path for new files and stores them in all given destinations. Only if all operations were successful, the file is deleted.

New files are not processed immediately, but only after a wait time (wait_time) has passed since they were detected. This is mainly to make sure the file is fully written before it is read: while the inotify watcher only reports files after they have been closed for writing (CLOSE_WRITE), the polling watcher and the initial scan in open() can see a file while its write is still in progress.

A file that fails to process is retried with exponentially growing backoff, up to backoff_cap, unless the failure is unrecoverable (see _handle_failure), in which case it’s left in place and never retried. See the error_after/backoff_cap constructor args for details.

Create a new image watcher.

Parameters:

  • watchpath – Path to watch.

  • destinations – Filename patterns for destinations.

  • poll – If True, watchpath is polled instead of watched by inotify.

  • poll_interval – Interval for polling in seconds, if poll is True.

  • wait_time – Time in seconds between adding a file to the list and processing it. Gives a file that is still being written time to finish (relevant for poll mode and the initial scan) and is also the base interval for retrying a file after a transient failure (see error_after).

  • pattern – Only watch/process files matching this fnmatch pattern, checked against the full path (so e.g. “*.fits” also matches files found in subdirectories). This is the single point of filtering, applied uniformly in add_file regardless of watch mode – inotify events, polling, and the initial scan in open() all discover candidates first and let add_file decide, none of them pre-filter by pattern before that.

  • flatten – For a non-templated destination (one without {placeholder}``s), whether to collapse the file's path to just its basename under that destination (the historical behavior, default) or to preserve its path relative to ``watchpath under the destination instead. Files found via recursive watching or polling only keep their subdirectory structure at the destination when this is False. Note this relies on the destination VFS root creating missing parent directories on write (the default for e.g. LocalFile) – a destination configured to not do that would repeatedly fail and re-queue any file whose relative path needs a parent directory that doesn’t exist yet.

  • error_after – Seconds a file may keep failing with a transient error (see below) before a single ERROR is logged for it. Defaults to 1800 (30 minutes), a middle ground for the deployments seen so far; how long a destination can legitimately be unreachable (a remote archive, an NFS mount, …) varies per deployment, so override it per config where an outage should page someone sooner or later. The file keeps being retried (with growing backoff, capped at backoff_cap) indefinitely either way; this only controls when the one-time alert fires, not whether retrying continues.

  • backoff_cap – Maximum seconds between retries of a file that keeps failing transiently. The interval starts at wait_time and doubles on each failure up to this cap.

Note

A failure is either transient (retried forever, with the backoff/error_after behavior above) or terminal (FileNotFoundError, PermissionError, IsADirectoryError, wherever in the pipeline they’re raised) – a terminal failure is logged as an ERROR immediately and the file is never re-queued; it’s simply left where it is in watchpath; neither watch mode re-detects an existing, unchanged file, so it stays visibly stuck rather than being retried forever or silently vanishing.

Note

If a directory holding a file that’s already queued (added but not yet processed by _worker) is renamed before that happens, the worker’s read of the file at its captured (pre-rename) path fails and the file is dropped rather than picked up at its new location – this is unrelated to, and not covered by, the recursive-watching support for directories renamed after they no longer hold queued files. Avoid renaming directories that may still hold queued/retrying files.

async add_file(filename: str) → None[source]

Add a file to the file queue.

Parameters:

filename (str) – Local filename of new file.

async cleanup_extra(filename: str) → None[source]

Can be overwritten by derived classes to do clean up after successful copying. All information are stored in self.current_file and can be checked against the given filename.

Note

This hook runs on the module’s event loop and must not block: no CPU-heavy or synchronous I/O here. Offload such work to a thread, e.g. with asyncio.to_thread.

Parameters:

filename – Input name of original file.

async close() → None[source]

Close image watcher.

async open() → None[source]

Open image watcher.

async process_extra(filename: str) → bool[source]

Can be overwritten by derived classes to do extra processing on files. All information are stored in self.current_file and can be checked against the given filename.

Note

This hook runs on the module’s event loop and must not block: no CPU-heavy or synchronous I/O here. Offload such work to a thread, e.g. with asyncio.to_thread.

Parameters:

filename – Input name of original file.

Returns:

Whether processing was successful

ImageWriter

class ImageWriter(filename: str = '/archive/{FNAME}', sources: str | list[str] | None = None, **kwargs: Any)

Bases: Module

Writes new images to disk.

Creates a new image writer.

Parameters:

  • filename – Pattern for filename to store images at.

  • sources – List of sources (e.g. cameras) to process images from or None for all.

async open() → None[source]

Open image writer.

async process_new_image_event(event: Event, sender: str) → bool[source]

Puts a new images in the DB with the given ID.

Parameters:

  • event – New image event

  • sender – Who sent the event?

Returns:

Success

Pipeline

class Pipeline(pipeline: list[dict[str, Any] | ImageProcessor], sources: str | list[str] | None = None, interval: int | None = None, **kwargs: Any)

Bases: Module, PipelineMixin

Runs an image pipeline.

Creates a new HTTP publisher for images.

Parameters:

  • pipeline – Pipeline to run on images.

  • sources – List of sources to process images from.

  • interval – Interval in seconds for automatic run of pipeline.

async open() → None[source]

Open module.

async process_new_image_event(event: Event, sender: str) → bool[source]

Runs a new image through the pipeline.

Parameters:

  • event – New image event

  • sender – Who sent the event?

Returns:

Success

Seeing

class Seeing(sources: str | list[str] | None = None, publisher: Publisher | dict[str, Any] | None = None, max_ellipticity: float = 0.2, correct_for_airmass: bool = True, **kwargs: Any)

Bases: Module

Measures seeing on reduced images with a catalog.

Creates a new seeing estimator.

Parameters:

  • sources – List of sources (e.g. cameras) to process images from or None for all.

  • publisher – Publisher to publish results to.

  • max_ellipticity – Maximum ellipticity for sources to consider.

  • correct_for_zenith – Whether to correct seeing for airmass.

async open() → None[source]

Open module.

async process_new_image_event(event: Event, sender: str) → bool[source]

Puts a new images in the DB with the given ID.

Parameters:

  • event – New image event

  • sender – Who sent the event?

Returns:

Success