Utilities API#

File I/O, global constants, multi-target tracking, argument handling, fiber overrides, and plotting helpers. See Package Architecture for how these fit into the pipeline.

Astropy-based I/O utilities for tramline map processing.

This module provides a simple interface for reading image data, header keywords, and other metadata from FITS files, replacing the Fortran TDFIO functions.

Usage example:

from kspecdr.io.image import ImageFile
with ImageFile('myfile.fits') as im_file:
    img = im_file.read_image_data()
    n_rows, n_cols = img.shape
    var = im_file.read_variance_data()
    fibre_types, nf = im_file.read_fiber_types(1000)
    value, comment = im_file.read_header_keyword('SPECTID')
    code = im_file.get_instrument_code()
class kspecdr.io.image.ImageFile(filename: str, mode: str = 'READ')[source]#

Bases: object

Astropy-based image file handler that replaces Fortran im_id functionality.

This class provides a simple interface for reading FITS files, similar to the Fortran TDFIO functions used in the original 2dfdr code.

add_history(history: str) → None[source]#

Add a history record to the primary HDU.

Parameters:

history (str) – History record to add

close() → None[source]#

Close the FITS file.

copy_fiber_table_from(source_file: ImageFile) → None[source]#

Copy fiber table from another file.

Parameters:

source_file (ImageFile) – Source file containing fiber table

delete_hdu(name: str) → bool[source]#

Delete an HDU by extension name.

Parameters:

name (str) – HDU extension name to remove.

Returns:

True if the HDU was found and removed, False otherwise.

Return type:

bool

delete_keyword(keyword: str) → None[source]#

Delete a keyword from the primary HDU.

Parameters:

keyword (str) – Keyword to delete

get_fiber_table_name() → str | None[source]#

Get the name of the fiber table HDU.

Returns:

Name of fiber table HDU if present, None otherwise

Return type:

str or None

get_header_value(keyword: str, default: str | None = None) → str[source]#

Get a header keyword value.

Parameters:

  • keyword (str) – Header keyword name

  • default (str, optional) – Default value if keyword not found

Returns:

Keyword value or default value

Return type:

str

get_instrument_code() → int[source]#

Get the instrument code from header.

Returns:

Instrument code

Return type:

int

get_size() → Tuple[int, int][source]#

Get the image dimensions (FITS convention).

Returns:

(nx, ny) dimensions of the image, where nx is width (NAXIS1) and ny is height (NAXIS2). Note: This corresponds to a numpy array of shape (ny, nx).

Return type:

tuple

has_fiber_table() → bool[source]#

Check if the file has a fiber table HDU.

Returns:

True if fiber table HDU exists, False otherwise

Return type:

bool

has_hdu(name: str) → bool[source]#

Check whether an HDU with the given EXTNAME exists.

Parameters:

name (str) – HDU extension name (case-insensitive comparison).

Return type:

bool

has_variance() → bool[source]#

Check if the file has a variance HDU.

Returns:

True if variance HDU exists, False otherwise

Return type:

bool

open(mode: str | None = None) → None[source]#

Open the FITS file.

Parameters:

mode (str, optional) – File access mode (‘READ’, ‘UPDATE’, ‘WRITE’). If None, uses the mode set during initialization.

read_fiber_table() → ndarray | None[source]#

Read fiber table data.

Returns:

Fiber table data if present, None otherwise

Return type:

np.ndarray or None

read_fiber_types(max_nfibres: int, overrides: Dict[int, str] | None = None) → Tuple[ndarray, int][source]#

Read fibre type information.

Parameters:

max_nfibres (int) – Maximum number of fibres

Returns:

(fiber_types, nf) - fiber type array and number of fibers

Return type:

tuple

read_header_keyword(keyword: str) → Tuple[str, str][source]#

Read a header keyword value and comment.

Parameters:

keyword (str) – Header keyword name

Returns:

(value, comment) - keyword value and comment

Return type:

tuple

read_image_data() → ndarray[source]#

Read image data from the primary HDU.

Returns:

Image data array with shape (rows, cols) [equivalent to (NAXIS2, NAXIS1)]

Return type:

np.ndarray

read_variance_data() → ndarray[source]#

Read variance data from the variance HDU.

Returns:

Variance data array with shape (rows, cols)

Return type:

np.ndarray

read_wave_data() → ndarray | None[source]#

Read wavelength data from the WAVELA HDU.

Returns:

Wavelength data array with shape (rows, cols) if found, else None

Return type:

np.ndarray or None

remove_fibers_beyond(max_fibers: int) → None[source]#

Remove fibers beyond a certain number (for TAIPAN).

Parameters:

max_fibers (int) – Maximum number of fibers to keep

save_as(filename: str) → None[source]#

