zyra.utils package

Utilities used across Zyra (dates, files, images, credentials).

Avoid importing optional heavy dependencies at package import time to keep the CLI lightweight when only a subset of functionality is needed (e.g., pipeline runner). Submodules can still be imported directly when required.

class zyra.utils.CredentialManager(filename: str | None = None, namespace: str | None = None)

Bases: object

Manage app credentials from a dotenv file.

Parameters:

• filename (str, optional) – Path to a dotenv file containing key=value pairs.

• namespace (str, optional) – Optional prefix to apply to all keys when stored/retrieved.

Examples

Namespaced keys:

cm = CredentialManager(".env", namespace="MYAPP_")
cm.read_credentials(expected_keys=["API_KEY"])  # expects MYAPP_API_KEY

add_credential(key: str, value: str) → None

Add or update a credential value in memory.

Parameters:

• key (str) – Base key name (namespace is applied automatically).

• value (str) – Credential value to store.

clear_credentials() → None

Remove all tracked credentials from memory.

delete_credential(key: str) → None

Delete a credential by key if present.

Parameters:

key (str) – Base key name (namespace is applied automatically).

get_credential(key: str) → str

Retrieve a credential value by key (with namespace applied).

Parameters:

key (str) – Base key name (namespace is applied automatically).

Returns:

Stored value for the namespaced key.

Return type:

str

Raises:

KeyError – If the key is not present in memory.

list_credentials(expected_keys: Iterable[str] | None = None) → list[str]

List tracked credential keys, checking for expected ones when provided.

Parameters:

expected_keys (Iterable[str], optional) – Keys to verify are present.

Returns:

Keys currently stored in memory.

Return type:

list of str

Raises:

KeyError – If any expected keys are missing.

read_credentials(expected_keys: Iterable[str] | None = None) → None

Read credentials from the dotenv file into memory.

Parameters:

expected_keys (Iterable[str], optional) – Keys that must be present; raises if any are missing.

Raises:

• FileNotFoundError – If the dotenv path cannot be resolved.

• KeyError – If any expected keys are missing after reading.

property tracked_keys: set[str]

Return the set of keys currently tracked in memory.

class zyra.utils.DateManager(date_formats: list[str] | None = None)

Bases: object

High-level utilities for working with dates and filenames.

Parameters:

date_formats (list of str, optional) – Preferred strftime-style formats to use when parsing dates from filenames (e.g., ["%Y%m%d"]).

Examples

Use a custom filename format first, then fall back to ISO-like detection:

dm = DateManager(["%Y%m%d%H%M%S"])
when = dm.extract_date_time("frame_20240101093000.png")

__init__(date_formats: list[str] | None = None) → None

Optionally store preferred date formats for filename parsing.

calculate_expected_frames(start_datetime: datetime, end_datetime: datetime, period_seconds: int) → int

Calculate expected frame count between two datetimes at a cadence.

Returns:

Number of expected frames (inclusive of endpoints).

Return type:

int

datetime_format_to_regex(datetime_format: str) → str

Convert a datetime format string to a regex pattern.

extract_date_time(string: str) → str | None

Extract a date string from a filename/text using known formats.

Tries known formats first; falls back to a simple ISO-like pattern.

extract_dates_from_filenames(directory_path: str, image_extensions: Iterable[str] = ('.jpg', '.jpeg', '.png', '.gif', '.bmp', '.dds')) → tuple[str | None, str | None]

Extract dates from the first and last image file names in a directory.

Parameters:

• directory_path (str) – Directory to scan for images.

• image_extensions (Iterable[str]) – File extensions to include when scanning.

Returns:

(first_date, last_date) as strings, or (None, None).

Return type:

tuple

find_missing_frames(directory, period_seconds, datetime_format, filename_format, filename_mask, start_datetime, end_datetime)

Find missing frames in a local directory with inconsistent period, only for image files.

find_missing_frames_and_predict_names(timestamps, period_seconds, filename_pattern)

Find gaps and overfrequent frames in timestamps and predict names.

