dran.fits package

Contents

dran.fits package#

Submodules#

dran.fits.backends module#

class dran.fits.backends.FitsBackendHandle(*args, **kwargs)[source]#

Bases: Protocol

Backend handle interface used by FITSReader.

This keeps FITSReader independent from astropy or fitsio APIs.

get_hdu_name(index)[source]#
Parameters:

index (int)

Return type:

str

get_hdu_type_name(index)[source]#
Parameters:

index (int)

Return type:

str

get_header(index)[source]#
Parameters:

index (int)

Return type:

Mapping[str, Any]

get_data(index)[source]#
Parameters:

index (int)

Return type:

Any

get_info()[source]#
Return type:

None

close()[source]#
Return type:

None

class dran.fits.backends.BackendSelection(name: str)[source]#

Bases: object

Parameters:

name (str)

name: str#
dran.fits.backends.select_backend()[source]#

Select a FITS backend.

Preference order: 1) fitsio 2) astropy

Return type:

BackendSelection

dran.fits.backends.open_fits_handle(path, memmap)[source]#

Open a FITS file with the selected backend.

Parameters:

  • path (Path) – FITS file path.

  • memmap (bool) – Used by astropy. fitsio ignores this flag.

Return type:

FitsBackendHandle

dran.fits.computed_columns module#

dran.fits.computed_columns.build_computed_column_names(log, band, src='', seed=None)[source]#

Build the unified computed column list for a given band and optional source.

Ordering: 1. COMM 2. COMM_WEATHER 3. band-specific weather groups (from WEATHER_BY_BAND) 4. computed labels (from build_computed_column_labels)

Special case: - If band is K and src is JUPITER, append JUP_WEATHER to the band weather groups.

Parameters:

  • log (Logger) – Logger.

  • band (str) – Band identifier (case-insensitive).

  • src (str) – Source name (case-insensitive).

  • seed (List[Any] | None) – Optional list to extend. If None, a new list is created.

Returns:

Assembled computed column list.

Return type:

List[Any]

dran.fits.conversion module#

dran.fits.conversion.counts_to_kelvin(counts, hz_per_k)[source]#

Convert raw detector counts to Kelvin using a Hz-per-K scale factor.

The conversion used mirrors the original behavior: (counts - counts[0]) / scale

If conversion fails for any reason (missing scale, invalid scale, scale=0), the function returns a zero array of the same length.

Parameters:

  • counts (ndarray) – Input counts array.

  • hz_per_k (Any) – Scale factor (Hz per Kelvin). Accepts any value convertible to float.

Returns:

Converted array in Kelvin, or a safe zero fallback.

Return type:

np.ndarray

dran.fits.hdu_rules module#

dran.fits.hdu_rules.scan_hdu_indices(hdu_len)[source]#

Return HDU indices that contain scan data based on your known file layouts.

Existing rules preserved: - If HDULEN == 5, scan index is [3] - If HDULEN == 7, scan indices are [3, 4, 5] - if HDULEN == 6, for older systems, scan indices are [3, 4, 5]

For unknown HDU lengths, return an empty list.

Parameters:

hdu_len (int)

Return type:

List[int]

dran.fits.init_fits_processing module#

dran.fits.init_fits_processing.process_fits_path(root_path, log, paths, args)[source]#

Dispatch processing for a single FITS file or a directory of FITS files. Returns extracted FITS records. Directory results are concatenated.

Parameters:
Return type:

List[Dict[str, Any]]

dran.fits.labels module#

dran.fits.labels.build_computed_column_labels(log, pol=('L', 'R'), pos=('O',), beam=('',), cols=('RMS', 'BSLOPE', 'BRMS', 'FLAG', 'S2N', 'TA', 'TAPEAKLOC'))[source]#

Generate computed column labels.

Rules: - Labels are built from beam + position + polarization + base column name. - For TA, append TAERR. - If position codes include multiple entries and include “O”, append:

  • PC

  • corrected TA (prefixed with C)

  • corrected TAERR

Parameters:

  • log (Logger) – Logger for diagnostics.

  • pol (Tuple[str, ...]) – Polarizations, for example (“L”, “R”).

  • pos (Tuple[str, ...]) – Position codes, for example (“N”, “S”, “O”).

  • beam (Tuple[str, ...]) – Beam identifiers, for example (“A”, “B”, “”).

  • cols (Tuple[str, ...]) – Base column names.

Returns:

Flattened list of label strings.

Return type:

List[str]

dran.fits.lazy_reader module#

class dran.fits.lazy_reader.LazyFITSReader(path, *, memmap=True, cache_data=True)[source]#

Bases: object

Lazy FITS reader.

Responsibilities: - Open and close a FITS file. - Provide header access immediately. - Load HDU data only when requested. - Offer HDU summaries without forcing data reads.

Parameters:
property handle: FitsBackendHandle#
open()[source]#
Return type:

None

close()[source]#
Return type:

None

clear_cache()[source]#
Return type:

None

hdu_count()[source]#
Return type:

int

get_info()[source]#
get_header(index)[source]#
Parameters:

index (int)

Return type:

Mapping[str, Any]

get_data(index)[source]#
Parameters:

index (int)

Return type:

Any

list_hdus()[source]#
Return type:

List[HduSummary]

dran.fits.observation_extractor module#

