mirar.processors.utils package

Module for general utility processors such as I/O and interacting with metadata

Submodules

mirar.processors.utils.cal_hunter module

Module for finding archival calibration images if these are missing

class mirar.processors.utils.cal_hunter.CalHunter(requirements: CalRequirement | list[CalRequirement], *args, **kwargs)

Bases: ImageLoader

Processor to find any missing calibration images, by searching previous nights of data

base_key = 'calhunt'

description()

Return a description of the processor

Returns:

A description of the processor

max_n_cpu: int = 1

class mirar.processors.utils.cal_hunter.CalRequirement(target_name, required_field: str, required_values: str | list[str])

Bases: object

Class to specify particular calibration files that must be present for processing

check_images(images: ImageBatch)

Check a batch of images, to see whether the calibration requirement is met. Adds any required images to the cache, then updates the check of the self.success

Parameters:

images – ImageBatch

Returns:

None

required_field: str

required_values: str | list[str]

target_name: str

exception mirar.processors.utils.cal_hunter.MissingCalibrationsError

Bases: ImageNotFoundError

Error to raise if calibration files are missing

mirar.processors.utils.cal_hunter.find_required_cals(latest_dir: str | ~pathlib.Path, night: str, requirements: list[~mirar.processors.utils.cal_hunter.CalRequirement], open_f: ~collections.abc.Callable[[str], ~mirar.data.image_data.Image] = <function open_raw_image>, images: ~mirar.data.image_data.ImageBatch = <mirar.data.image_data.ImageBatch object>, skip_latest_night: bool = False) → ImageBatch

Broad function to search for missing calibration files in previous nights

Parameters:

• latest_dir – The directory for the raw images

• night – The night being processed

• requirements – List of calibration requirements

• open_f – Function to open raw images

• images – Current image list (default: empty)

• skip_latest_night – Boolean to skip the directory of night being processed

Returns:

Updated image batch

mirar.processors.utils.cal_hunter.update_requirements(requirements: list[CalRequirement], images: ImageBatch) → list[CalRequirement]

Iteratively check a list of Cal Requirements against an image batch

Parameters:

• requirements – CalRequirements to check

• images – Images to check

Returns:

Updated CalRequirements

mirar.processors.utils.error_annotator module

Module for adding saved errors into image header metadata

class mirar.processors.utils.error_annotator.ErrorStackAnnotator(errorstack: ErrorStack, processed_images: list[str])

Bases: BaseImageProcessor

Processor to update image headers with processing failurs

base_key = 'errorannotate'

description() → str

Return a description of the processor

Returns:

A description of the processor

unpack_errorstack() → dict

Convert an errorstack to an image-indexed dictionary

Returns:

dictionary of errors

mirar.processors.utils.header_annotate module

Module for adding metadata to Image headers

class mirar.processors.utils.header_annotate.HeaderAnnotator(input_keys: str | list[str], output_key: str)

Bases: BaseImageProcessor

Processor for adding metadata to Image headers

base_key = 'header_annotator'

description() → str

Return a description of the processor

Returns:

A description of the processor

class mirar.processors.utils.header_annotate.HeaderEditor(edit_keys: str | list[str], values: str | float | int | list)

Bases: BaseImageProcessor

Processor for modifying metadata in Image headers

base_key = 'header_editor'

description() → str

Return a description of the processor

Returns:

A description of the processor

mirar.processors.utils.header_reader module

Module for reading/logging values from image headers

class mirar.processors.utils.header_reader.HeaderReader(*keys)

Bases: BaseImageProcessor

Processor to extract data from image headers, and print it to the log

base_key = 'header_reader'

description() → str

Return a description of the processor

Returns:

A description of the processor

mirar.processors.utils.image_loader module

Module for loading images

exception mirar.processors.utils.image_loader.BadImageError

Bases: ProcessorError

Exception for bad images

class mirar.processors.utils.image_loader.ImageListLoader(img_list: list[Path], load_image: Callable[[str], Image | list[Image]] = None)

Bases: BaseImageProcessor

Processor to load raw images.

base_key = 'loadlist'

static default_load_image(path: str | ~pathlib.Path, open_f: ~typing.Callable[[str | ~pathlib.Path], tuple[~numpy.ndarray, ~astropy.io.fits.header.Header]] = <function open_fits>) → Image

Function to open a raw image as an Image object

Parameters:

• path – path of raw image

• open_f – function to open the raw image

Returns:

Image object

description()

Return a description of the processor

Returns:

A description of the processor

image_type

alias of Image

class mirar.processors.utils.image_loader.ImageLoader(input_sub_dir: str = 'raw', input_img_dir: str | Path = PosixPath('/home/docs'), load_image: Callable[[str], Image | list[Image]] = None)

Bases: BaseImageProcessor

Processor to load raw images.

base_key = 'load'

static default_load_image(path: str | ~pathlib.Path, open_f: ~typing.Callable[[str | ~pathlib.Path], tuple[~numpy.ndarray, ~astropy.io.fits.header.Header]] = <function open_fits>) → Image

Function to open a raw image as an Image object

