opticam_new.reduction.catalog

Classes

Catalog

Create a catalog of sources from OPTICAM data.

Functions

save_catalog(filters, stacked_images, catalogs, ...)

save_stacked_images(stacked_images, out_directory, ...)

Save the stacked images to a compressed FITS file.

save_backgrounds(camera_files, background_median, ...)

Plot the time-varying background for each camera.

get_batches(input)

get_batch_size(L)

Compute the batch size for a given input length.

Module Contents

class opticam_new.reduction.catalog.Catalog(out_directory, data_directory=None, c1_directory=None, c2_directory=None, c3_directory=None, rebin_factor=1, flat_corrector=None, background=None, finder=None, threshold=5, aperture_selector=np.median, remove_cosmic_rays=True, number_of_processors=cpu_count() // 2, show_plots=True, verbose=True)

Create a catalog of sources from OPTICAM data.

Parameters:

  • out_directory (str)

  • data_directory (None | str)

  • c1_directory (None | str)

  • c2_directory (None | str)

  • c3_directory (None | str)

  • rebin_factor (int)

  • flat_corrector (None | opticam_new.reduction.correctors.FlatFieldCorrector)

  • background (None | Callable)

  • finder (None | Callable)

  • threshold (float)

  • aperture_selector (Callable)

  • remove_cosmic_rays (bool)

  • number_of_processors (int)

  • show_plots (bool)

  • verbose (bool)

verbose = True
out_directory
logger
data_directory = None
c1_directory = None
c2_directory = None
c3_directory = None
rebin_factor = 1
flat_corrector = None
aperture_selector
threshold = 5
remove_cosmic_rays = True
number_of_processors
show_plots = True
file_paths = []
ignored_files = []
colours = ['tab:blue', 'tab:orange', 'tab:green', 'tab:red', 'tab:purple', 'tab:brown', 'tab:pink',...
transforms
unaligned_files = []
catalogs
psf_params
_scan_data_directory()

Scan the data directory for files and extract the MJD, filter, binning, and gain from each file header.

Raises

ValueError

If more than 3 filters are found.

ValueError

If the binning is not consistent.

Return type:

None

_get_header_info(file)

Get the MJD, filter, binning, and gain from a file header.

Parameters

filestr

The file path.

Returns

Tuple[float, str, str, float]

The BMJD, filter, binning, and gain dictionaries.

Raises

KeyError

If the file header does not contain the required keys.

Parameters:

file (str)

Return type:

Tuple[numpy.typing.ArrayLike | None, str | None, str | None, float | None]

_parse_header_results(results)

Parse the results returned by self._get_header_info().

Parameters

resultsTuple

The results.

Returns

Tuple[str, str]

The filter dictionary (file : filter).

Raises

ValueError

If more than 3 filters are found.

ValueError

If the binning is not consistent.

Parameters:

results (Tuple[float, float, str, str, float])

Return type:

Dict[str, str]

_log_parameters()

Log any and all object parameters to a JSON file.

_plot_time_between_files()

Plot the times between each file for each camera.

Parameters

showbool

Whether to display the plot.

Return type:

None

_set_psf_params(fltr)

Set the PSF parameters for a given filter based on the catalog data.

Parameters

fltrstr

The filter for which to set the PSF parameters.

Parameters:

fltr (str)

Return type:

None

_get_data(file, return_error=False)

Get data from a file.

Parameters

filestr

Directory path to file.

return_errorbool, optional

Whether to return the error array, by default False.

Returns

NDArray | Tuple[NDArray, NDArray]

The data array or the data and error arrays.

Parameters:

  • file (str)

  • return_error (bool)

Return type:

numpy.typing.NDArray | Tuple[numpy.typing.NDArray, numpy.typing.NDArray]

_get_source_coords_from_image(image, bkg=None, away_from_edge=False, n_sources=None)

Get an array of source coordinates from an image in descending order of source brightness.

Parameters

imageNDArray

The non-background-subtracted image from which to extract source coordinates.

bkgBackground2D, optional

The background of the image, by default None. If None, the background is estimated from the image.

away_from_edgebool, optional

Whether to exclude sources near the edge of the image, by default False.

n_sourcesint, optional

The number of source coordinates to return, by default None (all sources will be returned).

Returns

NDArray

The source coordinates in descending order of brightness.

Parameters:

  • image (numpy.typing.NDArray)

  • bkg (photutils.background.Background2D | None)

  • away_from_edge (bool | None)

  • n_sources (int | None)

Return type:

numpy.typing.NDArray

create_catalogs(max_catalog_sources=50, n_alignment_sources=3, transform_type='translation', rotation_limit=None, translation_limit=None, scale_limit=None, overwrite=False, show_diagnostic_plots=False)

Initialise the source catalogs for each camera. Some aspects of this method are parallelised for speed.

Parameters

max_catalog_sourcesint, optional

The maximum number of sources above the specified threshold that will be included in the catalog, by default 50. Only the brightest max_catalog_sources sources are included in the catalog.

n_alignment_sourcesint, optional

The (maximum) number of sources to use for image alignment, by default 3. If transform_type=’translation’, n_alignment_sources must be >= 1, and the brightest n_alignment_sources sources are used for image alignment. If transform_type=’affine’, n_alignment_sources must be >= 3 and represents that maximum number of sources that may be used for image alignment.

transform_typeLiteral[‘affine’, ‘translation’], optional

The type of transform to use for image alignment, by default ‘translation’. ‘translation’ performs simple x, y translations, while ‘affine’ uses astroalign.find_transform().

rotation_limitfloat, optional

The maximum rotation limit (in degrees) for affine transformations, by default None (no limit).

scale_limitfloat, optional

The maximum scale limit for affine transformations, by default None (no limit).

translation_limitfloat | Iterable[float], optional

The maximum translation limit for transformations, by default None (no limit). Can be a scalar value that applies to both x- and y-translations, or an iterable where the first value defines the x-translation limit and the second value defines the y-translation limit.

overwritebool, optional

Whether to overwrite existing catalogs, by default False.

show_diagnostic_plotsbool, optional

Whether to show diagnostic plots, by default False. Diagnostic plots are saved to out_directory, so this parameter only affects whether the plots are displayed in the console.

Parameters:

  • max_catalog_sources (int)

  • n_alignment_sources (int)

  • transform_type (Literal['affine', 'translation'])

  • rotation_limit (float | None)

  • translation_limit (float | Iterable[float] | None)

  • scale_limit (float | None)

  • overwrite (bool)

  • show_diagnostic_plots (bool)

Return type:

None

_align_images(batch, reference_image_shape, reference_coords, transform_type, rotation_limit, scale_limit, translation_limit, n_alignment_sources)

Align an image based on some reference coordinates.

Parameters

file: str

The file path.

reference_imageNDArray

The reference image.

reference_coordsNDArray

The source coordinates in the reference image.

transform_typeLiteral[‘affine’, ‘translation’]

The type of transform to use for image alignment.

rotation_limitfloat | None

The maximum rotation limit (in degrees) for image alignment.

scale_limitfloat | None

The maximum scaling limit for image alignment.

translation_limitIterable[float] | None

The maximum translation limit for image alignment.

n_alignment_sourcesint

The (maximum) number of sources to use for image alignment.

Returns

Tuple[List[float], float, float]

The transform parameters, background median, and background RMS.

Parameters:

  • batch (List[str])

  • reference_image_shape (Tuple[int])

  • reference_coords (numpy.typing.NDArray)

  • transform_type (Literal['affine', 'translation'])

  • rotation_limit (float | None)

  • scale_limit (float | None)

  • translation_limit (Iterable[float] | None)

  • n_alignment_sources (int)

Return type:

Tuple[numpy.typing.NDArray, Dict[str, float], Dict[str, float], Dict[str, float]]

_valid_transform(file, transform, rotation_limit, scale_limit, translation_limit)

Find whether a transform is valid given some transform limits.

Parameters

filestr

The file being transformed.

transformSimilarityTransform

The transform.

rotation_limitfloat | None

The rotation limit.

scale_limitfloat | None

The scale limit.

translation_limitIterable[float] | None

The translation limit.

Returns

bool

Whether the transform is valid.

Parameters:

  • file (str)

  • transform (skimage.transform.SimilarityTransform)

  • rotation_limit (float | None)

  • scale_limit (float | None)

  • translation_limit (Iterable[float] | None)

Return type:

bool

_parse_alignment_results(results, fltr)

Parse the results of image alignment.

Parameters

resultsTuple

The results.

fltrstr

The filter.

Returns

Tuple[NDArray, Dict[str, float], Dict[str, float]]:

The stacked image, background medians, and background RMSs.

Parameters:

  • results (Tuple)

  • fltr (str)

Return type:

Tuple[numpy.typing.NDArray, Dict[str, float], Dict[str, float]]

_plot_background_meshes(stacked_images, show)

Plot the background meshes on top of the catalog images.

Parameters

stacked_imagesDict[str, NDArray]

The stacked images for each camera.

showbool

Whether to display the plot.

Parameters:

  • stacked_images (Dict[str, numpy.typing.NDArray])

  • show (bool)

Return type:

None

_visualise_psfs(image, fltr, show)

Generate PSF plots for each source in an image.

Parameters

imageNDArray

The image (not background subtracted).

fltrstr

The image filter.

show: bool

Whether to display the plot.

Parameters:

  • image (numpy.typing.NDArray)

  • fltr (str)

  • show (bool)

Return type:

None

create_gifs(keep_frames=True, overwrite=False)

Create alignment gifs for each camera. Some aspects of this method are parallelised for speed. The frames are saved in out_directory/diag/*-band_gif_frames and the GIFs are saved in out_directory/cat.

Parameters

keep_framesbool, optional

Whether to save the GIF frames in out_directory/diag, by default True. If False, the frames will be deleted after the GIF is saved.

overwritebool, optional

Whether to overwrite existing GIFs, by default False.

Parameters:

  • keep_frames (bool)

  • overwrite (bool)

Return type:

None

_create_gif_frames(file, fltr)

Create a gif frames from a batch of images and save it to the out_directory.

Parameters

filestr

The list of file names in the batch.

fltrstr

The filter.

Parameters:

  • file (str)

  • fltr (str)

Return type:

None

_compile_gif(fltr, keep_frames)

Create a gif from the frames saved in out_directory.

Parameters

fltrstr

The filter.

keep_framesbool

Whether to keep the frames after the gif is saved.

Parameters:

  • fltr (str)

  • keep_frames (bool)

Return type:

None

photometry(photometer)

Perform photometry on the catalogs using the provided photometer.

Parameters

photometerBasePhotometer

The photometer. Should be a subclass of BasePhotometer, or implement a compute method that follows the BasePhotometer interface.

Parameters:

photometer (opticam_new.reduction.photometers.BasePhotometer)

Return type:

None

_photometry(photometer, source_coords, fltr, file)
Parameters:
Return type:

Dict[str, List]

identify_gaps(files)

Identify gaps in the observation sequence and logs them to log_dir/diag/gaps.txt.

Parameters

filesList[str]

The list of files for a single filter.

Parameters:

files (List[str])

Return type:

None

opticam_new.reduction.catalog.save_catalog(filters, stacked_images, catalogs, out_directory, show)
Parameters:

  • filters (List[str])

  • stacked_images (Dict[str, numpy.typing.NDArray])

  • catalogs (Dict[str, astropy.table.QTable])

  • out_directory (str)

  • show (bool)

Return type:

None

opticam_new.reduction.catalog.save_stacked_images(stacked_images, out_directory, overwrite)

Save the stacked images to a compressed FITS file.

Parameters

stacked_imagesDict[str, NDArray]

The stacked images (filter: stacked image).

Parameters:

  • stacked_images (Dict[str, numpy.typing.NDArray])

  • out_directory (str)

  • overwrite (bool)

Return type:

None

opticam_new.reduction.catalog.save_backgrounds(camera_files, background_median, background_rms, bmjds, t_ref, out_directory, show)

Plot the time-varying background for each camera.

Parameters

camera_filesDict[str, str]

The files for each camera {fltr: file}.

background_medianDict[str, List]

The median background for each camera.

background_rmsDict[str, List]

The background RMS for each camera.

bmjdsDict[str, float]

The Barycentric MJD dates for each image {file: BMJD}.

t_reffloat

The reference BMJD.

out_directorystr

The directory to which the resulting files will be saved.

show: bool

Whether to display the plot.

Parameters:

  • camera_files (Dict[str, str])

  • background_median (Dict[str, Dict[str, numpy.typing.NDArray]])

  • background_rms (Dict[str, Dict[str, numpy.typing.NDArray]])

  • bmjds (Dict[str, float])

  • t_ref (float)

  • out_directory (str)

  • show (bool)

Return type:

None

opticam_new.reduction.catalog.get_batches(input)
Parameters:

input (List[Any])

Return type:

List[List[Any]]

opticam_new.reduction.catalog.get_batch_size(L)

Compute the batch size for a given input length.

Parameters

Lint

The length of the input.

Returns

int

The batch size.

Parameters:

L (int)

Return type:

int