find_start_end_datetimes(directory: str)

Find earliest and latest datetimes from filenames in a directory.

get_date_range(period: str) → tuple[datetime, datetime]

Compute a date range ending at the current minute from a period spec.

Parameters:

period (str) – Period string such as "1Y", "6M", "7D", or "24H".

Returns:

Start and end datetimes for the period ending at “now” (rounded to minute).

Return type:

(datetime, datetime)

get_date_range_iso(iso_duration: str) → tuple[datetime, datetime]

Compute a date range ending now from an ISO-8601 duration (e.g., P1Y, P6M, P7D, PT24H).

Supports a subset of ISO-8601: years (Y), months (M), days (D), hours (H) with the “P…T…” structure. Examples: “P1Y”, “P6M”, “P7D”, “PT24H”.

is_date_in_range(filepath: str, start_date: datetime, end_date: datetime) → bool

Check if a filename contains a date within a range.

Parameters:

• filepath (str) – Path or filename containing a date stamp.

• start_date (datetime) – Inclusive start of the permitted range.

• end_date (datetime) – Inclusive end of the permitted range.

Returns:

True if a parsed date falls within the range, else False.

Return type:

bool

parse_timestamps_from_filenames(filenames, datetime_format)

Parse timestamps from filenames based on the given format.

class zyra.utils.FileUtils

Bases: object

Namespace for file-related helper routines.

Examples

While most functions are provided at module-level, a class instance can be created if you prefer an object to group related operations.

class zyra.utils.JSONFileManager(file_path: str)

Bases: object

Convenience wrapper for manipulating JSON files.

Parameters:

file_path (str) – Path to a JSON file on disk.

Examples

Read, update, and save:

jm = JSONFileManager("./data.json")
jm.data["foo"] = "bar"
jm.save_file()

read_file() → None

Read the JSON file from disk into memory.

Returns:

Populates self.data or sets it to None on error.

Return type:

None

save_file(new_file_path: str | None = None) → None

Write the in-memory data back to disk.

Parameters:

new_file_path (str, optional) – If provided, save to this path instead of overwriting the original.

Returns:

This method returns nothing.

Return type:

None

update_dataset_times(target_id: str, directory: str) → str

Update start/end times for a dataset using directory image dates.

Parameters:

• target_id (str) – Dataset identifier to match in the JSON payload.

• directory (str) – Directory to scan for frame timestamps.

Returns:

Status message describing the outcome of the update.

Return type:

str

zyra.utils.remove_all_files_in_directory(directory: str) → None

Remove all files and subdirectories under a directory.

Parameters:

directory (str) – Directory to clean.

Returns:

This function returns nothing.

Return type:

None

Notes

Errors are reported via logging.error for consistency with the rest of the codebase.

Modules

Credential storage and retrieval utilities.

This module provides CredentialManager, a small helper class to securely load, access, and manage credentials from a .env-style file without leaking them into the global process environment.

Examples

Load and access values:

from zyra.utils.credential_manager import CredentialManager

cm = CredentialManager("./.env")
cm.read_credentials(expected_keys=["API_KEY"])
token = cm.get_credential("API_KEY")

Use as a context manager:

with CredentialManager("./.env") as cm:
    cm.read_credentials()
    do_work(cm.get_credential("ACCESS_TOKEN"))

class zyra.utils.credential_manager.CredentialManager(filename: str | None = None, namespace: str | None = None)

Bases: object

Manage app credentials from a dotenv file.

Parameters:

• filename (str, optional) – Path to a dotenv file containing key=value pairs.

• namespace (str, optional) – Optional prefix to apply to all keys when stored/retrieved.

Examples

Namespaced keys:

cm = CredentialManager(".env", namespace="MYAPP_")
cm.read_credentials(expected_keys=["API_KEY"])  # expects MYAPP_API_KEY

add_credential(key: str, value: str) → None

Add or update a credential value in memory.

Parameters:

• key (str) – Base key name (namespace is applied automatically).

• value (str) – Credential value to store.

clear_credentials() → None