Save the image file to a new filename.

Parameters:

filename (str) – Path to the new FITS file

save_primary_as(filename: str) → None[source]#

Save the primary HDU to a new filename.

Parameters:

filename (str)

set_class(class_type: str) → None[source]#

Set the CLASS keyword in the primary HDU.

Parameters:

class_type (str) – Class type to set

set_header_value(keyword: str, value, comment: str | None = None) → None[source]#

Set a header keyword value.

Parameters:

  • keyword (str) – Header keyword name

  • value – Keyword value

  • comment (str, optional) – Comment string for the keyword

write_fiber_table(fiber_data: ndarray, table_name: str = 'FIBRES') → None[source]#

Write fiber table data.

Parameters:

  • fiber_data (np.ndarray) – Fiber table data

  • table_name (str, optional) – Name of the fiber table HDU (‘FIBRES’ or ‘FIBRES_IFU’)

write_image_data(data: ndarray) → None[source]#

Write image data to the primary HDU.

Parameters:

data (np.ndarray) – Image data array with shape (nx, ny)

write_shifts_data(data: ndarray) → None[source]#

Write shifts data to the SHIFTS HDU.

Parameters:

data (np.ndarray) – Shifts data array with shape (ny, nx)

write_variance_data(data: ndarray) → None[source]#

Write variance data to the variance HDU.

Parameters:

data (np.ndarray) – Variance data array with shape (ny, nx)

write_wave_data(data: ndarray) → None[source]#

Write wavelength data to the WAVELA HDU.

Parameters:

data (np.ndarray) – Wavelength data array with shape (ny, nx)

Instrument constants and global definitions.

Multi-Target Tracking (MTT) algorithms.

kspecdr.tracking.multi_target_tracking(pk_grid: ndarray, nsteps: int, max_ntraces: int, max_displacement: float, *, min_fraction: float = 0.5, gap_limit: int | None = None, missing_cost: float | None = None, use_float32: bool = False) → Tuple[int, ndarray][source]#

Link per-step peak detections into traces using a Fortran-like Multi-Target Tracking (MTT) approach (PK_GRID2TRACES from 2dfdr) with LAP (Hungarian) assignment.