Parameters:

• path – path of raw image

• open_f – function to open the raw image

Returns:

Image object

description()

Return a description of the processor

Returns:

A description of the processor

image_type

alias of Image

exception mirar.processors.utils.image_loader.InvalidImage

Bases: NoncriticalProcessingError

Image should be skipped

class mirar.processors.utils.image_loader.LoadImageFromHeader(header_key: str = 'RAWPATH', copy_header_keys: str | list[str] = None, load_image: ~collections.abc.Callable[[str | ~pathlib.Path], ~mirar.data.image_data.Image] = <function open_raw_image>)

Bases: BaseImageProcessor

Class to load images from header information

base_key = 'load_from_header'

description()

Return a description of the processor

Returns:

A description of the processor

class mirar.processors.utils.image_loader.MEFLoader(input_sub_dir: str = 'raw', input_img_dir: str | Path = PosixPath('/home/docs'), load_image: Callable[[str], Image | list[Image]] = None)

Bases: ImageLoader

Processor to load MEF images.

base_key = 'load_mef'

static default_load_image(path: str | ~pathlib.Path, open_f: ~typing.Callable[[str | ~pathlib.Path], tuple[~astropy.io.fits.header.Header, list[~numpy.ndarray], list[~astropy.io.fits.header.Header]]] = <function open_mef_fits>, extension_key: str | None = None) → list[Image]

Function to open a raw image as an Image object

Parameters:

• path – path of raw image

• open_f – function to open the raw image

• extension_key – key to use to number the MEF frames

Returns:

Image object

mirar.processors.utils.image_loader.load_from_dir(input_dir: str | Path, open_f: Callable[[str | Path], Image | list[Image]]) → ImageBatch

Function to load all images in a directory

Parameters:

• input_dir – Input directory

• open_f – Function to open images

Returns:

ImageBatch object

mirar.processors.utils.image_loader.load_from_list(img_list: list[str | Path], open_f: Callable[[str | Path], Image | list[Image]]) → ImageBatch

Load images from a list of files

Parameters:

• img_list – Image list

• open_f – Function to open images

Returns:

ImageBatch object

mirar.processors.utils.image_loader.unzip(zipped_list: list[str]) → list[str]

Function to unzip a list of files?

Parameters:

zipped_list – List of zipped files

Returns:

List of renamed files

mirar.processors.utils.image_modifier module

Module to modify an image using a custom user-defined function

class mirar.processors.utils.image_modifier.CustomImageBatchModifier(image_batch_modifier: Callable[[ImageBatch], ImageBatch])

Bases: BaseImageProcessor

Class to modify an image using a custom user-defined function

base_key = 'custom_image_modifier'

description()

Return a description of the processor

Returns:

A description of the processor

mirar.processors.utils.image_plotter module

Module to plot 2D image-data as a pdf or png file.

class mirar.processors.utils.image_plotter.ImagePlotter(output_sub_dir: str = 'plots', plot_format: str = 'png', annotate_fields: str | list[str] | None = None)

Bases: BaseImageProcessor

Processor to plot images. Attributes :param plot_format: pdf or png?

base_key = 'plot'

description()

Return a description of the processor

Returns:

A description of the processor

mirar.processors.utils.image_rejector module

Module containing functions and processors to filter images

class mirar.processors.utils.image_rejector.ImageRejector(*args: tuple[str, str | list[str]])

Bases: BaseImageProcessor, CleanupProcessor

Processor to reject images based on the headers

base_key = 'reject'

description()

Return a description of the processor

Returns:

A description of the processor

mirar.processors.utils.image_rejector.filter_images(images: ImageBatch, header_key: str = 'TARGET', reject_values: str | list[str] = 'science') → ImageBatch

Finds the subset of images in the batch with do not have a value of <header_key> equal to a value in <reject values>

Parameters:

• images – images to filter

• header_key – key to check

• reject_values – unacceptable value(s)

Returns:

subset of passing images

mirar.processors.utils.image_saver module

Module for saving images

class mirar.processors.utils.image_saver.ImageSaver(output_dir_name: str, write_mask: bool = False, output_dir: str | Path = PosixPath('/home/docs'), compress: bool = False)

Bases: BaseImageProcessor

Processor to save images

base_key = 'save'

description()

Return a description of the processor

Returns:

A description of the processor

mirar.processors.utils.image_selector module

Module containing processors and functions to select a subset of images from a batch

class mirar.processors.utils.image_selector.ImageBatcher(split_key: str | list[str])

Bases: BaseImageProcessor

Module to split ImageBatch object into multiple DataBatch objects.

Images are batched using the split_key argument. For example, you can batch by filter, like this:

ImageBatcher(split_key=”filter”)

which will return N batches for the N different filters present in the directory you are reducing. If you do not require batching at some point in your reductions, you can split by BASE_NAME_KEY:

ImageBatcher(split_key=BASE_NAME_KEY)

which returns ImageBatches of length 1, one for each file in the directory you’re working with.

base_key = 'batch'

description() → str

Return a description of the processor

Returns:

A description of the processor