Remove all tracked credentials from memory.

delete_credential(key: str) → None

Delete a credential by key if present.

Parameters:

key (str) – Base key name (namespace is applied automatically).

get_credential(key: str) → str

Retrieve a credential value by key (with namespace applied).

Parameters:

key (str) – Base key name (namespace is applied automatically).

Returns:

Stored value for the namespaced key.

Return type:

str

Raises:

KeyError – If the key is not present in memory.

list_credentials(expected_keys: Iterable[str] | None = None) → list[str]

List tracked credential keys, checking for expected ones when provided.

Parameters:

expected_keys (Iterable[str], optional) – Keys to verify are present.

Returns:

Keys currently stored in memory.

Return type:

list of str

Raises:

KeyError – If any expected keys are missing.

read_credentials(expected_keys: Iterable[str] | None = None) → None

Read credentials from the dotenv file into memory.

Parameters:

expected_keys (Iterable[str], optional) – Keys that must be present; raises if any are missing.

Raises:

• FileNotFoundError – If the dotenv path cannot be resolved.

• KeyError – If any expected keys are missing after reading.

property tracked_keys: set[str]

Return the set of keys currently tracked in memory.

Date/time utilities for parsing, ranges, and frame calculations.

Provides DateManager for extracting timestamps from filenames, building date ranges from period specs (e.g., 1Y, 6M, 7D, 24H), and validating or interpolating time-based frame sequences.

Examples

Parse dates and compute a range:

from zyra.utils.date_manager import DateManager

dm = DateManager(["%Y%m%d"])
start, end = dm.get_date_range("7D")
ok = dm.is_date_in_range("frame_20240102.png", start, end)

class zyra.utils.date_manager.DateManager(date_formats: list[str] | None = None)

Bases: object

High-level utilities for working with dates and filenames.

Parameters:

date_formats (list of str, optional) – Preferred strftime-style formats to use when parsing dates from filenames (e.g., ["%Y%m%d"]).

Examples

Use a custom filename format first, then fall back to ISO-like detection:

dm = DateManager(["%Y%m%d%H%M%S"])
when = dm.extract_date_time("frame_20240101093000.png")

__init__(date_formats: list[str] | None = None) → None

Optionally store preferred date formats for filename parsing.

calculate_expected_frames(start_datetime: datetime, end_datetime: datetime, period_seconds: int) → int

Calculate expected frame count between two datetimes at a cadence.

Returns:

Number of expected frames (inclusive of endpoints).

Return type:

int

datetime_format_to_regex(datetime_format: str) → str

Convert a datetime format string to a regex pattern.

extract_date_time(string: str) → str | None

Extract a date string from a filename/text using known formats.

Tries known formats first; falls back to a simple ISO-like pattern.

extract_dates_from_filenames(directory_path: str, image_extensions: Iterable[str] = ('.jpg', '.jpeg', '.png', '.gif', '.bmp', '.dds')) → tuple[str | None, str | None]

Extract dates from the first and last image file names in a directory.

Parameters:

• directory_path (str) – Directory to scan for images.

• image_extensions (Iterable[str]) – File extensions to include when scanning.

Returns:

(first_date, last_date) as strings, or (None, None).

Return type:

tuple

find_missing_frames(directory, period_seconds, datetime_format, filename_format, filename_mask, start_datetime, end_datetime)

Find missing frames in a local directory with inconsistent period, only for image files.

find_missing_frames_and_predict_names(timestamps, period_seconds, filename_pattern)

Find gaps and overfrequent frames in timestamps and predict names.

find_start_end_datetimes(directory: str)

Find earliest and latest datetimes from filenames in a directory.

get_date_range(period: str) → tuple[datetime, datetime]

Compute a date range ending at the current minute from a period spec.

Parameters:

period (str) – Period string such as "1Y", "6M", "7D", or "24H".

Returns:

Start and end datetimes for the period ending at “now” (rounded to minute).

Return type:

(datetime, datetime)

get_date_range_iso(iso_duration: str) → tuple[datetime, datetime]