This is designed to be close in spirit to the Fortran MULTI_TARGET_TRACKING:

  • Unique assignment (1:1) between existing tracks and current-step points.

  • Cost uses Euclidean distance in (x, seq) space:

    cost = sqrt(dx^2 + gap^2)

  • Proximity gating:

    abs(dx) <= max_displacement gap <= gap_limit (default: nsteps//4, like Fortran’s NSEQ/4)

  • Performance: filters out tracks/points with no possible associations before forming the LAP matrix (critical for large fiber counts).

Parameters:

  • pk_grid (np.ndarray) – Peak grid array (nsteps, >= something). pk_grid[step, j] is peak position or 0.0 if absent.

  • nsteps (int) – Number of steps (sequences).

  • max_ntraces (int) – Maximum number of traces to track/return.

  • max_displacement (float) – Maximum allowed displacement in x for candidate associations (Fortran MAX_DIST).

  • min_fraction (float, optional) – Keep only traces with assigned points > min_fraction * nsteps. (Matches PK_GRID2TRACES default behavior when min_fraction=0.5.)

  • gap_limit (int, optional) – Maximum allowed gap (current_step - last_step) for candidate association. If None, uses nsteps//4 (Fortran uses NSEQ/4).

  • missing_cost (float, optional) – Cost for assigning a track to “missing” (dummy) instead of a real point. If None, uses slightly above the worst plausible real cost.

  • use_float32 (bool, optional) – Use float32 for cost computations to reduce memory and improve speed.

Returns:

(ntraces, trace_pts) –

ntracesint

Number of significant traces after filtering.

trace_ptsnp.ndarray

Shape (max_ntraces, nsteps), sorted by median position. Missing = 0.0.

Return type:

Tuple[int, np.ndarray]

Argument parsing, validation, and normalization utilities.

kspecdr.utils.args.init_args(args: Dict[str, Any]) → Dict[str, Any][source]#

Initialize and normalize args in place.

  • Parse and cache fiber type overrides from FIB<no> keys.

kspecdr.utils.args.validate_files_exist(args: Dict[str, Any], keys: Iterable[str], *, required: bool) → None[source]#

Validate that files referenced by args exist.

Parameters:

  • args (dict) – Argument dictionary.

  • keys (iterable) – Argument keys to validate.

  • required (bool) – Whether missing values should raise immediately.

kspecdr.utils.args.validate_reduce_object_args(args: Dict[str, Any]) → None[source]#

Sanity checks matching 2dfdr REDUCE_OBJECT_ARG_CHECKS.

Fiber-related helpers.

kspecdr.utils.fiber.apply_fiber_overrides(fiber_types: ndarray, overrides: Dict[int, str]) → ndarray[source]#

Apply 1-based fiber overrides to a fiber type array.

kspecdr.utils.fiber.get_override_from_args(args: Dict[str, Any]) → Dict[int, str][source]#

Return cached overrides from args or parse and cache them.

kspecdr.utils.fiber.parse_fib_overrides(args: Dict[str, Any]) → Dict[int, str][source]#

Parse FIB<no> arguments into a fiber type override map.

Returns:

Mapping of 1-based fiber number to single-character type code.

Return type:

dict

Spectral plotting utilities for KSPEC reduced data.

kspecdr.utils.plot.plot_red_fiber(red_path: str | Path, fiber_idx: int, *, ext_flux: str = 'PRIMARY', ext_wave: str = 'WAVELA', **kwargs) → tuple[matplotlib.pyplot.Figure, matplotlib.axes.Axes][source]#

Read a reduced FITS file and plot a single fiber spectrum.

Parameters:

  • red_path (str or Path) – Path to the _red.fits file.

  • fiber_idx (int) – Zero-based fiber index (row in the PRIMARY/WAVELA arrays).

  • ext_flux (str) – FITS extension name for flux (default "PRIMARY").

  • ext_wave (str) – FITS extension name for wavelength (default "WAVELA").

  • **kwargs – Forwarded to plot_spectrum().

Return type:

fig, ax

kspecdr.utils.plot.plot_red_file(red_path: str | Path, fiber_indices: Sequence[int] | None = None, *, ncols: int = 2, ext_flux: str = 'PRIMARY', ext_wave: str = 'WAVELA', target_z: float = 0.0, ylim: tuple | None = None, figsize_per_panel: tuple = (10, 3), **kwargs) → tuple[matplotlib.pyplot.Figure, ndarray][source]#

Plot spectra for multiple fibers from a single reduced file, one panel each.

Parameters:

  • red_path (str or Path) – Path to the _red.fits file.

  • fiber_indices (sequence of int, optional) – Which fibers to plot. Defaults to all fibers.

  • ncols (int) – Number of columns in the subplot grid.

  • ext_flux (str) – FITS extension names.

  • ext_wave (str) – FITS extension names.

  • target_z (float) – Redshift applied to line annotations.

  • ylim ((ymin, ymax), optional) – Shared Y limits for all panels.

  • figsize_per_panel ((w, h)) – Size of each individual panel.

  • **kwargs – Forwarded to plot_spectrum().

Return type:

fig, axes (axes is a 2-D ndarray of Axes)

kspecdr.utils.plot.plot_spectrum(wave: ndarray, flux: ndarray, *, ax: matplotlib.axes.Axes | None = None, target_z: float = 0.0, ref_wave: ndarray | None = None, ref_flux: ndarray | None = None, ref_label: str = 'ref', ref_color: str = 'tab:red', ylim: tuple | None = None, title: str | None = None, xlabel: str = 'Wavelength ($\\AA$)', ylabel: str = 'Flux (10$^{-17}$ erg s$^{-1}$ cm$^{-2}$ $\\AA^{-1}$)', show_emission: bool = True, show_absorption: bool = True, show_sky: bool = True, show_telluric: bool = True, figsize: tuple = (10, 4)) → tuple[matplotlib.pyplot.Figure, matplotlib.axes.Axes][source]#

Single-panel spectral plot with emission, absorption, sky, and telluric feature annotations.

Parameters:

  • wave (array-like) – Observed-frame wavelength array [Å].

  • flux (array-like) – Flux array (same length as wave).

  • ax (Axes, optional) – Existing axes to plot into. A new figure is created if None.

  • target_z (float) – Object redshift applied to rest-frame line wavelengths.

  • ref_wave (array-like, optional) – Optional reference spectrum (e.g. SDSS) to overplot in ref_color.

  • ref_flux (array-like, optional) – Optional reference spectrum (e.g. SDSS) to overplot in ref_color.

  • ref_label (str) – Legend label for the reference spectrum.

  • ref_color (str) – Colour for the reference spectrum.

  • ylim ((ymin, ymax), optional) – Y-axis limits. If None, set from the 99th percentile of |flux|.

  • title (str, optional) – Axes title.

  • xlabel (str) – Axis labels.

  • ylabel (str) – Axis labels.

  • show_emission (bool) – Toggle individual annotation layers.

  • show_absorption (bool) – Toggle individual annotation layers.

  • show_sky (bool) – Toggle individual annotation layers.

  • show_telluric (bool) – Toggle individual annotation layers.

  • figsize ((width, height)) – Figure size when a new figure is created.

Return type:

fig, ax