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'

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

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'

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'

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'

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'

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.ImageLoader(input_sub_dir: str = 'raw', input_img_dir: str | Path = PosixPath('/home/docs'), load_image: Callable[[str], Image | list[Image]] | None = 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

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 = 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'

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 = 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.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_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'

mirar.processors.utils.image_rejector.filter_images(images: ImageBatch, header_key: str = 'TARGNAME', 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'))

Bases: BaseImageProcessor

Processor to save images

base_key = 'save'

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'

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'

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'

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 = 'TARGNAME', 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.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 = None, only_extract_num: int | None = 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'

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.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'