"""
Directory-backed catalog for XGC ADIOS output discovery.
This module scans one dataset root, groups XGC ``*.bp`` outputs into logical
products, collects ADIOS metadata when requested, and builds read plans that map
user-facing logical steps to concrete ADIOS source/step pairs. It does not read
payload arrays; readers execute the returned ``ReadPlan`` objects.
"""
from __future__ import annotations
from collections import defaultdict
from dataclasses import replace
from pathlib import Path
import re
from typing import Callable, Dict, Iterable, List, Optional, Tuple
import numpy as np
from .types import (
CatalogArrayRead,
DataProduct,
MissingStepPolicy,
ProductLayout,
ProductType,
ReadFragment,
ReadPlan,
SourceInfo,
StepFragment,
StepInfo,
VariableInfo,
VariableMetadata,
)
from .run_manifest import (
TextArtifact,
build_run_manifest,
manifest_text_artifacts,
read_manifest_text,
)
from xgc_analysis.adios_file_reader import AdiosFileSourceReader
_BP_SEQ_RE = re.compile(r"^(?P<base>.+)\.(?P<index>\d+)\.bp$")
_ALL_STEPS_RANGE = range(0, 10**12)
_LOGICAL_STEP_VARIABLE_NAMES = ("gstep", "step", "timestep")
MetadataReader = Callable[[Path], tuple]
CatalogRefresh = Callable[[], Iterable[DataProduct]]
CatalogStateKey = Tuple[Tuple[str, str, int, int], ...]
CatalogStateReader = Callable[[], CatalogStateKey]
[docs]
class SimulationCatalog:
"""
Catalog of ADIOS-readable XGC products in one dataset root.
The catalog is intentionally independent of ``Simulation`` so GUI code and
automation scripts can inspect available products without constructing mesh
or field objects. A ``Simulation`` may hold a catalog instance when heavier
analysis context is needed.
"""
def __init__(
self,
root_dir: Path,
products: Iterable[DataProduct],
*,
refresh_func: Optional[CatalogRefresh] = None,
state_key_func: Optional[CatalogStateReader] = None,
state_key: Optional[CatalogStateKey] = None,
source_reader=None,
manifest: Optional[dict] = None,
text_reader=None,
):
"""
Store discovered products for one root directory.
Parameters
----------
root_dir : pathlib.Path
Dataset root used for discovery.
products : iterable of DataProduct
Product entries already discovered by a backend.
refresh_func : callable or None, optional
Backend hook that returns a freshly discovered product iterable.
Directory-backed catalogs install this hook automatically through
:func:`open_catalog`; hand-built catalogs may omit it.
state_key_func : callable or None, optional
Backend hook that returns a cheap signature for the current dataset
state. The directory backend uses BP path names, file types,
modification times, and sizes.
state_key : tuple or None, optional
Initial state signature. If omitted and ``state_key_func`` is
provided, the constructor computes the initial signature.
source_reader : callable or None, optional
Backend hook used by readers to execute catalog read plans. The
directory backend installs a FileReader source reader by default.
manifest : dict or None, optional
Optional run manifest attached by the backend. Directory-backed
catalogs use this to expose text artifacts and packaging metadata.
text_reader : callable or None, optional
Optional hook with signature ``reader(path) -> str`` used by
:meth:`read_text`.
"""
self.root_dir = Path(root_dir)
self.products: Dict[str, DataProduct] = {product.key: product for product in products}
self._refresh_func = refresh_func
self._state_key_func = state_key_func
self.state_key = state_key if state_key is not None else self._read_state_key()
self.source_reader = source_reader
self.manifest = manifest or {}
self.text_artifacts: Dict[str, TextArtifact] = {
artifact.path: artifact for artifact in manifest_text_artifacts(self.manifest)
}
self._text_reader = text_reader
[docs]
def close(self) -> None:
"""
Release backend resources held by this catalog.
Directory-backed catalogs normally do not hold open files. Campaign
catalogs may keep an ACA ``FileReader`` open so repeated read-plan
execution does not reopen an expensive archive. If the installed
``source_reader`` exposes ``close()``, this method delegates to it.
"""
close = getattr(self.source_reader, "close", None)
if close is not None:
close()
def __enter__(self):
"""Return this catalog for ``with`` statement use."""
return self
def __exit__(self, exc_type, exc, traceback) -> None:
"""Close backend resources when leaving a ``with`` block."""
self.close()
def __repr__(self) -> str:
"""Return a compact developer representation of this catalog."""
backend = _catalog_backend_name(self)
return (
f"SimulationCatalog(backend='{backend}', products={len(self.products)}, "
f"texts={len(self.text_artifacts)}, root='{self.root_dir}')"
)
def __str__(self) -> str:
"""Return a compact human-readable catalog overview."""
return self.overview()
[docs]
def refresh(self) -> "SimulationCatalog":
"""
Rebuild this catalog from its backend and update it in place.
Directory-backed catalogs created by :func:`open_catalog` rescan the
dataset root, regroup products, reread ADIOS metadata when configured to
do so, and replace ``self.products`` with the refreshed product map.
Existing :class:`ReadPlan` objects are not mutated; callers should
request new read plans after refreshing.
Returns
-------
SimulationCatalog
``self``, for convenient chaining.
Raises
------
RuntimeError
If the catalog was constructed without a refresh backend.
"""
if self._refresh_func is None:
raise RuntimeError("This catalog was constructed without a refresh backend.")
products = self._refresh_func()
self.products = {product.key: product for product in products}
self.state_key = self._read_state_key()
self._refresh_manifest()
return self
[docs]
def refresh_if_changed(self) -> bool:
"""
Refresh this catalog only if the backend state key has changed.
This is intended for polling running simulations from scripts or the
GUI. The directory backend keeps the state-key computation cheap by
stat-ing BP paths and their contents rather than opening ADIOS files.
When a change is detected, this method calls :meth:`refresh`.
Returns
-------
bool
True if the catalog was refreshed, False if the current state key
matches ``self.state_key``.
Raises
------
RuntimeError
If the catalog was constructed without a state-key backend.
"""
if self._state_key_func is None:
raise RuntimeError("This catalog was constructed without a state-key backend.")
current_state_key = self._read_state_key()
if current_state_key == self.state_key:
return False
self.refresh()
return True
def _read_state_key(self) -> Optional[CatalogStateKey]:
"""Return the backend state key, or ``None`` when no backend provides one."""
if self._state_key_func is None:
return None
return self._state_key_func()
[docs]
def list_products(self) -> List[DataProduct]:
"""
Return all discovered products ordered by product key.
Returns
-------
list[DataProduct]
Stable, alphabetically ordered product list for UI display or
scripted inspection.
"""
return [self.products[key] for key in sorted(self.products)]
[docs]
def get_product(self, product_key: str) -> DataProduct:
"""
Return one product entry by key.
Parameters
----------
product_key : str
Catalog key such as ``xgc.3d.bp`` or ``xgc.oneddiag.bp``.
Raises
------
KeyError
If the product is not present in the catalog.
"""
if product_key not in self.products:
raise KeyError(f"Unknown XGC product: {product_key}")
return self.products[product_key]
[docs]
def list_variables(self, product_key: str) -> List[VariableInfo]:
"""
Return variables available for one product.
Parameters
----------
product_key : str
Catalog key for the product to inspect.
Returns
-------
list[VariableInfo]
Variable metadata ordered by native variable name.
"""
product = self.get_product(product_key)
return [product.variables[key] for key in sorted(product.variables)]
[docs]
def product_names(self) -> List[str]:
"""
Return product keys as a short string list for interactive browsing.
Returns
-------
list[str]
Product keys ordered alphabetically. This is the name-only
companion to :meth:`list_products`, which returns full
:class:`DataProduct` records for programmatic use.
"""
return sorted(self.products)
[docs]
def variable_names(self, product_key: str) -> List[str]:
"""
Return variable names for one product as a short string list.
Parameters
----------
product_key : str
Catalog product key such as ``"xgc.2d.bp"``.
Returns
-------
list[str]
Native variable names ordered alphabetically.
"""
return sorted(self.get_product(product_key).variables)
[docs]
def list_texts(self) -> List[TextArtifact]:
"""
Return text artifacts discovered for this catalog.
Returns
-------
list[TextArtifact]
Manifest text records ordered by relative path. Directory-backed
catalogs include standard XGC input/config files, recursively found
files referenced by ``input``, and detected XGC stdout logs.
Campaign-backed catalogs may return an empty list until the
campaign packaging schema is fully standardized.
"""
return [self.text_artifacts[key] for key in sorted(self.text_artifacts)]
[docs]
def text_names(self) -> List[str]:
"""
Return catalog text artifact paths as a short string list.
Returns
-------
list[str]
Manifest-relative text artifact paths ordered alphabetically.
"""
return sorted(self.text_artifacts)
[docs]
def has_text(self, artifact_path: str) -> bool:
"""
Return whether a text artifact is known to the catalog.
Parameters
----------
artifact_path : str
Manifest-relative artifact path, for example ``"input"``.
"""
return artifact_path in self.text_artifacts
[docs]
def read_text(self, artifact_path: str) -> str:
"""
Read one catalog text artifact.
Parameters
----------
artifact_path : str
Manifest-relative artifact path.
Returns
-------
str
Text content decoded by the backend.
Raises
------
KeyError
If the artifact is not present in the catalog manifest.
RuntimeError
If the backend did not install a text reader.
"""
if artifact_path not in self.text_artifacts:
raise KeyError(f"Unknown catalog text artifact: {artifact_path}")
if self._text_reader is None:
raise RuntimeError("This catalog backend does not provide text reads.")
return self._text_reader(artifact_path)
[docs]
def available_steps(self, product_key: str, variable: Optional[str] = None) -> List[StepInfo]:
"""
Return deduplicated logical steps for one product and optional variable.
Parameters
----------
product_key : str
Catalog product key.
variable : str or None, optional
If provided, only sources that advertise the variable contribute
steps. If omitted, all sources for the product contribute.
Returns
-------
list[StepInfo]
Logical steps in ascending order. When several sources advertise
the same logical step, the selected fragment follows the
``newer_source_wins`` policy and other candidates are retained as
duplicates.
"""
product = self.get_product(product_key)
candidates: Dict[int, List[StepFragment]] = defaultdict(list)
for source in product.sources:
if variable is not None and variable not in source.variables:
continue
for logical_step, adios_step, time_value in _iter_source_steps(source):
candidates[int(logical_step)].append(
StepFragment(
source_id=source.source_id,
source_path=source.path,
logical_step=int(logical_step),
adios_step=int(adios_step),
file_index=source.file_index,
time=time_value,
)
)
out = []
for logical_step in sorted(candidates):
fragments = candidates[logical_step]
selected = max(fragments, key=_fragment_newness_key)
duplicates = [
StepFragment(
source_id=fragment.source_id,
source_path=fragment.source_path,
logical_step=fragment.logical_step,
adios_step=fragment.adios_step,
file_index=fragment.file_index,
time=fragment.time,
selected=False,
)
for fragment in fragments
if fragment != selected
]
out.append(
StepInfo(
logical_step=logical_step,
selected_fragment=selected,
duplicate_fragments=duplicates,
)
)
return out
[docs]
def plan_read(
self,
product_key: str,
variable: str,
steps: Iterable[int],
missing: str | MissingStepPolicy = MissingStepPolicy.RAISE,
) -> ReadPlan:
"""
Resolve a variable read over logical steps into ADIOS source fragments.
Parameters
----------
product_key : str
Catalog product key.
variable : str
Native XGC/ADIOS variable name to read.
steps : iterable[int]
Requested logical step values, typically XGC ``gstep`` values or
filename-derived indices when no explicit step variable exists.
missing : {"raise", "skip", "zero"} or MissingStepPolicy, optional
Policy for requested logical steps that are not available.
Returns
-------
ReadPlan
Source fragments grouped by ADIOS-readable source.
Raises
------
KeyError
If a requested step is unavailable and ``missing`` is ``"raise"``.
ValueError
If ``missing`` is not a supported missing-step policy.
"""
missing_policy = MissingStepPolicy(missing)
requested_steps = [int(step) for step in steps]
step_map = {info.logical_step: info for info in self.available_steps(product_key, variable)}
missing_steps = [step for step in requested_steps if step not in step_map]
if missing_steps and missing_policy == MissingStepPolicy.RAISE:
missing_str = ", ".join(str(step) for step in missing_steps)
raise KeyError(f"Missing logical step(s) for {product_key}:{variable}: {missing_str}")
grouped: Dict[Tuple[str, Path, Optional[int]], List[Tuple[int, int]]] = defaultdict(list)
for step in requested_steps:
if step not in step_map:
continue
selected = step_map[step].selected_fragment
grouped[(selected.source_id, selected.source_path, selected.file_index)].append((step, selected.adios_step))
fragments = [
ReadFragment(
source_id=source_id,
source_path=source_path,
variable=variable,
logical_steps=[item[0] for item in items],
adios_steps=[item[1] for item in items],
file_index=file_index,
)
for (source_id, source_path, file_index), items in grouped.items()
]
return ReadPlan(
product_key=product_key,
variable=variable,
requested_steps=requested_steps,
fragments=fragments,
missing_steps=missing_steps,
missing_policy=missing_policy,
)
[docs]
def read_arrays(
self,
product_key: str,
variables: Iterable[str] | str | None = None,
steps: Iterable[int] | int | None = None,
*,
missing: str | MissingStepPolicy = MissingStepPolicy.RAISE,
source_reader=None,
) -> CatalogArrayRead:
"""
Read catalog variables into plain NumPy arrays.
Parameters
----------
product_key : str
Catalog product key such as ``"xgc.2d.bp"``.
variables : iterable[str], str, or None, optional
Variables to read. If omitted, all variables advertised by the
product are read.
steps : iterable[int], int, or None, optional
Logical XGC steps to read. If omitted, each variable is read over
the logical steps available for that variable.
missing : {"raise", "skip", "zero"} or MissingStepPolicy, optional
Missing-step policy for explicit ``steps``. ``"zero"`` records
missing steps but does not synthesize zero arrays.
source_reader : callable or None, optional
Optional read-plan source backend. If omitted,
``self.source_reader`` is used when present, otherwise the regular
FileReader-backed local BP backend is used.
Returns
-------
CatalogArrayRead
Variable-first array mapping plus read provenance.
"""
product = self.get_product(product_key)
variable_list = _normalize_read_array_variables(product, variables)
source_reader = source_reader or self.source_reader
plans = []
for variable in variable_list:
if steps is None:
variable_steps = [info.logical_step for info in self.available_steps(product_key, variable)]
else:
variable_steps = _normalize_step_selector(steps)
plans.append(self.plan_read(product_key, variable, variable_steps, missing=missing))
from .read_plan_executor import execute_read_plans # pylint: disable=import-outside-toplevel
execution = execute_read_plans(plans, source_reader=source_reader)
arrays: Dict[str, Dict[int, np.ndarray]] = defaultdict(dict)
for record in execution.records:
arrays[record.variable][record.logical_step] = np.asarray(record.value)
return CatalogArrayRead(
arrays={variable: dict(step_values) for variable, step_values in arrays.items()},
records=list(execution.records),
missing_steps_by_variable=dict(execution.missing_steps_by_variable),
)
[docs]
def to_summary(
self,
*,
include_variables: bool = True,
include_sources: bool = False,
include_texts: bool = True,
include_provenance: bool = False,
max_products: int | None = None,
max_variables: int | None = None,
max_steps: int = 20,
max_sources: int = 20,
) -> dict:
"""
Return a stable JSON-like summary of catalog contents.
Parameters
----------
include_variables : bool, optional
Include per-variable dtype, shape, units, description, and
coordinate coverage.
include_sources : bool, optional
Include source ids and source-level step counts. Full paths and
modification times are omitted unless ``include_provenance`` is
True.
include_texts : bool, optional
Include catalog text artifacts.
include_provenance : bool, optional
Include backend/source provenance such as campaign path, source
paths, modification times, file indices, and ADIOS-step counts.
max_products : int or None, optional
Limit number of products in the returned summary. ``None`` means
include all products.
max_variables : int or None, optional
Limit number of variables per product. ``None`` means include all
variables.
max_steps : int, optional
Maximum number of coordinate samples to include for ``gstep``,
``tindex``, and ``time``.
max_sources : int, optional
Maximum number of source records to include per product.
Returns
-------
dict
Plain JSON-like data with schema name
``"xgc-catalog-summary-v1"``. The method intentionally returns
built-in containers rather than catalog dataclasses so loosely
coupled tools can inspect catalog contents without importing the
internal object model.
"""
product_keys = self.product_names()
selected_product_keys = _limit_sequence(product_keys, max_products)
products = {}
for product_key in selected_product_keys:
product = self.get_product(product_key)
product_record = {
"type": product.product_type.value,
"family": product.product_family,
"layout": product.layout.value,
"source_count": len(product.sources),
"variable_count": len(product.variables),
"coordinates": _coordinate_summary(
_selected_coordinate_records(product),
max_samples=max_steps,
),
}
if include_variables:
variable_keys = _limit_sequence(sorted(product.variables), max_variables)
variables = {}
for variable_name in variable_keys:
info = product.variables[variable_name]
variables[variable_name] = _variable_summary(
info,
_coordinate_summary(
_selected_coordinate_records(product, variable_name),
max_samples=max_steps,
),
)
product_record["variables"] = variables
product_record["variables_truncated"] = max(0, len(product.variables) - len(variable_keys))
if include_sources or include_provenance:
sources = _limit_sequence(product.sources, max_sources)
product_record["sources"] = [
_source_summary(source, include_provenance=include_provenance)
for source in sources
]
product_record["sources_truncated"] = max(0, len(product.sources) - len(sources))
products[product_key] = product_record
out = {
"schema": "xgc-catalog-summary-v1",
"backend": _catalog_backend_name(self),
"root": str(self.root_dir),
"product_count": len(self.products),
"products_truncated": max(0, len(product_keys) - len(selected_product_keys)),
"products": products,
}
if include_texts:
out["text_count"] = len(self.text_artifacts)
out["texts"] = {
artifact.path: {
"class": artifact.artifact_class,
"role": artifact.role,
"embedded": artifact.embedded,
"source": artifact.source,
}
for artifact in self.list_texts()
}
if include_provenance:
out["provenance"] = _catalog_provenance(self)
return out
[docs]
def overview(
self,
*,
max_products: int | None = None,
max_vars: int = 0,
max_steps: int = 5,
) -> str:
"""
Return a compact human-readable overview of catalog contents.
Parameters
----------
max_products : int or None, optional
Maximum number of products to show. ``None`` means show all product
rows.
max_vars : int, optional
Maximum number of variable rows to show below each product. The
default ``0`` keeps the overview at product level.
max_steps : int, optional
Maximum number of coordinate samples to show when formatting step
coverage.
Returns
-------
str
Multi-line text intended for ``print(catalog)`` and interactive
inspection. Use :meth:`to_summary` for machine-readable output.
"""
lines = [
f"XGC catalog ({_catalog_backend_name(self)})",
f"root: {self.root_dir}",
f"products: {len(self.products)}, text artifacts: {len(self.text_artifacts)}",
]
product_keys = self.product_names()
selected_product_keys = _limit_sequence(product_keys, max_products)
for product_key in selected_product_keys:
product = self.get_product(product_key)
coords = _coordinate_summary(
_selected_coordinate_records(product),
max_samples=max_steps,
)
step_text = _format_coordinate_summary(coords)
lines.append(
f"{product_key:<28} {product.product_type.value:<22} "
f"{product.layout.value:<15} vars={len(product.variables):<4} "
f"sources={len(product.sources):<3} {step_text}"
)
for variable_name in _limit_sequence(sorted(product.variables), max_vars):
info = product.variables[variable_name]
lines.append(f" {_format_variable_overview(variable_name, info)}")
if max_vars is not None and len(product.variables) > max_vars:
lines.append(f" ... {len(product.variables) - max_vars} more variables")
if len(product_keys) > len(selected_product_keys):
lines.append(f"... {len(product_keys) - len(selected_product_keys)} more products")
return "\n".join(lines)
[docs]
def describe(self, name: str, *, max_steps: int = 10, max_vars: int = 50) -> str:
"""
Return focused human-readable details for a product or variable.
Parameters
----------
name : str
Product key, variable name, or explicit ``"product:variable"``
selector. Variable-only selectors search all products and show all
matches.
max_steps : int, optional
Maximum number of ``gstep``, ``tindex``, and ``time`` samples to
show.
max_vars : int, optional
Maximum number of variables to show when describing a product.
Returns
-------
str
Multi-line description suitable for interactive use.
Raises
------
KeyError
If ``name`` does not match a product or variable.
"""
if ":" in name:
product_key, variable_name = name.split(":", 1)
product = self.get_product(product_key)
if variable_name not in product.variables:
raise KeyError(f"Variable '{variable_name}' not found in product '{product_key}'.")
return self._describe_variable(product, variable_name, max_steps=max_steps)
if name in self.products:
return self._describe_product(self.get_product(name), max_steps=max_steps, max_vars=max_vars)
matches = [
(product, name)
for product in self.list_products()
if name in product.variables
]
if not matches:
raise KeyError(f"No catalog product or variable matches '{name}'.")
blocks = [
self._describe_variable(product, variable_name, max_steps=max_steps)
for product, variable_name in matches
]
return "\n\n".join(blocks)
def _describe_product(self, product: DataProduct, *, max_steps: int, max_vars: int) -> str:
"""
Return human-readable details for one product.
Parameters
----------
product : DataProduct
Product to describe.
max_steps : int
Maximum number of coordinate samples to show.
max_vars : int
Maximum number of variables to show.
"""
coords = _coordinate_summary(
_selected_coordinate_records(product),
max_samples=max_steps,
)
lines = [
f"Product: {product.key}",
f"type: {product.product_type.value}",
f"family: {product.product_family}",
f"layout: {product.layout.value}",
f"sources: {len(product.sources)}",
f"variables: {len(product.variables)}",
f"coordinates: {_format_coordinate_summary(coords)}",
]
for variable_name in _limit_sequence(sorted(product.variables), max_vars):
lines.append(f" {_format_variable_overview(variable_name, product.variables[variable_name])}")
if len(product.variables) > max_vars:
lines.append(f" ... {len(product.variables) - max_vars} more variables")
return "\n".join(lines)
def _describe_variable(self, product: DataProduct, variable_name: str, *, max_steps: int) -> str:
"""
Return human-readable details for one product variable.
Parameters
----------
product : DataProduct
Product containing the variable.
variable_name : str
Variable to describe.
max_steps : int
Maximum number of coordinate samples to show.
"""
info = product.variables[variable_name]
metadata = info.metadata
coords = _coordinate_summary(
_selected_coordinate_records(product, variable_name),
max_samples=max_steps,
)
lines = [
f"Variable: {product.key}:{variable_name}",
f"dtype: {info.dtype}",
f"shape: {info.shape_dims}",
f"raw_shape: {info.shape}",
f"steps: {info.step_count}",
f"units: {metadata.units}",
f"description: {metadata.description}",
f"coordinates: {_format_coordinate_summary(coords)}",
]
if metadata.axes:
lines.append(f"axes: {metadata.axes}")
return "\n".join(lines)
def _refresh_manifest(self) -> None:
"""Refresh attached manifest-derived text artifact records if available."""
if not self.manifest:
return
if hasattr(self, "campaign_path"):
return
try:
self.manifest = build_run_manifest(self.root_dir)
except Exception:
return
self.text_artifacts = {
artifact.path: artifact for artifact in manifest_text_artifacts(self.manifest)
}
def _catalog_backend_name(catalog: SimulationCatalog) -> str:
"""
Return the backend label used by public catalog summaries.
Parameters
----------
catalog : SimulationCatalog
Catalog to identify.
"""
if hasattr(catalog, "campaign_path"):
return "campaign"
if getattr(catalog, "_refresh_func", None) is not None:
return "directory"
return "unknown"
def _catalog_provenance(catalog: SimulationCatalog) -> dict:
"""
Return backend-level provenance for machine-readable summaries.
Parameters
----------
catalog : SimulationCatalog
Catalog to summarize.
"""
out = {
"backend": _catalog_backend_name(catalog),
"root": str(catalog.root_dir),
}
campaign_path = getattr(catalog, "campaign_path", None)
if campaign_path is not None:
out["campaign_path"] = str(campaign_path)
if catalog.state_key is not None:
out["state_key_records"] = len(catalog.state_key)
return out
def _limit_sequence(items, limit: int | None):
"""
Return ``items`` as a list truncated to ``limit`` entries.
Parameters
----------
items : iterable
Values to list and possibly truncate.
limit : int or None
Maximum number of items. ``None`` means no limit.
"""
item_list = list(items)
if limit is None:
return item_list
return item_list[:max(0, int(limit))]
def _variable_summary(info: VariableInfo, coordinates: dict) -> dict:
"""
Return a JSON-like summary for one catalog variable.
Parameters
----------
info : VariableInfo
Variable metadata record.
coordinates : dict
Coordinate coverage summary for this variable.
"""
metadata = info.metadata
return {
"dtype": info.dtype,
"shape": list(info.shape_dims),
"raw_shape": info.shape,
"is_scalar": bool(info.is_scalar),
"step_count": int(info.step_count),
"units": metadata.units,
"description": metadata.description,
"axes": list(metadata.axes),
"mesh_context": metadata.mesh_context,
"coordinate_context": metadata.coordinate_context,
"species_context": metadata.species_context,
"time_context": metadata.time_context,
"centering": metadata.centering,
"normalization": metadata.normalization,
"coordinates": coordinates,
}
def _source_summary(source: SourceInfo, *, include_provenance: bool) -> dict:
"""
Return a JSON-like summary for one physical/catalog source.
Parameters
----------
source : SourceInfo
Source metadata record.
include_provenance : bool
Include source path and modification time when True.
"""
out = {
"source_id": source.source_id,
"file_index": source.file_index,
"adios_step_count": int(source.adios_step_count),
"variable_count": len(source.variables),
}
if include_provenance:
out["path"] = str(source.path)
out["mtime"] = float(source.mtime)
return out
def _selected_coordinate_records(product: DataProduct, variable: str | None = None) -> List[dict]:
"""
Return deduplicated coordinate records for a product and optional variable.
Parameters
----------
product : DataProduct
Product whose source coordinates should be summarized.
variable : str or None, optional
If provided, only sources advertising the variable contribute records.
"""
candidates: Dict[int, List[dict]] = defaultdict(list)
for source in product.sources:
if variable is not None and variable not in source.variables:
continue
for logical_step, adios_step, time_value in _iter_source_steps(source):
tindex_value = (
source.tindex_values[adios_step]
if adios_step < len(source.tindex_values)
else None
)
candidates[int(logical_step)].append(
{
"logical_step": int(logical_step),
"adios_step": int(adios_step),
"time": time_value,
"tindex": tindex_value,
"source": source,
}
)
selected = []
for logical_step in sorted(candidates):
selected.append(
max(
candidates[logical_step],
key=lambda record: (_source_newness_key(record["source"]), record["adios_step"]),
)
)
return selected
def _coordinate_summary(records: List[dict], *, max_samples: int) -> dict:
"""
Return compact ``gstep``, ``tindex``, and ``time`` coverage summaries.
Parameters
----------
records : list[dict]
Deduplicated coordinate records from :func:`_selected_coordinate_records`.
max_samples : int
Maximum number of sample values per coordinate.
"""
out = {}
logical_steps = [record["logical_step"] for record in records]
out["gstep"] = _axis_summary(logical_steps, max_samples=max_samples)
tindex_values = [record["tindex"] for record in records if record["tindex"] is not None]
if tindex_values:
out["tindex"] = _axis_summary(tindex_values, max_samples=max_samples)
time_values = [record["time"] for record in records if record["time"] is not None]
if time_values:
out["time"] = _axis_summary(time_values, max_samples=max_samples)
return out
def _axis_summary(values, *, max_samples: int) -> dict:
"""
Return count/range/sample summary for one coordinate axis.
Parameters
----------
values : iterable
Coordinate values in catalog order.
max_samples : int
Maximum number of leading samples to include.
"""
value_list = list(values)
if not value_list:
return {"count": 0, "first": None, "last": None, "sample": []}
sample = value_list[:max(0, int(max_samples))]
return {
"count": len(value_list),
"first": value_list[0],
"last": value_list[-1],
"sample": sample,
}
def _format_coordinate_summary(coordinates: dict) -> str:
"""
Format coordinate coverage for human-readable summaries.
Parameters
----------
coordinates : dict
Coordinate summary returned by :func:`_coordinate_summary`.
"""
parts = []
for name in ("gstep", "tindex", "time"):
axis = coordinates.get(name)
if not axis or axis.get("count", 0) == 0:
continue
parts.append(
f"{name}={axis.get('first')}..{axis.get('last')} "
f"({axis.get('count')})"
)
return ", ".join(parts) if parts else "steps=0"
def _format_variable_overview(name: str, info: VariableInfo) -> str:
"""
Format one compact variable row for human-readable output.
Parameters
----------
name : str
Variable name.
info : VariableInfo
Variable metadata record.
"""
units = f" [{info.metadata.units}]" if info.metadata.units else ""
return (
f"{name:<24} {info.dtype:<10} shape={info.shape_dims} "
f"steps={info.step_count}{units}"
)
[docs]
def open_catalog(
root_dir: str | Path,
*,
collect_metadata: bool = True,
metadata_reader: Optional[MetadataReader] = None,
) -> SimulationCatalog:
"""
Discover XGC products in a directory and return a catalog.
Parameters
----------
root_dir : str or pathlib.Path
Directory containing XGC output files.
collect_metadata : bool, optional
If True, open each ADIOS source to collect variable and step metadata.
If False, only filenames, product grouping, and basic source metadata
are collected.
metadata_reader : callable or None, optional
Test or backend hook with signature ``reader(path) -> (variables,
step_values, time_values, adios_step_count)``.
Returns
-------
SimulationCatalog
Directory-backed catalog for ``root_dir``. Missing or non-directory
roots produce an empty catalog.
"""
root = Path(root_dir).expanduser().resolve()
products = discover_directory_products(
root,
collect_metadata=collect_metadata,
metadata_reader=metadata_reader,
)
manifest = build_run_manifest(root)
return SimulationCatalog(
root,
products,
refresh_func=lambda: discover_directory_products(
root,
collect_metadata=collect_metadata,
metadata_reader=metadata_reader,
),
state_key_func=lambda: directory_state_key(root),
state_key=directory_state_key(root),
source_reader=AdiosFileSourceReader(),
manifest=manifest,
text_reader=lambda artifact_path: read_manifest_text(root, artifact_path),
)
def _normalize_read_array_variables(product: DataProduct, variables) -> List[str]:
"""
Normalize a catalog ``read_arrays`` variable selector.
Parameters
----------
product : DataProduct
Product whose variable namespace should be used for validation.
variables : iterable[str], str, or None
User variable selector.
"""
if variables is None:
return sorted(product.variables)
if isinstance(variables, str):
variable_list = [variables]
else:
variable_list = list(variables)
missing = [variable for variable in variable_list if variable not in product.variables]
if missing:
raise KeyError(f"Variable(s) not found in catalog product '{product.key}': {', '.join(missing)}")
return variable_list
def _normalize_step_selector(steps) -> List[int]:
"""
Normalize an integer or iterable step selector to a list of integers.
Parameters
----------
steps : iterable[int] or int
Logical step selector.
"""
if isinstance(steps, int):
return [int(steps)]
return [int(step) for step in steps]
[docs]
def directory_state_key(root_dir: str | Path) -> CatalogStateKey:
"""
Return a cheap filesystem signature for directory-backed catalog contents.
The signature covers top-level ``*.bp`` files/directories, all entries
beneath BP directories, and manifest-discovered text/input artifacts. It
intentionally uses only relative path, path type, modification time, and
size, so it can be used for polling without opening ADIOS metadata. A
changed key means callers should call :meth:`SimulationCatalog.refresh`; an
unchanged key means the catalog can usually be reused.
Parameters
----------
root_dir : str or pathlib.Path
Dataset root scanned by the directory backend.
Returns
-------
tuple
Stable, sorted records ``(relative_path, kind, mtime_ns, size)``. The
tuple is empty when the root does not exist or is not a directory.
"""
root = Path(root_dir).expanduser().resolve()
if not root.exists() or not root.is_dir():
return ()
records = []
for path in sorted(root.iterdir(), key=lambda item: item.name):
if path.name.endswith(".bp"):
records.extend(_path_state_records(root, path))
records.extend(_manifest_artifact_state_records(root))
return tuple(records)
def _manifest_artifact_state_records(root: Path) -> List[Tuple[str, str, int, int]]:
"""
Return state-key records for manifest-discovered non-BP artifacts.
Parameters
----------
root : pathlib.Path
Dataset root.
"""
try:
manifest = build_run_manifest(root)
except Exception:
return []
artifact_paths = set()
for artifact in manifest.get("text_artifacts", []):
artifact_paths.add(str(artifact.get("path", "")))
for artifact in manifest.get("input_artifacts", []):
artifact_paths.add(str(artifact.get("path", "")))
records = []
for artifact_path in sorted(path for path in artifact_paths if path):
path = root / artifact_path
if path.name.endswith(".bp"):
continue
records.extend(_path_state_records(root, path))
return records
def _path_state_records(root: Path, path: Path) -> List[Tuple[str, str, int, int]]:
"""
Return state-key records for one BP file or directory tree.
Parameters
----------
root : pathlib.Path
Dataset root used to make returned paths relative and stable.
path : pathlib.Path
Top-level BP file or BP directory.
Returns
-------
list[tuple[str, str, int, int]]
Records containing relative POSIX path, path kind (``"dir"`` or
``"file"``), nanosecond modification time, and size. Paths that
disappear during polling are skipped so concurrently updated outputs do
not make the catalog refresh check fail.
"""
paths = [path]
if path.is_dir():
try:
paths.extend(sorted(path.rglob("*"), key=lambda item: item.relative_to(root).as_posix()))
except OSError:
pass
records = []
for item in paths:
try:
stat = item.stat()
except OSError:
continue
kind = "dir" if item.is_dir() else "file"
records.append((item.relative_to(root).as_posix(), kind, int(stat.st_mtime_ns), int(stat.st_size)))
return records
[docs]
def discover_directory_products(
root_dir: Path,
*,
collect_metadata: bool = True,
metadata_reader: Optional[MetadataReader] = None,
) -> List[DataProduct]:
"""
Discover BP products directly contained in ``root_dir``.
Files are grouped using XGC naming conventions. A sequence such as
``xgc.3d.00010.bp`` and ``xgc.3d.00012.bp`` becomes one product keyed as
``xgc.3d.bp`` with two sources. Non-sequence files such as
``xgc.oneddiag.bp`` become single-source products.
Parameters
----------
root_dir : pathlib.Path
Directory to scan. The scan is non-recursive.
collect_metadata : bool, optional
Whether to open ADIOS sources and collect variables/steps.
metadata_reader : callable or None, optional
Optional metadata reader override, mainly for tests.
Returns
-------
list[DataProduct]
Discovered products sorted by product key.
"""
if not root_dir.exists() or not root_dir.is_dir():
return []
entries = [path for path in root_dir.iterdir() if path.name.endswith(".bp")]
grouped: Dict[str, List[Tuple[Optional[int], Path]]] = defaultdict(list)
for path in entries:
match = _BP_SEQ_RE.match(path.name)
if match:
grouped[f"{match.group('base')}.bp"].append((int(match.group("index")), path))
else:
grouped[path.name].append((None, path))
read_metadata = metadata_reader or _read_adios_metadata
products = []
for key in sorted(grouped):
source_items = sorted(grouped[key], key=lambda item: (-1 if item[0] is None else item[0], item[1].name))
sources = [
_build_source_info(
path,
file_index=file_index,
collect_metadata=collect_metadata,
metadata_reader=read_metadata,
)
for file_index, path in source_items
]
product_type, family = classify_product(key)
layout = _layout_for_sources(sources)
variables = _merge_variables(sources)
products.append(
DataProduct(
key=key,
label=key,
product_type=product_type,
layout=layout,
sources=sources,
variables=variables,
product_family=family,
)
)
return products
[docs]
def classify_product(product_key: str) -> Tuple[ProductType, str]:
"""
Classify an XGC product key by established filename conventions.
Parameters
----------
product_key : str
Logical product key after sequence grouping, for example
``xgc.3d.bp`` rather than ``xgc.3d.00010.bp``.
Returns
-------
tuple[ProductType, str]
Product type and coarse product family. Unknown products return
``(ProductType.UNKNOWN, "")`` so callers can still inspect them.
"""
if product_key == "xgc.mesh.bp":
return ProductType.MESH_GEOMETRY, "mesh"
if product_key == "xgc.equil.bp":
return ProductType.EQUILIBRIUM, "mesh"
if product_key in {"xgc.bfield.bp", "xgc.bfieldm.bp", "xgc.current_drive.bp"}:
return ProductType.MAGNETIC_FIELD, "mesh"
if product_key in {
"xgc.f0.mesh.bp",
"xgc.volumes.bp",
"xgc.saved_volumes.bp",
"xgc.grad_rz.bp",
"xgc.smooth_pol.bp",
"xgc.hyp_vis_rad.bp",
"xgc.cnv_to_surf.bp",
"xgc.cnv_from_surf.bp",
"xgc.ff_1dp_fwd.bp",
"xgc.ff_1dp_rev.bp",
"xgc.ff_hdp_fwd.bp",
"xgc.ff_hdp_rev.bp",
}:
return ProductType.MESH_GEOMETRY, "mesh"
if product_key in {"xgc.2d.bp", "xgc.loop_vol.bp"}:
return ProductType.FIELD_2D, "field"
if product_key == "xgc.3d.bp":
return ProductType.FIELD_3D, "field"
if product_key == "xgc.f0.bp":
return ProductType.DISTRIBUTION_FUNCTION, "distribution"
if product_key == "xgc.f2d.bp":
return ProductType.FMOMENT_2D, "moment"
if product_key == "xgc.f3d.bp":
return ProductType.FMOMENT_3D, "moment"
if product_key == "xgc.heatdiag2.bp":
return ProductType.HEAT_DIAG, "diagnostic"
if product_key == "xgc.oneddiag.bp":
return ProductType.ONE_D_DIAG, "diagnostic"
if product_key == "xgc.neutrals.bp":
return ProductType.NEUTRAL_DIAG, "diagnostic"
if product_key == "xgc.fsourcediag.bp":
return ProductType.FSOURCE_DIAG, "diagnostic"
if product_key == "xgc.sheathdiag.bp":
return ProductType.SHEATH_DIAG, "diagnostic"
if product_key == "xgc.diffusion_coeff.bp":
return ProductType.DIFFUSION_COEFFICIENTS, "workflow"
if product_key == "xgc.diffusion_profiles.bp":
return ProductType.DIFFUSION_PROFILES, "workflow"
if product_key == "xgc.units.bp":
return ProductType.ANALYSIS, "analysis"
return ProductType.UNKNOWN, ""
def _build_source_info(
path: Path,
*,
file_index: Optional[int],
collect_metadata: bool,
metadata_reader: MetadataReader,
) -> SourceInfo:
"""
Build source metadata for one physical ADIOS-readable file.
Parameters
----------
path : pathlib.Path
Source path.
file_index : int or None
Sequence index parsed from the filename, if present.
collect_metadata : bool
Whether to call ``metadata_reader``.
metadata_reader : callable
Function used to collect variables, step values, time values, and ADIOS
step count.
Returns
-------
SourceInfo
Source record. Metadata-read failures are intentionally downgraded to
empty metadata so filename discovery remains usable on incomplete or
temporarily unreadable output.
"""
try:
mtime = path.stat().st_mtime
except OSError:
mtime = 0.0
variables: Dict[str, VariableInfo] = {}
step_values: List[int] = []
time_values: List[float] = []
tindex_values: List[int] = []
adios_step_count = 1
if collect_metadata:
try:
metadata = metadata_reader(path)
if len(metadata) == 4:
variables, step_values, time_values, adios_step_count = metadata
else:
variables, step_values, time_values, tindex_values, adios_step_count = metadata
except Exception:
variables = {}
step_values = []
time_values = []
tindex_values = []
adios_step_count = 1
return SourceInfo(
source_id=path.name,
path=path,
file_index=file_index,
mtime=mtime,
variables=variables,
step_values=step_values,
time_values=time_values,
tindex_values=tindex_values,
adios_step_count=max(1, int(adios_step_count or 1)),
)
def _read_adios_metadata(path: Path) -> Tuple[Dict[str, VariableInfo], List[int], List[float], List[int], int]:
"""
Read lightweight ADIOS metadata from one BP or ACA source.
Parameters
----------
path : pathlib.Path
ADIOS-readable BP source or HPC-Campaign ACA file.
Returns
-------
tuple
``(variables, step_values, time_values, tindex_values, adios_step_count)`` where
``variables`` maps native variable names to ``VariableInfo`` objects,
``step_values`` comes from ``gstep``, ``step``, or ``timestep`` when readable,
``time_values`` comes from ``time`` when readable,
``tindex_values`` comes from ``tindex`` when readable, and
``adios_step_count`` is observed or inferred from ADIOS metadata.
Notes
-----
Metadata discovery is a finite snapshot operation. ``FileReader`` exposes
the steps available when the source is opened, which matches catalog refresh
semantics for offline and periodically monitored datasets.
"""
import adios2 # pylint: disable=import-outside-toplevel
vars_info = {}
attrs_info = {}
step_values = []
time_values = []
tindex_values = []
step_count = 0
with adios2.FileReader(str(path)) as reader:
vars_info = reader.available_variables() or {}
attrs_info = reader.available_attributes() or {}
step_count = _max_step_count_from_vars(vars_info)
for step_id in range(int(step_count or 0)):
_append_scalar_step_value(
reader,
vars_info,
_LOGICAL_STEP_VARIABLE_NAMES,
step_values,
int,
step_id,
)
_append_scalar_step_value(reader, vars_info, ("time",), time_values, float, step_id)
_append_scalar_step_value(reader, vars_info, ("tindex",), tindex_values, int, step_id)
return _variables_from_adios_info(vars_info, attrs_info), step_values, time_values, tindex_values, step_count
def _append_scalar_step_value(reader, vars_info, names, out, value_type, step_id):
"""
Append the first readable scalar value from a set of candidate names.
Parameters
----------
reader
Open ADIOS FileReader.
vars_info : dict
Available variable metadata for the current source.
names : iterable[str]
Candidate variable names in priority order.
out : list
Destination list mutated in place.
value_type : type
Conversion type, usually ``int`` for step values or ``float`` for time.
step_id : int
ADIOS step id to inspect.
"""
for name in names:
if name not in vars_info:
continue
try:
value = reader.read(name, step_selection=[int(step_id), 1])
except Exception:
return
try:
import numpy as np # pylint: disable=import-outside-toplevel
out.append(value_type(np.asarray(value).reshape(-1)[0]))
except Exception:
return
return
def _variables_from_adios_info(vars_info, attrs_info) -> Dict[str, VariableInfo]:
"""
Convert raw ADIOS metadata dictionaries into catalog variable records.
Parameters
----------
vars_info : dict
Mapping returned by ``available_variables``.
attrs_info : dict
Mapping returned by ``available_attributes``.
Returns
-------
dict[str, VariableInfo]
Parsed variable metadata keyed by native ADIOS variable name.
"""
attrs_by_var = _attrs_by_variable(attrs_info)
variables = {}
for name, info in (vars_info or {}).items():
shape = str(info.get("Shape", "") or "")
dtype = str(info.get("Type", "") or "")
shape_dims, step_count, is_scalar = _parse_shape_spec(shape)
step_count = _available_steps_count(info) or step_count
variables[name] = VariableInfo(
name=name,
dtype=dtype,
shape=shape,
shape_dims=shape_dims,
is_scalar=is_scalar,
step_count=step_count,
metadata=_semantic_metadata(attrs_by_var.get(name, {})),
)
return variables
def _attrs_by_variable(attrs_info) -> Dict[str, Dict[str, object]]:
"""
Group ADIOS attributes by variable name.
ADIOS commonly exposes variable attributes as keys like
``variable/description``. Attributes without a variable prefix are ignored
by the initial catalog model.
"""
attrs = defaultdict(dict)
for key, info in (attrs_info or {}).items():
if "/" not in key:
continue
var_name, attr_name = key.split("/", 1)
attrs[var_name][attr_name] = info
return dict(attrs)
def _semantic_metadata(attrs: Dict[str, object]) -> VariableMetadata:
"""
Extract XGC semantic metadata fields from one variable's attributes.
Missing fields are returned as empty strings or empty lists so old outputs
without rich metadata remain valid catalog inputs. Both ``unit`` and
``units`` are accepted because existing XGC outputs use the singular form
while the metadata guidance uses the plural field name.
"""
return VariableMetadata(
description=_attr_value(attrs.get("description")),
units=_attr_value(attrs.get("units")) or _attr_value(attrs.get("unit")),
axes=_split_attr_list(attrs.get("axes")),
mesh_context=_attr_value(attrs.get("mesh_context")),
coordinate_context=_attr_value(attrs.get("coordinate_context")),
species_context=_attr_value(attrs.get("species_context")),
time_context=_attr_value(attrs.get("time_context")),
centering=_attr_value(attrs.get("centering")),
normalization=_attr_value(attrs.get("normalization")),
)
def _attr_value(info) -> str:
"""
Return the string value from one ADIOS attribute info dictionary.
Parameters
----------
info : object
Attribute metadata entry. Non-dictionary entries return ``""``.
"""
if not isinstance(info, dict):
return ""
value = info.get("Value", "")
if isinstance(value, (list, tuple)):
return ", ".join(_strip_attr_quotes(str(item)) for item in value)
return _strip_attr_quotes(str(value))
def _strip_attr_quotes(value: str) -> str:
"""
Remove one pair of ADIOS string-attribute quotes if present.
Some ADIOS builds return string attribute values with the literal quote
characters included, for example ``"m^-3"``. Numeric-looking or unquoted
values are returned unchanged.
"""
text = str(value)
if len(text) >= 2 and text[0] == '"' and text[-1] == '"':
return text[1:-1]
return text
def _split_attr_list(info) -> List[str]:
"""
Parse a comma-separated ADIOS attribute value into a list of strings.
Empty or missing attributes return an empty list.
"""
value = _attr_value(info)
if not value:
return []
return [item.strip() for item in value.split(",") if item.strip()]
def _parse_shape_spec(shape_text: str) -> Tuple[List[int], int, bool]:
"""
Parse an ADIOS shape string into payload dimensions and step count.
Parameters
----------
shape_text : str
ADIOS shape text such as ``10*{150}``, ``10*scalar``, or ``{1000, 8}``.
Returns
-------
tuple[list[int], int, bool]
Payload dimensions, leading ADIOS step count if encoded in the shape,
and whether the payload is scalar.
"""
text = str(shape_text or "").strip()
step_count = 0
payload = text
match = re.match(r"^\s*(\d+)\s*\*\s*(.+)\s*$", text)
if match:
step_count = int(match.group(1))
payload = match.group(2).strip()
nums = re.findall(r"\d+", payload or "")
payload_dims = [int(num) for num in nums]
payload_is_scalar = payload == "" or "scalar" in payload.lower()
return payload_dims, step_count, payload_is_scalar
def _max_step_count_from_vars(vars_info) -> int:
"""
Infer the maximum ADIOS step count advertised by variable metadata.
Parameters
----------
vars_info : dict
Raw ADIOS variable metadata.
"""
step_counts = []
for info in vars_info.values():
_dims, step_count, _is_scalar = _parse_shape_spec(str(info.get("Shape", "") or ""))
step_counts.append(_available_steps_count(info) or step_count)
return max(step_counts or [0])
def _available_steps_count(info) -> int:
"""
Return ``AvailableStepsCount`` from an ADIOS variable metadata record.
Parameters
----------
info : dict
Raw metadata for one ADIOS variable.
"""
try:
return int(info.get("AvailableStepsCount", 0) or 0)
except Exception:
return 0
def _layout_for_sources(sources: List[SourceInfo]) -> ProductLayout:
"""
Infer the physical product layout from source and step counts.
Multiple one-step sources are treated as a file sequence. A single
multi-step source is treated as an internal-step product. Multiple sources
with at least one multi-step source are marked mixed.
"""
if len(sources) > 1:
if any(source.adios_step_count > 1 for source in sources):
return ProductLayout.MIXED
return ProductLayout.FILE_SEQUENCE
if sources and sources[0].adios_step_count > 1:
return ProductLayout.INTERNAL_STEPS
return ProductLayout.STATIC
def _merge_variables(sources: List[SourceInfo]) -> Dict[str, VariableInfo]:
"""
Merge source variable metadata using the newer-source-wins policy.
Later sources in the sorted newness order overwrite earlier metadata for
variables with the same name. The returned product-level ``step_count`` is
the number of deduplicated logical steps available for that variable across
all sources, not just the ADIOS-internal step prefix from one source shape.
"""
variable_sources: Dict[str, List[SourceInfo]] = defaultdict(list)
for source in sources:
for name in source.variables:
variable_sources[name].append(source)
variables = {}
for name, sources_for_variable in variable_sources.items():
newest_source = max(sources_for_variable, key=_source_newness_key)
variables[name] = replace(
newest_source.variables[name],
step_count=_deduplicated_step_count(sources_for_variable),
)
return variables
def _deduplicated_step_count(sources: List[SourceInfo]) -> int:
"""
Count unique logical steps advertised by a set of sources.
Parameters
----------
sources : list[SourceInfo]
Sources that contain the variable being merged.
Returns
-------
int
Number of unique logical steps after applying the same logical-step
mapping used by read planning.
"""
logical_steps = set()
for source in sources:
for logical_step, _adios_step, _time_value in _iter_source_steps(source):
logical_steps.add(int(logical_step))
return len(logical_steps)
def _iter_source_steps(source: SourceInfo):
"""
Yield logical/adios/time step triples for one source.
Logical steps prefer explicit ``gstep``, ``step``, or ``timestep`` values in
that order. If no explicit step values are available, file-sequence sources
use the filename index plus ADIOS step offset, and non-sequence sources fall
back to ADIOS step ordinal.
"""
step_count = max(1, int(source.adios_step_count or 1))
for adios_step in range(step_count):
if adios_step < len(source.step_values):
logical_step = source.step_values[adios_step]
elif source.file_index is not None:
logical_step = source.file_index + adios_step
else:
logical_step = adios_step
time_value = source.time_values[adios_step] if adios_step < len(source.time_values) else None
yield int(logical_step), adios_step, time_value
def _source_newness_key(source: SourceInfo):
"""
Return the ordering key used to prefer newer source-level metadata.
Modification time is primary. The filename index and path provide stable
tie-breakers.
"""
file_index = source.file_index if source.file_index is not None else -1
return source.mtime, file_index, str(source.path)
def _fragment_newness_key(fragment: StepFragment):
"""
Return the ordering key used to select duplicate logical-step fragments.
The policy intentionally prefers newer source modification times over older
file-sequence indices so regenerated append-only products can supersede old
one-step files.
"""
file_index = fragment.file_index if fragment.file_index is not None else -1
try:
mtime = fragment.source_path.stat().st_mtime
except OSError:
mtime = 0.0
return mtime, file_index, str(fragment.source_path)