Compute a date range ending now from an ISO-8601 duration (e.g., P1Y, P6M, P7D, PT24H).

Supports a subset of ISO-8601: years (Y), months (M), days (D), hours (H) with the “P…T…” structure. Examples: “P1Y”, “P6M”, “P7D”, “PT24H”.

is_date_in_range(filepath: str, start_date: datetime, end_date: datetime) → bool

Check if a filename contains a date within a range.

Parameters:

• filepath (str) – Path or filename containing a date stamp.

• start_date (datetime) – Inclusive start of the permitted range.

• end_date (datetime) – Inclusive end of the permitted range.

Returns:

True if a parsed date falls within the range, else False.

Return type:

bool

parse_timestamps_from_filenames(filenames, datetime_format)

Parse timestamps from filenames based on the given format.

Filesystem helper utilities.

Lightweight helpers for common file and directory operations used across the project.

Examples

Clean a scratch directory:

from zyra.utils.file_utils import remove_all_files_in_directory

remove_all_files_in_directory("./scratch")

class zyra.utils.file_utils.FileUtils

Bases: object

Namespace for file-related helper routines.

Examples

While most functions are provided at module-level, a class instance can be created if you prefer an object to group related operations.

zyra.utils.file_utils.remove_all_files_in_directory(directory: str) → None

Remove all files and subdirectories under a directory.

Parameters:

directory (str) – Directory to clean.

Returns:

This function returns nothing.

Return type:

None

Notes

Errors are reported via logging.error for consistency with the rest of the codebase.

Read, update, and write JSON files used as simple configs/datasets.

Provides JSONFileManager to persist simple JSON structures and to update dataset start/end times using dates inferred from a directory of frames.

Examples

Update a dataset time window:

from zyra.utils.json_file_manager import JSONFileManager

jm = JSONFileManager("./config.json")
jm.update_dataset_times("my-dataset", "./frames")
jm.save_file()

class zyra.utils.json_file_manager.JSONFileManager(file_path: str)

Bases: object

Convenience wrapper for manipulating JSON files.

Parameters:

file_path (str) – Path to a JSON file on disk.

Examples

Read, update, and save:

jm = JSONFileManager("./data.json")
jm.data["foo"] = "bar"
jm.save_file()

read_file() → None

Read the JSON file from disk into memory.

Returns:

Populates self.data or sets it to None on error.

Return type:

None

save_file(new_file_path: str | None = None) → None

Write the in-memory data back to disk.

Parameters:

new_file_path (str, optional) – If provided, save to this path instead of overwriting the original.

Returns:

This method returns nothing.

Return type:

None

update_dataset_times(target_id: str, directory: str) → str

Update start/end times for a dataset using directory image dates.

Parameters:

• target_id (str) – Dataset identifier to match in the JSON payload.

• directory (str) – Directory to scan for frame timestamps.

Returns:

Status message describing the outcome of the update.

Return type:

str

zyra.utils.cli_helpers.configure_logging_from_env(default: str = 'info') → None

Set logging levels based on VERBOSITY env (supports ZYRA_*/DATAVIZHUB_*).

Values: debug|info|quiet. Defaults to ‘info’. - debug: root=DEBUG - info: root=INFO - quiet: root=ERROR (suppress most logs) Also dials down noisy third-party loggers (matplotlib, cartopy, botocore, requests).

zyra.utils.cli_helpers.detect_format_bytes(b: bytes) → str

Detect basic format from magic bytes.

Returns one of: "netcdf", "grib2", or "unknown".

zyra.utils.cli_helpers.is_grib2_bytes(b: bytes) → bool

Return True if bytes look like GRIB (b"GRIB").

zyra.utils.cli_helpers.is_netcdf_bytes(b: bytes) → bool

Return True if bytes look like NetCDF (classic CDF or HDF5-based).

Recognizes magic headers: - Classic NetCDF: b"CDF" - NetCDF4/HDF5: b"‰HDF"

zyra.utils.cli_helpers.parse_levels_arg(val) → int | list[float]