update_dataset(dataset: Dataset) → Dataset

Update a dataset after processing

Parameters:

dataset – Initial dataset

Returns:

Updated dataset

class mirar.processors.utils.image_selector.ImageDebatcher

Bases: BaseImageProcessor

Processor to group all incoming ImageBatch objects into a single batch. This is helpful if you’ve already batched at an earlier stage in your workflow, and you want to start over and batch by a different split key.

base_key = 'debatch'

description() → str

Return a description of the processor

Returns:

A description of the processor

update_dataset(dataset: Dataset) → Dataset

Update a dataset after processing

Parameters:

dataset – Initial dataset

Returns:

Updated dataset

class mirar.processors.utils.image_selector.ImageRebatcher(split_key: str | list[str])

Bases: ImageBatcher

Processor to regroup all incoming ImageBatch objects into a single batch, and then split by new keys. This is helpful if you’ve already batched at an earlier stage in your workflow, and you want to start over and batch by a different split key.

base_key = 'rebatch'

description() → str

Return a description of the processor

Returns:

A description of the processor

update_dataset(dataset: Dataset) → Dataset

Update a dataset after processing

Parameters:

dataset – Initial dataset

Returns:

Updated dataset

class mirar.processors.utils.image_selector.ImageSelector(*args: tuple[str, str | list[str]])

Bases: BaseImageProcessor, CleanupProcessor

Processor to only select a subset of images from a batch. Images can be selected using header keywords. For example, using:

ImageSelector((“OBSCLASS”, “SCIENCE”))

selects Images with header[“OBSCLASS”]==”SCIENCE”

base_key = 'select'

description()

Return a description of the processor

Returns:

A description of the processor

exception mirar.processors.utils.image_selector.ParsingError

Bases: KeyError, ProcessorError

Exception arising due to errors in parsing Image headers

mirar.processors.utils.image_selector.select_from_images(batch: ImageBatch, key: str = 'TARGET', target_values: str | list[str] = 'science') → ImageBatch

Returns a subset of images in a batch with have values of <key> equal to a value in <target values>

Parameters:

• batch – image batch to sort

• key – header key to filter on

• target_values – accepted value(s) for key

Returns:

image batch containing the subset of images which pass

mirar.processors.utils.image_selector.split_images_into_batches(images: ImageBatch, split_key: str | list[str]) → Dataset

Function to split a single ImageBatch object into multiple DataBatch objects. Each new batch will have the same value of <split_key>. Returns a dataset containing the new batches

Parameters:

• images – Image batch to split

• split_key – Key to split batch

Returns:

Dataset containing new image batches

mirar.processors.utils.mode_masker module

Module for filling NaN values into an image

class mirar.processors.utils.mode_masker.ModeMasker

Bases: BaseImageProcessor

Processor to mask the most common value in an image

base_key = 'modemask'

description()

Return a description of the processor

Returns:

A description of the processor

mirar.processors.utils.multi_ext_parser module

Module for loading and creating multi-extension FITS (MEF) images

class mirar.processors.utils.multi_ext_parser.MultiExtParser(input_sub_dir: str = 'raw', output_sub_dir: str = 'raw_split', input_img_dir: str = PosixPath('/home/docs'), output_img_dir: str | ~pathlib.Path = PosixPath('/home/docs'), load_image: ~collections.abc.Callable[[str], [<class 'numpy.ndarray'>, <class 'astropy.io.fits.header.Header'>]] = <function open_fits>, skip_first: bool = False, extension_num_header_key: str = None, only_extract_num: int = None)

Bases: BaseImageProcessor

Processor to split multi-extension FITS (MEF) images into single-extension files. Should be run before ImageLoader, especially for SEDMv2

base_key = 'load'

description()

Return a description of the processor

Returns:

A description of the processor

parse(path: str) → list

Function to open a raw MEF image, write each extension to a new file

Parameters:

path – path of raw MEF image

Returns:

new paths of single-extension files

mirar.processors.utils.multi_ext_parser.load_from_dir(input_dir: str | Path, parse_f: Callable[[list[str | Path]], Image]) → ImageBatch

Function to parse all MEF images in a directory

Parameters:

• input_dir – directory path

• parse_f – function to parse MEF images

Returns:

nothing…

mirar.processors.utils.nan_filler module

Module for filling NaN values in an image

class mirar.processors.utils.nan_filler.NanFiller

Bases: BaseImageProcessor

Processor to fill image nans

base_key = 'nanfill'

description()

Return a description of the processor

Returns:

A description of the processor

mirar.processors.utils.simulate_realtime module

Module to ‘simulate’ real-time data, by copying images into a directory. Used to test performance of the :module:`wintedrp.processors`

class mirar.processors.utils.simulate_realtime.RealtimeImageSimulator(input_img_dir: Path, input_img_names: str | list[str], output_dir_name: str, output_dir: str | Path = PosixPath('/home/docs'))

Bases: BaseImageProcessor

Processor to copy images into a directory, thereby simulating real-time data acquisition.

base_key = 'simrealtime'

description() → str

Return a description of the processor

Returns:

A description of the processor