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, **kwargs: Any)
Bases:
ModuleWatch 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 inopen()can see a file while its write is still in progress. The same wait time is also used to delay re-processing of files whose destination copy failed (see_worker).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 spaces out re-queued files after failed destination copies.
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_fileregardless of watch mode – inotify events, polling, and the initial scan inopen()all discover candidates first and letadd_filedecide, 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 ``watchpathunder 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.
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 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:
ModuleWrites 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.
Pipeline
- class Pipeline(pipeline: list[dict[str, Any] | ImageProcessor], sources: str | list[str] | None = None, interval: int | None = None, **kwargs: Any)
Bases:
Module,PipelineMixinRuns 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.
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:
ModuleMeasures 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.