"""Shared run-result and nearby-config resolution."""
from __future__ import annotations
from dataclasses import dataclass
from functools import cached_property
from pathlib import Path
from typing import Mapping
import numpy as np
from beach.io.history import BatchHistory
from beach.io.types import RunResult
def resolve_result(value: RunResult | object) -> RunResult:
"""Resolve a result object while preserving the accepted Beach-like protocol."""
if isinstance(value, RunResult):
return value
candidate = getattr(value, "result", None)
if isinstance(candidate, RunResult):
return candidate
raise TypeError("result must be RunResult or Beach.")
def find_config_path_near_output(output_dir: Path) -> Path | None:
"""Find the nearest ``beach.toml`` using the historical search order."""
candidates = (
output_dir / "beach.toml",
output_dir.parent / "beach.toml",
output_dir.parent.parent / "beach.toml",
)
return next((candidate for candidate in candidates if candidate.exists()), None)
def resolve_config_path(
output_dir: Path,
*,
config_path: str | Path | None,
) -> Path | None:
"""Resolve an explicit or nearby config path with stable error behavior."""
if config_path is None:
return find_config_path_near_output(output_dir)
path = Path(config_path)
if not path.exists():
raise ValueError(f'config file is not found: "{path}".')
return path
def load_toml(path: Path) -> dict[str, object]:
"""Load TOML for post-processing helpers using their historical errors."""
try:
import tomllib # py311+
with path.open("rb") as stream:
return tomllib.load(stream)
except ModuleNotFoundError:
try:
import tomli # type: ignore
with path.open("rb") as stream:
return tomli.load(stream)
except ModuleNotFoundError as exc:
raise ValueError(
"TOML parser is missing. Use Python 3.11+ or install tomli: "
"`python -m pip install tomli`."
) from exc
def load_config_for_output(
output_dir: Path,
*,
config_path: str | Path | None,
) -> Mapping[str, object] | None:
"""Load a config for output reconstruction using internal reader paths.
Pre-grouped and pre-parts configs remain readable alongside archived output data;
retain that persisted-output contract while those archives are supported.
Simulator input loading rejects flat configs separately.
"""
path = resolve_config_path(output_dir, config_path=config_path)
if path is None:
return None
from beach.config._layout import is_grouped, to_runtime_layout
config = load_toml(path)
if not is_grouped(config):
return config
# Keep archived mesh tables verbatim while 1.x and pre-parts runs are supported.
# Mesh is already read directly by the output helpers, independently of input validation.
mesh = config.pop("mesh", None)
runtime = to_runtime_layout(config) if is_grouped(config) else config
if mesh is not None:
runtime["mesh"] = mesh
return runtime
[ドキュメント]
@dataclass(frozen=True)
class RunContext:
"""Resolved run plus lazily loaded, cached nearby configuration."""
result: RunResult
requested_config_path: Path | None = None
@property
def output_dir(self) -> Path:
return self.result.directory
[ドキュメント]
@classmethod
def from_value(
cls,
value: RunResult | object,
*,
config_path: str | Path | None = None,
) -> "RunContext":
"""Return the context of a result, a context, or a ``Beach`` session.
A ``Beach`` contributes its own :attr:`Beach.context`, so its config path and
cached config carry over. An explicit ``config_path`` replaces the inherited one.
"""
context = value if isinstance(value, cls) else getattr(value, "context", None)
if isinstance(context, cls):
if config_path is None:
return context
return cls(result=context.result, requested_config_path=Path(config_path))
return cls(
result=resolve_result(value),
requested_config_path=None if config_path is None else Path(config_path),
)
@cached_property
def config_path(self) -> Path | None:
return resolve_config_path(
self.result.directory,
config_path=self.requested_config_path,
)
@cached_property
def config(self) -> Mapping[str, object] | None:
return load_config_for_output(self.output_dir, config_path=self.config_path)
@cached_property
def sim(self) -> Mapping[str, object] | None:
receipt = self.result.receipts.field_reconstruction
if receipt is not None:
return receipt.as_legacy_sim_mapping()
if self.config is None:
return None
sim = self.config.get("sim")
return sim if isinstance(sim, Mapping) else None
@cached_property
def domain(self) -> Mapping[str, object] | None:
if self.result.receipts.field_reconstruction is not None:
return None
if self.config is None:
return None
domain = self.config.get("domain")
return domain if isinstance(domain, Mapping) else None
@cached_property
def field_boundary(self) -> Mapping[str, object] | None:
if self.result.receipts.field_reconstruction is not None:
return None
if self.config is None:
return None
boundary = self.config.get("field_boundary")
return boundary if isinstance(boundary, Mapping) else None
[ドキュメント]
def charges_at(self, step: int | None) -> np.ndarray:
if step is None:
return self.result.charges
history = self.result.history.charge
if history is None or not history.has_data:
if step == -1:
return self.result.charges
raise ValueError(
"charge_history.csv is required when step is specified and must not be empty."
)
if step == -1:
return history.get_step(-1)
return history.get_step(step)
[ドキュメント]
def mesh_ids_or_default(self) -> np.ndarray:
result = self.result
if result.mesh.mesh_ids is None or result.mesh.mesh_ids.size != result.summary.mesh_nelem:
return np.ones(result.summary.mesh_nelem, dtype=np.int64)
return result.mesh.mesh_ids.astype(np.int64, copy=False)
[ドキュメント]
def require_triangle_source_model(self) -> None:
if self.result.field.source_model != "triangle_p0":
raise ValueError(
"This operation requires field_source_model='triangle_p0'; "
f"the run uses {self.result.field.source_model!r}. "
"Centroid point-source reconstruction was removed."
)
[ドキュメント]
def require_triangles(self) -> np.ndarray:
if self.result.mesh.triangles is None:
raise ValueError(
"mesh_triangles.csv is not found. Re-run Fortran with latest output format."
)
return self.result.mesh.triangles
[ドキュメント]
def require_history(self) -> BatchHistory:
history = self.result.history.charge
if history is None or not history.has_data:
raise ValueError(
"charge_history.csv is not found or empty. Enable history output and rerun."
)
return history