dran.fits.observation_extractor.build_observation_record(path, paths, source_name, frequency_mhz, band, hdu_len, logger)[source]#

Build an observation result record with common file metadata fields.

Notes - Keys are uppercase to match a “record schema” style used by CSV/DB exports. - “band” is accepted for schema completeness, even if not used yet.

Parameters:

  • path (Path) – Input FITS file path.

  • source_name (str) – Source name. Stored in normalized form for folder conventions.

  • frequency_mhz (float) – Center frequency derived from directory or context. Rounded to int for folder naming.

  • band (str) – Band label (e.g., L/S/C/X). Stored for completeness.

  • hdu_len (int) – Number of HDUs in the FITS file.

  • logger (Logger) – Application logger.

  • paths (ProjectPaths)

Return type:

Dict[str, Any]

dran.fits.observation_extractor.extract_observation(path, paths, band, log)[source]#

Extract one observation record from a FITS file path.

This preserves your existing behavior: - Uses headers to determine SRCNAME, CENTFREQ, BAND, HDULEN. - Populates a dict record using init_obs_row and get_complete_header_set. - Extracts scan arrays and derives weather quantities.

Returns:

A populated observation record dict.

Return type:

ObsRecord

Parameters:

dran.fits.observed_keys module#

dran.fits.observed_keys.build_observed_header_key_groups(band, log)[source]#

Build observed header key groups for a given band.

The indices reflect the existing downstream expectations in your pipeline.

For L/S:

0: PR_KEYS 1: FS_KEYS 2: ND_KEYS 3: [] 4: C_KEYS 5: [] (reserved for computed injection)

For other bands:

0: PR_KEYS 1: FS_KEYS 2: ND_KEYS 3: [] 4: [] 5: [] 6: C_KEYS 7: []

Parameters:

  • band (str) – Band identifier (case-insensitive).

  • log (Logger) – Logger.

Returns:

Dict[int, List[str]] mapping group index to cloned key lists.

Return type:

Dict[int, List[str]]

dran.fits.populate module#

dran.fits.populate.populate_scan_arrays(obs, scans_table, hdu_index, header_name, column_names, log)[source]#

Populate observation fields from a scans table. Extracts metadata and scan data for the given HDU, computes derived quantities for ZC headers, converts LCP and RCP counts to Kelvin, and stores results on the observation object.

Parameters:
Return type:

None

dran.fits.processing_fit module#

dran.fits.processing_fit.to_jsonable(value)[source]#

Convert a value into a JSON-serializable form. Handles dataclasses, objects with to_dict, scalars, collections, mappings, and plain objects, replacing non-finite floats with null equivalents.

Parameters:

value (Any)

Return type:

Any

dran.fits.processing_fit.dumps_json(value)[source]#

JSON-dump with a safe default conversion.

Parameters:

value (Any)

Return type:

str

dran.fits.processing_fit.populate_row(file_data, band, paths, log, args)[source]#

Run fitting per DATA column and populate derived fields back into the row dict. Returns the updated row (last row in file_data, consistent with existing behavior).

Parameters:
Return type:

Dict[str, Any]

dran.fits.schema module#

dran.fits.schema.build_header_key_schema(band, src, log)[source]#

Build the complete header key schema (observed + computed) for band and source.

Injection rule: - For L and S bands, computed columns are injected at HDU index 5. - For other bands, computed columns are injected at HDU index 7.

Parameters:

  • band (str) – Band identifier.

  • src (str) – Source name (used for special-case computed columns).

  • log (Logger) – Logger.

Returns:

Observed groups plus computed group injected.

Return type:

Dict[str, Any]

dran.fits.types module#

class dran.fits.types.HduSummary[source]#

Bases: TypedDict

index: int#
extname: str#
type: str#
rows: int | str#
cols: int | str#
shape: int | str#
has_data: bool#
class dran.fits.types.ObsRecord[source]#

Bases: TypedDict

Observation record produced by extraction.

This matches the existing behavior where keys are added dynamically. It stays flexible, while still providing a named type for clarity.

FILEPATH: str#
FILENAME: str#
OBSNAME: str#
PLOT_SAVE_DIR: str#
HDULEN: int#
OBJECT: str#
CENTFREQ: float#
BAND: str#
SCAN_ERROR: str | None#
class dran.fits.types.ScanArrays(x, y, scan_id, band, source, meta, series)[source]#

Bases: object

Normalized, analysis-ready arrays for one scan.

Notes - x is the independent axis. For example frequency, offset, or sample index. - y is the primary dependent series. For example TA or power. - meta keeps scalar values that you want to show in UI or logs. - series lets you store multiple named arrays if you have more than one y.

Parameters:
x: ndarray#
y: ndarray#
scan_id: str#
band: str#
source: str#
meta: Mapping[str, Any]#
series: Mapping[str, ndarray]#
class dran.fits.types.PopulateOptions(x_column, y_column, scan_id_column='SCAN', source_column='SRC', band_column='BAND', drop_non_finite=True, sort_by_x=True, extra_series=None)[source]#

Bases: object

Options to control how arrays are constructed from FITS content.

Parameters:
x_column: str#
y_column: str#
scan_id_column: str#
source_column: str#
band_column: str#
drop_non_finite: bool#
sort_by_x: bool#
extra_series: Sequence[str] | None#

Module contents#

Contents