"""Standalone ADIOS FileReader-based BP reader."""
from __future__ import annotations
from pathlib import Path
from typing import Iterable, Sequence
import adios2
from .adios_file_reader import (
available_step_count,
available_variable_names,
read_adios_file_steps,
)
[docs]
def ReadBPFile(
filename,
variables: Sequence[str] | str | None = None,
step_range: tuple[int, int] | None = None,
*,
steps: Iterable[int] | int | None = None,
open_timeout_secs=None,
step_timeout_secs=None,
):
"""
Read selected variables and ADIOS steps from a BP file with FileReader.
Parameters
----------
filename : str or pathlib.Path
ADIOS-readable BP file or BP directory.
variables : sequence[str], str, or None, optional
Variable names to read. If omitted, all variables advertised by the
file are read.
step_range : tuple[int, int] or None, optional
Legacy half-open ADIOS step range ``(start, end)``. This path clips the
requested range to the currently available steps, preserving the old
``ReadBPFile(..., step_range=(0, huge))`` "read all available" behavior.
steps : iterable[int], int, or None, optional
Explicit ADIOS step ids to read. This path is strict: unavailable steps
raise ``IndexError`` and unavailable variables raise ``KeyError``.
If both ``steps`` and ``step_range`` are omitted, the last available
ADIOS step is read.
open_timeout_secs, step_timeout_secs
Accepted for compatibility with the previous Stream-based
implementation. FileReader is a finite snapshot API and does not use
open/step wait timeouts.
Returns
-------
dict[int, dict[str, object]]
Step-major mapping ``{adios_step: {variable_name: value}}``.
Raises
------
FileNotFoundError
If ``filename`` does not exist.
KeyError
If an explicitly requested variable is not available.
IndexError
If an explicitly requested ADIOS step is outside the available range.
ValueError
If both ``steps`` and ``step_range`` are supplied.
"""
if steps is not None and step_range is not None:
raise ValueError("Specify either `steps` or `step_range`, not both.")
if open_timeout_secs is not None or step_timeout_secs is not None:
# Compatibility-only parameters. Keep accepting them silently because
# old callers may pass them through generic read hooks.
pass
source_path = Path(filename)
if not source_path.exists():
raise FileNotFoundError(f"ADIOS BP source not found: {source_path}")
with adios2.FileReader(str(source_path)) as reader:
available_steps = int(available_step_count(reader))
if available_steps <= 0:
return {}
variable_list = _resolve_variables(reader, variables)
requested_steps = _resolve_steps(
steps=steps,
step_range=step_range,
available_steps=available_steps,
)
if not requested_steps:
return {}
return read_adios_file_steps(
source_path,
variable_list,
requested_steps,
file_reader=reader,
)
[docs]
def read_bp_file(
filename,
variables: Sequence[str] | str | None = None,
steps: Iterable[int] | int | None = None,
):
"""
Strict standalone BP reader using explicit ADIOS step ids.
Parameters
----------
filename : str or pathlib.Path
ADIOS-readable BP file or BP directory.
variables : sequence[str], str, or None, optional
Variable names to read. If omitted, all available variables are read.
steps : iterable[int], int, or None, optional
ADIOS step ids to read. If omitted, the last available step is read.
Returns
-------
dict[int, dict[str, object]]
Step-major mapping ``{adios_step: {variable_name: value}}``.
"""
return ReadBPFile(filename, variables=variables, steps=steps)
def _resolve_variables(reader, variables) -> list[str]:
"""
Resolve and validate the requested variable names.
Parameters
----------
reader : adios2.FileReader
Open FileReader handle.
variables : sequence[str], str, or None
User variable selector.
"""
available = set(available_variable_names(reader))
if variables is None:
return sorted(available)
if isinstance(variables, str):
variable_list = [variables]
else:
variable_list = list(variables)
missing = [variable for variable in variable_list if variable not in available]
if missing:
raise KeyError(f"Variable(s) not found in BP source: {', '.join(missing)}")
return variable_list
def _resolve_steps(*, steps, step_range, available_steps: int) -> list[int]:
"""
Resolve user step selectors to ADIOS step ids.
``steps`` is strict. ``step_range`` preserves legacy clipping behavior so
old "read everything" calls with a large upper bound continue to work.
"""
if steps is None and step_range is None:
return [available_steps - 1]
if steps is not None:
if isinstance(steps, int):
requested = [int(steps)]
else:
requested = [int(step) for step in steps]
missing = [step for step in requested if step < 0 or step >= available_steps]
if missing:
raise IndexError(
"ADIOS step(s) outside available range "
f"0..{available_steps - 1}: {', '.join(str(step) for step in missing)}"
)
return requested
start, end = int(step_range[0]), int(step_range[1])
start = max(0, start)
end = min(int(available_steps), end)
if end <= start:
return []
return list(range(start, end))