Parse levels from int or comma-separated floats.

zyra.utils.cli_helpers.read_all_bytes(path_or_dash: str) → bytes

Read all bytes from a path or ‘-’ (stdin).

zyra.utils.cli_helpers.temp_file_from_bytes(data: bytes, *, suffix: str = '') → Iterator[str]

Write bytes to a NamedTemporaryFile and yield its path; delete on exit.

zyra.utils.io_utils.open_input(path_or_dash: str) → BinaryIO

Return a readable binary file-like for path or ‘-’ (stdin).

zyra.utils.io_utils.open_output(path_or_dash: str) → BinaryIO

Return a writable binary file-like for path or ‘-’ (stdout).

zyra.utils.geo_utils.detect_crs_from_csv(df) → str | None

zyra.utils.geo_utils.detect_crs_from_path(path: str, *, var: str | None = None) → str | None

zyra.utils.geo_utils.detect_crs_from_xarray(ds) → str | None

zyra.utils.geo_utils.to_cartopy_crs(crs: str | None)

Return a Cartopy CRS object for an EPSG string, defaulting to PlateCarree.

Returns None if Cartopy is unavailable.

zyra.utils.geo_utils.warn_if_mismatch(input_crs: str | None, *, target_crs: str = 'EPSG:4326', reproject: bool = False, context: str = '') → None

GRIB utilities used by connectors and managers.

This module centralizes protocol-agnostic helpers for working with GRIB2 index files (.idx), calculating byte ranges, and performing parallel multi-range downloads.

Notes

• The .idx file path is assumed to be the GRIB file path with a .idx suffix appended, unless a path already ending in .idx is provided.

• Pattern filtering uses regular expressions via re.search().

zyra.utils.grib.compute_chunks(total_size: int, chunk_size: int = 524288000) → list[str]

Compute contiguous byte ranges that partition a file.

The final range uses the file size as the inclusive end byte (matching the behavior used by nodd_fetch.py).

Parameters:

• total_size (int) – Size of the file in bytes.

• chunk_size (int, default 500MB) – Upper bound for each chunk.

Returns:

Range header strings, e.g., ["bytes=0-1048575", ...].

Return type:

list of str

zyra.utils.grib.ensure_idx_path(path: str) → str

Return the .idx path for a GRIB file or pass through an explicit idx path.

Parameters:

path (str) – The GRIB file path or .idx path.

Returns:

path + '.idx' if path does not already end with .idx, otherwise returns path unchanged.

Return type:

str

zyra.utils.grib.idx_to_byteranges(lines: list[str], search_regex: str) → dict[str, str]

Convert .idx lines plus a variable regex into HTTP Range headers.

Parameters:

• lines (list of str) – Lines from a GRIB .idx file.

• search_regex (str) – Regular expression to select desired GRIB lines (e.g., “PRES:surface”).

Returns:

Mapping of {"bytes=start-end": matching_idx_line} suitable for use as Range headers.

Return type:

dict

zyra.utils.grib.parallel_download_byteranges(download_func: Callable[[str, str], bytes], key_or_url: str, byte_ranges: Iterable[str], *, max_workers: int = 10) → bytes

Download multiple byte ranges in parallel and concatenate in input order.

Parameters:

• download_func (Callable) – Function accepting (key_or_url, range_header) and returning bytes.

• key_or_url (str) – The resource identifier for the remote object.

• byte_ranges (Iterable[str]) – Iterable of Range header strings (e.g., “bytes=0-99”). Order matters and is preserved in the output concatenation.

• max_workers (int, default=10) – Maximum number of worker threads.

Returns:

The concatenated payload of all requested ranges in the input order.

Return type:

bytes

zyra.utils.grib.parse_idx_lines(idx_bytes_or_text: bytes | str) → list[str]

Parse a GRIB index payload into non-empty lines.

Parameters:

idx_bytes_or_text (bytes or str) – Raw .idx file content.

Returns:

The non-empty, newline-split lines of the index.

Return type:

list of str