"""Helpers for reading static catalog products."""
from __future__ import annotations
from pathlib import Path
from typing import Iterable, Mapping, Sequence
from .read_plan_executor import SourceReader, execute_read_plans
from .types import MissingStepPolicy
[docs]
def read_static_variables(
catalog,
product_key: str,
variables: Sequence[str],
*,
source_reader: SourceReader | None = None,
missing: str | MissingStepPolicy = MissingStepPolicy.RAISE,
logical_step: int = 0,
) -> Mapping[str, object]:
"""
Read variables from one static catalog product.
Parameters
----------
catalog : xgc_analysis.catalog.SimulationCatalog
Catalog containing ``product_key``.
product_key : str
Static product key such as ``"xgc.mesh.bp"`` or ``"xgc.f0.mesh.bp"``.
variables : sequence[str]
ADIOS variable names to read.
source_reader : callable or None, optional
Optional read backend with the standard ``SourceReader`` signature.
``None`` uses ``catalog.source_reader`` when present, otherwise the
FileReader-backed regular BP backend.
missing : {"raise", "skip", "zero"} or MissingStepPolicy, optional
Missing-variable or missing-step policy. ``"zero"`` is treated like
``"skip"`` here because static readers do not have enough context to
synthesize product-specific zero values.
logical_step : int, optional
Logical step used for static products. Directory-backed static
products normally expose logical step ``0``.
Returns
-------
dict[str, object]
Values keyed by variable name. Missing values are omitted when
``missing`` is ``"skip"`` or ``"zero"``.
Raises
------
KeyError
If the product or a required variable/step is unavailable.
"""
missing_policy = MissingStepPolicy(missing)
source_reader = source_reader or getattr(catalog, "source_reader", None)
product = catalog.get_product(product_key)
plans = []
for variable in variables:
if variable not in product.variables:
if missing_policy == MissingStepPolicy.RAISE:
raise KeyError(
f"Variable '{variable}' not found in catalog product '{product_key}'."
)
continue
plan_missing = (
MissingStepPolicy.SKIP
if missing_policy == MissingStepPolicy.ZERO
else missing_policy
)
plans.append(
catalog.plan_read(product_key, variable, [int(logical_step)], missing=plan_missing)
)
execution = execute_read_plans(plans, source_reader=source_reader)
return {record.variable: record.value for record in execution.records}
[docs]
def static_product_source_path(catalog, product_key: str) -> Path:
"""
Return the selected path for a static catalog product.
Parameters
----------
catalog : xgc_analysis.catalog.SimulationCatalog
Catalog containing ``product_key``.
product_key : str
Static product key to resolve.
Returns
-------
pathlib.Path
Path for the newest source advertised by the product.
Notes
-----
This is a compatibility helper for code paths that still require a local
BP path, such as lazy sparse-matrix wrappers. New readers should prefer
:func:`read_static_variables` so the same code can later use a
campaign-backed ``SourceReader``.
"""
product = catalog.get_product(product_key)
if not product.sources:
raise KeyError(f"Catalog product '{product_key}' has no sources.")
return max(product.sources, key=lambda source: (source.mtime, str(source.path))).path
[docs]
def has_static_product(catalog, product_key: str, variables: Iterable[str] | None = None) -> bool:
"""
Return whether ``catalog`` advertises a static product and optional variables.
Parameters
----------
catalog : xgc_analysis.catalog.SimulationCatalog
Catalog to inspect.
product_key : str
Product key to check.
variables : iterable[str] or None, optional
Variables that must be advertised by the product. If omitted, only the
product itself is checked.
"""
if catalog is None or product_key not in getattr(catalog, "products", {}):
return False
if variables is None:
return True
product = catalog.get_product(product_key)
return all(variable in product.variables for variable in variables)