dran.utils package

Submodules

dran.utils.config module

class dran.utils.config.ObservationPathParts(root_data_dir: pathlib.Path, category: str, source: str, frequency: int, filename: str, full_path: pathlib.Path, wavelength_cm: float | None = None, beam: str | None = None, band_folder: str | None = None)

Bases: object

Parameters:

• root_data_dir (Path)

• category (str)

• source (str)

• frequency (int)

• filename (str)

• full_path (Path)

• wavelength_cm (float | None)

• beam (str | None)

• band_folder (str | None)

root_data_dir: Path

category: str

source: str

frequency: int

filename: str

full_path: Path

wavelength_cm: float | None = None

beam: str | None = None

band_folder: str | None = None

class dran.utils.config.ObservationPathPartsFolder(root_data_dir: pathlib.Path, category: str, source: str, frequency: int, foldername: str, full_path: pathlib.Path, wavelength_cm: float | None = None, beam: str | None = None, band_folder: str | None = None)

Bases: object

Parameters:

• root_data_dir (Path)

• category (str)

• source (str)

• frequency (int)

• foldername (str)

• full_path (Path)

• wavelength_cm (float | None)

• beam (str | None)

• band_folder (str | None)

root_data_dir: Path

category: str

source: str

frequency: int

foldername: str

full_path: Path

wavelength_cm: float | None = None

beam: str | None = None

band_folder: str | None = None

class dran.utils.config.ProjectPaths(workdir: pathlib.Path, db_path: pathlib.Path, log_path: pathlib.Path, invalid_files_path: pathlib.Path, symlinks_path: pathlib.Path, diagnostics_dir: pathlib.Path, plots_dir: pathlib.Path)

Bases: object

Parameters:

• workdir (Path)

• db_path (Path)

• log_path (Path)

• invalid_files_path (Path)

• symlinks_path (Path)

• diagnostics_dir (Path)

• plots_dir (Path)

workdir: Path

db_path: Path

log_path: Path

invalid_files_path: Path

symlinks_path: Path

diagnostics_dir: Path

plots_dir: Path

exception dran.utils.config.ObservationPathError

Bases: Exception

Raised when an observation path cannot be parsed.

dran.utils.frequency_utils module

dran.utils.frequency_utils.get_band_from_frequency(frequency, log)

Determine the satellite frequency band for a given frequency (in MHz).

Parameters:

• frequency (float | int) – Frequency in megahertz (MHz).

• log (Logger)

Returns:

The corresponding band identifier

(e.g., ‘L’, ‘S’, ‘C’, ‘CM’, ‘X’, ‘Ku’, ‘K’, ‘Ka’).

Return type:

str

Raises:

ValueError – If the input frequency is invalid or outside known bands.

dran.utils.frequency_utils.get_frequency_range_from_band(band, log)

Return the frequency range in MHz for a given band identifier.

Parameters:

• band (str) – Band identifier such as ‘L’, ‘S’, ‘C’, ‘CM’, ‘X’, ‘Ku’, ‘K’, ‘Ka’.

• log (Logger) – Logger instance.

Returns:

Tuple of start and end frequency in MHz.

Raises:

• TypeError – Invalid band type.

• ValueError – Unknown band.

Return type:

tuple[int, int]

dran.utils.fs module

dran.utils.fs.build_paths(workdir)

Parameters:

workdir (Path)

Return type:

ProjectPaths

dran.utils.fs.resolve_existing_path(path, log, paths=None)

Resolve a filesystem path strictly.

Returns:

Resolved absolute path.

Raises:

• FileNotFoundError – If the path does not exist.

• OSError – If the path cannot be resolved.

Parameters:

• path (Path)

• log (Logger)

• paths (ProjectPaths | None)

Return type:

Path

dran.utils.fs.parse_source_directory_path(path)

Extract frequency and source name from a directory path.

Expected structure:

…/<SOURCE>/<FREQ>/ …/<SOURCE>/*_<FREQ>/

Returns:

(frequency, source)

Parameters:

path (str | Path)

Return type:

Tuple[int, str]

dran.utils.fs.parse_source_frequency_band_from_path_if_folder(path, log)

Parse (source_name, frequency_mhz, band) from a FITS file path.

Assumes a directory layout like:

…/<SOURCE_NAME>/<FREQUENCY_MHZ>/<file>.fits

Returns:

(source_name_upper, frequency_mhz, band)

Parameters:

• path (Path)

• log (Logger)

Return type:

Tuple[str, int, str]

dran.utils.fs.parse_plot_path(path)

Extract source name, frequency, and filename from a plot path.

Expected structure:

base / <src_name> / <freq> / <filename>

Returns:

(src_name, freq, filename)

Parameters:

path (str | Path)

Return type:

Tuple[str, str, str]

dran.utils.fs.resolve_existing_path_without_logger(path)

Resolve a filesystem path strictly.

Returns:

Resolved absolute path.

Parameters:

path (Path)

Return type:

Path

dran.utils.fs.parse_observation_path(path_str)

Parse observation file paths for DRAN.

Supported layouts: data/calibration/3C123/12178/file.fits data/calibration/3C123_12178/file.fits data/calibration/3C123_13NB/file.fits data/calibration/3C123_13NB_dichroic_on/file.fits data/calibration/3C123_35/file.fits data/calibration/3C123_3.5NB/file.fits data/calibration/Jup/file.fits

Parameters:

path_str (str)

Return type:

ObservationPathParts

dran.utils.fs.parse_observation_path_if_folder(path_str)

Parse observation file paths for DRAN.

Supported layouts: data/calibration/3C123/12178/ data/calibration/3C123_12178/ data/calibration/3C123_13NB/ data/calibration/3C123_13NB_dichroic_on/ data/calibration/3C123_35/ data/calibration/3C123_3.5NB/ data/calibration/3C123/ data/calibration/Jup/

Parameters:

path_str (str)

Return type:

ObservationPathPartsFolder

dran.utils.fs.ensure_directory_exists(base_folder, log=None)

Ensure the given path exists and is a directory.

Parameters:

• base_folder (Path)

• log (Logger | None)

Return type:

None

dran.utils.fs.ensure_output_directories(paths, log, *, overwrite_if_contains='diagnostic')

Ensure required output directories exist.

Default behavior: - If the directory path contains the token “diagnostic”, it is recreated. - All other directories are created if missing.

Parameters:

• paths (Iterable[Path])

• log (Logger)

• overwrite_if_contains (str)

Return type:

None

dran.utils.fs.recreate_dir(folder_path, log)

Recreate a directory.

If the directory already exists, it is removed with all contents, then recreated (including parents).

Parameters:

• folder_path (Path)

• log (Logger)

Return type:

None

dran.utils.fs.create_dir(folder_path, log)

Ensure a directory exists.

Creates parent directories as needed. If it already exists, it is left unchanged.

Parameters:

• folder_path (Path)

• log (Logger)

Return type:

None

dran.utils.fs.clear_diagnostics_dir(path, log)

Remove all files and subdirectories from a diagnostics directory. Safely deletes directory contents if the path exists. Logs a warning for any item that cannot be removed.

Parameters:

• path (Path)

• log (Logger)

Return type:

None

dran.utils.fs.compute_file_hash(path, *, chunk_size=1048576)

Compute a SHA-256 hash for a file by streaming its contents.

Parameters:

• path (Path)

• chunk_size (int)

Return type:

str

dran.utils.fs.load_invalid_paths(paths, log)

Load previously recorded invalid paths from disk.

Returns an empty set if the log file does not exist.

Parameters:

• paths (ProjectPaths)

• log (Logger)

Return type:

Set[str]

dran.utils.fs.load_symlink_paths(paths, log)

Load previously recorded symlink paths from disk.

Parameters:

• paths (ProjectPaths)

• log (Logger)

Return type:

Set[str]

dran.utils.fs.append_invalid_path(path, paths, log)

Append a path to the invalid path registry on disk.

Creates parent directories for the registry file if needed.

Parameters:

• path (Path)

• paths (ProjectPaths)

• log (Logger)

Return type:

None

dran.utils.fs.append_symlink_path(path, paths, log)

Append a symlink path to the symlink registry on disk.

Parameters:

• path (Path)

• paths (ProjectPaths)

• log (Logger)

Return type:

None

dran.utils.fs.record_invalid_path_once(path, paths, log, reason)

Record an invalid path only once.

If the path is already present, nothing is appended.

Parameters:

• path (Path)

• paths (ProjectPaths)

• log (Logger)

• reason (str)

Return type:

None

dran.utils.fs.record_symlink_path_once(path, paths, log)

Record a symlink path only once in the symlink registry.

Parameters:

• path (Path)

• paths (ProjectPaths)

• log (Logger)

Return type:

None

dran.utils.port_utils module

dran.utils.port_utils.find_pids_by_port(port)

Find all PIDs using a given port.

Uses lsof, which is available on macOS and most Linux systems.

Parameters:

port (int)

Return type:

List[int]

dran.utils.port_utils.kill_processes_on_port(port, force=False)

Kill all processes using a given port.

Parameters:

• port (int) – Port number to check

• force (bool) – If True, use SIGKILL (-9). Otherwise SIGTERM (default)

Return type:

None

dran.utils.time_utils module

dran.utils.time_utils.doy_to_date(year, doy)

Convert day-of-year to a calendar date.

Parameters:

• year (int) – Four-digit year, for example 2024.

• doy (int) – Day of year, starting at 1.

Returns:

datetime.date instance.

Return type:

date

dran.utils.time_utils.parse_doy_timestamp(value)

Parse a timestamp in the form YYYYdDOY_HHhMMmSSs.

Example input:

2023d281_00h01m59s

Returns:

datetime.datetime instance in UTC-like naive time.

Parameters:

value (str)

Return type:

datetime

Module contents