"""Persistence helpers for versioned solved-case artifacts."""
from __future__ import annotations
from collections.abc import Mapping
from dataclasses import dataclass
from pathlib import Path
from typing import Literal, cast
import numpy as np
import yaml
from buffalo_panel.assembly import Freestream2D
from buffalo_panel.basis.descriptors import (
LINE_CONSTANT,
LINE_LINEAR,
LINE_POINT,
LINE_QUAD,
SURF_CONSTANT,
SURF_POINT,
SURF_QUAD_LINEAR,
SURF_TRI_LINEAR,
SURF_TRI_QUAD,
BasisDescriptor,
)
from buffalo_panel.config.internal.schema import PanelCaseSpec
from buffalo_panel.families.dof_maps import (
DofMap,
PerElementDofMap,
SharedGlobalDofMap,
)
from buffalo_panel.families.element_family import ElementFamily
from buffalo_panel.geometry.line2d import (
LinePanelGeometry2D,
PointElementLineGeometry2D,
)
from buffalo_panel.geometry.supports import LINE_2D, POINT_2D, SupportDescriptor
from buffalo_panel.kernels.registry_builds import build_2d_registry
from buffalo_panel.post.internal.hess_smith import recover_hess_smith_solution
from buffalo_panel.post.internal.lumped_vortex import (
recover_lumped_vortex_solution,
)
from buffalo_panel.post.internal.results import BodyReference2D, PanelSolution2D
from buffalo_panel.singularities.descriptors import (
DOUBLET,
SOURCE,
VORTEX,
SingularityDescriptor,
)
from buffalo_panel.type_aliases import FloatArray
_ARTIFACT_VERSION = 1
_SUPPORTED_SUFFIXES = {".yaml", ".yml"}
_SMALL_NUMBER = 1.0e-8
_CONNECTIVITY_COLUMN_COUNT = 2
type DofMapKind = Literal["per_element", "shared_global"]
_BASIS_BY_NAME: dict[str, BasisDescriptor] = {
LINE_POINT.name: LINE_POINT,
LINE_CONSTANT.name: LINE_CONSTANT,
LINE_LINEAR.name: LINE_LINEAR,
LINE_QUAD.name: LINE_QUAD,
SURF_POINT.name: SURF_POINT,
SURF_CONSTANT.name: SURF_CONSTANT,
SURF_TRI_LINEAR.name: SURF_TRI_LINEAR,
SURF_TRI_QUAD.name: SURF_TRI_QUAD,
SURF_QUAD_LINEAR.name: SURF_QUAD_LINEAR,
}
_SINGULARITY_BY_NAME: dict[str, SingularityDescriptor] = {
SOURCE.name: SOURCE,
VORTEX.name: VORTEX,
DOUBLET.name: DOUBLET,
}
_SUPPORT_BY_NAME: dict[str, SupportDescriptor] = {
POINT_2D.name: POINT_2D,
LINE_2D.name: LINE_2D,
}
@dataclass(slots=True)
class ArtifactCaseMetadata:
"""User-facing metadata describing how one solution was generated."""
schema_version: int
length_unit: str
time_unit: str
angle_unit: str
case_name: str
case_description: str | None = None
@dataclass(slots=True)
class ArtifactFreestream2D:
"""Freestream state persisted with one solved artifact."""
speed: float
alpha_deg: float
@dataclass(slots=True)
class LinePanelGeometryArtifact:
"""Persisted geometry snapshot for one two-dimensional line-panel solve."""
geometry_id: int
geometry_name: str
x: list[float]
y: list[float]
connectivity: list[list[int]]
body_panel_indices: list[int]
trailing_edge_panel_index: int | None
collocation_fraction: float
element_fraction: float | None = None
@dataclass(slots=True)
class _PointElementLinePanelGeometry2D(LinePanelGeometry2D):
"""Artifact-rebuilt geometry that also carries point-element locations."""
x_element: FloatArray
y_element: FloatArray
@property
def source_points(self) -> FloatArray:
"""Return the x,y point-element locations for every panel."""
return np.column_stack((self.x_element, self.y_element))
@dataclass(slots=True)
class DofMapArtifact:
"""Persisted degree-of-freedom map metadata for one family."""
kind: DofMapKind
global_indices: list[int] | None = None
global_index: int | None = None
n_elements: int | None = None
@dataclass(slots=True)
class ElementFamilyArtifact:
"""Persisted family metadata needed to interpret solved unknowns."""
name: str
support: str
singularity: str
basis: str
n_elements: int
panel_indices: list[int]
dof_map: DofMapArtifact
metadata: dict[str, object] | None = None
[docs]
@dataclass(slots=True)
class SolvedCaseArtifact:
"""Versioned solved-case artifact for one persisted Buffalo Panel run."""
artifact_version: int
formulation: str
backend: str
case: ArtifactCaseMetadata
freestream: ArtifactFreestream2D
body_reference: BodyReference2D
geometry: LinePanelGeometryArtifact
families: list[ElementFamilyArtifact]
solution_vector: list[float]
[docs]
def build_hess_smith_artifact(
*,
spec: PanelCaseSpec,
geometry: LinePanelGeometry2D,
freestream: Freestream2D,
body_reference: BodyReference2D,
backend: str,
solution_vector: FloatArray,
source_families: tuple[ElementFamily, ...],
vortex_families: tuple[ElementFamily, ...],
) -> SolvedCaseArtifact:
"""
Build a persisted artifact from one solved Hess-Smith case.
Parameters
----------
spec : PanelCaseSpec
Structured case definition that produced the solved state.
geometry : LinePanelGeometry2D
Resolved geometry used during the solve.
freestream : Freestream2D
Runtime freestream state used by the formulation.
body_reference : BodyReference2D
Reference quantities used for integrated coefficient recovery.
backend : str
Registered kernel backend used for the solve.
solution_vector : FloatArray
Raw solved global coefficient vector.
source_families : tuple[ElementFamily, ...]
Source families interpreted by the Hess-Smith formulation.
vortex_families : tuple[ElementFamily, ...]
Vortex families interpreted by the Hess-Smith formulation.
Returns
-------
SolvedCaseArtifact
Versioned YAML-ready artifact for the solved case.
"""
family_records = [
_family_artifact_from_family(family)
for family in source_families + vortex_families
]
return SolvedCaseArtifact(
artifact_version=_ARTIFACT_VERSION,
formulation=spec.solver.formulation,
backend=backend,
case=ArtifactCaseMetadata(
schema_version=spec.schema_version,
length_unit=spec.units.length,
time_unit=spec.units.time,
angle_unit=spec.units.angle,
case_name=spec.case.name,
case_description=spec.case.description,
),
freestream=ArtifactFreestream2D(
speed=float(freestream.speed),
alpha_deg=float(freestream.alpha_deg),
),
body_reference=body_reference,
geometry=_geometry_artifact_from_geometry(geometry),
families=family_records,
solution_vector=[
float(value) for value in np.asarray(solution_vector).tolist()
],
)
[docs]
def build_lumped_vortex_artifact(
*,
spec: PanelCaseSpec,
geometry: LinePanelGeometry2D,
freestream: Freestream2D,
body_reference: BodyReference2D,
backend: str,
solution_vector: FloatArray,
vortex_families: tuple[ElementFamily, ...],
) -> SolvedCaseArtifact:
"""
Build a persisted artifact from one solved lumped-vortex case.
Parameters
----------
spec : PanelCaseSpec
Structured case definition that produced the solved state.
geometry : LinePanelGeometry2D
Resolved geometry used during the solve.
freestream : Freestream2D
Runtime freestream state used by the formulation.
body_reference : BodyReference2D
Reference quantities used for integrated coefficient recovery.
backend : str
Registered kernel backend used for the solve.
solution_vector : FloatArray
Raw solved global coefficient vector.
vortex_families : tuple[ElementFamily, ...]
Vortex families interpreted by the lumped-vortex formulation.
Returns
-------
SolvedCaseArtifact
Versioned YAML-ready artifact for the solved case.
"""
family_records = [
_family_artifact_from_family(family) for family in vortex_families
]
return SolvedCaseArtifact(
artifact_version=_ARTIFACT_VERSION,
formulation=spec.solver.formulation,
backend=backend,
case=ArtifactCaseMetadata(
schema_version=spec.schema_version,
length_unit=spec.units.length,
time_unit=spec.units.time,
angle_unit=spec.units.angle,
case_name=spec.case.name,
case_description=spec.case.description,
),
freestream=ArtifactFreestream2D(
speed=float(freestream.speed),
alpha_deg=float(freestream.alpha_deg),
),
body_reference=body_reference,
geometry=_geometry_artifact_from_geometry(geometry),
families=family_records,
solution_vector=[
float(value) for value in np.asarray(solution_vector).tolist()
],
)
[docs]
def save_solved_case_artifact(
artifact: SolvedCaseArtifact,
path: str | Path,
) -> None:
"""
Serialize one solved-case artifact to YAML.
Parameters
----------
artifact : SolvedCaseArtifact
Persisted artifact to write.
path : str | Path
Output YAML path.
"""
artifact_path = Path(path)
_require_supported_suffix(artifact_path)
text = yaml.safe_dump(
artifact_to_mapping(artifact),
sort_keys=False,
)
artifact_path.write_text(text, encoding="utf-8")
[docs]
def load_solved_case_artifact(path: str | Path) -> SolvedCaseArtifact:
"""
Load one solved-case artifact from YAML.
Parameters
----------
path : str | Path
YAML artifact path.
Returns
-------
SolvedCaseArtifact
Parsed and validated solved-case artifact.
"""
artifact_path = Path(path)
_require_supported_suffix(artifact_path)
raw_data = yaml.safe_load(artifact_path.read_text(encoding="utf-8"))
if not isinstance(raw_data, Mapping):
raise ValueError("Solved-case artifact must contain one mapping root.")
return artifact_from_mapping(cast(Mapping[str, object], raw_data))
def artifact_to_mapping(artifact: SolvedCaseArtifact) -> dict[str, object]:
"""Convert one solved-case artifact into a YAML-friendly mapping."""
return {
"artifact_version": artifact.artifact_version,
"formulation": artifact.formulation,
"backend": artifact.backend,
"case": {
"schema_version": artifact.case.schema_version,
"units": {
"length": artifact.case.length_unit,
"time": artifact.case.time_unit,
"angle": artifact.case.angle_unit,
},
"name": artifact.case.case_name,
"description": artifact.case.case_description,
},
"freestream": {
"speed": artifact.freestream.speed,
"alpha_deg": artifact.freestream.alpha_deg,
},
"body_reference": {
"reference_length": artifact.body_reference.reference_length,
"moment_reference_x": artifact.body_reference.moment_reference_x,
"moment_reference_y": artifact.body_reference.moment_reference_y,
},
"geometry": {
"geometry_id": artifact.geometry.geometry_id,
"geometry_name": artifact.geometry.geometry_name,
"x": artifact.geometry.x,
"y": artifact.geometry.y,
"connectivity": artifact.geometry.connectivity,
"collocation_fraction": artifact.geometry.collocation_fraction,
"element_fraction": artifact.geometry.element_fraction,
"body_panel_indices": artifact.geometry.body_panel_indices,
"trailing_edge_panel_index": (
artifact.geometry.trailing_edge_panel_index
),
},
"families": [_family_mapping(family) for family in artifact.families],
"solution_vector": artifact.solution_vector,
}
def artifact_from_mapping(data: Mapping[str, object]) -> SolvedCaseArtifact:
"""Build one solved-case artifact from parsed YAML mapping data."""
version = _as_int(data["artifact_version"], "artifact_version")
if version != _ARTIFACT_VERSION:
raise ValueError(f"Unsupported solved-case artifact version {version}.")
case = _require_mapping(data["case"], "case")
units = _require_mapping(case["units"], "case.units")
freestream = _require_mapping(data["freestream"], "freestream")
body_reference = _require_mapping(data["body_reference"], "body_reference")
geometry = _require_mapping(data["geometry"], "geometry")
families = _require_list(data["families"], "families")
return SolvedCaseArtifact(
artifact_version=version,
formulation=_as_str(data["formulation"], "formulation"),
backend=_as_str(data["backend"], "backend"),
case=ArtifactCaseMetadata(
schema_version=_as_int(
case["schema_version"], "case.schema_version"
),
length_unit=_as_str(units["length"], "case.units.length"),
time_unit=_as_str(units["time"], "case.units.time"),
angle_unit=_as_str(units["angle"], "case.units.angle"),
case_name=_as_str(case["name"], "case.name"),
case_description=_optional_str(case.get("description")),
),
freestream=ArtifactFreestream2D(
speed=_as_float(freestream["speed"], "freestream.speed"),
alpha_deg=_as_float(
freestream["alpha_deg"], "freestream.alpha_deg"
),
),
body_reference=BodyReference2D(
reference_length=_as_float(
body_reference["reference_length"],
"body_reference.reference_length",
),
moment_reference_x=_as_float(
body_reference["moment_reference_x"],
"body_reference.moment_reference_x",
),
moment_reference_y=_as_float(
body_reference["moment_reference_y"],
"body_reference.moment_reference_y",
),
),
geometry=LinePanelGeometryArtifact(
geometry_id=_as_int(
geometry["geometry_id"], "geometry.geometry_id"
),
geometry_name=_as_str(
geometry["geometry_name"],
"geometry.geometry_name",
),
x=_as_float_list(geometry["x"], "geometry.x"),
y=_as_float_list(geometry["y"], "geometry.y"),
connectivity=[
_as_int_list(pair, "geometry.connectivity item")
for pair in _require_list(
geometry["connectivity"],
"geometry.connectivity",
)
],
collocation_fraction=_as_float(
geometry["collocation_fraction"],
"geometry.collocation_fraction",
),
element_fraction=_optional_float(geometry.get("element_fraction")),
body_panel_indices=_as_int_list(
geometry["body_panel_indices"],
"geometry.body_panel_indices",
),
trailing_edge_panel_index=_optional_int(
geometry.get("trailing_edge_panel_index")
),
),
families=[
_family_from_mapping(
_require_mapping(family, "family"),
)
for family in families
],
solution_vector=_as_float_list(
data["solution_vector"],
"solution_vector",
),
)
[docs]
def solution_from_artifact(artifact: SolvedCaseArtifact) -> PanelSolution2D:
"""
Recover one runtime post-processing solution from a solved artifact.
Parameters
----------
artifact : SolvedCaseArtifact
Persisted artifact to recover.
Returns
-------
PanelSolution2D
Runtime post-processing interface reconstructed from ``artifact``.
"""
geometry = _geometry_from_artifact(artifact.geometry)
families = tuple(
_family_from_artifact(family) for family in artifact.families
)
if (
any(family.basis.name == LINE_POINT.name for family in families)
and artifact.geometry.element_fraction is None
):
raise ValueError(
"Line-point family artifacts require geometry.element_fraction."
)
source_families = tuple(
family for family in families if family.singularity.name == SOURCE.name
)
vortex_families = tuple(
family for family in families if family.singularity.name == VORTEX.name
)
freestream = Freestream2D(
speed=artifact.freestream.speed,
alpha_deg=artifact.freestream.alpha_deg,
)
solution_vector = np.asarray(artifact.solution_vector, dtype=np.float64)
if artifact.formulation == "hess_smith":
return recover_hess_smith_solution(
solution_vector,
source_families,
vortex_families,
geometry,
freestream,
build_2d_registry(),
artifact.backend,
artifact.body_reference,
)
if artifact.formulation == "lumped_vortex":
return recover_lumped_vortex_solution(
solution_vector,
vortex_families,
geometry,
freestream,
build_2d_registry(),
artifact.backend,
artifact.body_reference,
)
raise NotImplementedError(
f"Artifact recovery is not implemented for {artifact.formulation!r}."
)
def _geometry_artifact_from_geometry(
geometry: LinePanelGeometry2D,
) -> LinePanelGeometryArtifact:
"""Convert runtime geometry into its persisted artifact form."""
collocation_fraction = _infer_uniform_collocation_fraction(geometry)
if collocation_fraction is None:
raise ValueError(
"Artifact geometry requires one uniform collocation fraction."
)
return LinePanelGeometryArtifact(
geometry_id=geometry.geometry_id,
geometry_name=geometry.geometry_name,
x=[float(value) for value in geometry.x.tolist()],
y=[float(value) for value in geometry.y.tolist()],
connectivity=geometry.connectivity.astype(int).tolist(),
collocation_fraction=collocation_fraction,
element_fraction=_infer_uniform_element_fraction(geometry),
body_panel_indices=geometry.body_panel_indices.astype(int).tolist(),
trailing_edge_panel_index=geometry.trailing_edge_panel_index,
)
def _geometry_from_artifact(
artifact: LinePanelGeometryArtifact,
) -> LinePanelGeometry2D:
"""Rebuild line-panel geometry from the persisted artifact fields."""
x = np.asarray(artifact.x, dtype=np.float64)
y = np.asarray(artifact.y, dtype=np.float64)
connectivity = np.asarray(artifact.connectivity, dtype=np.int32)
body_panel_indices = np.asarray(
artifact.body_panel_indices,
dtype=np.int32,
)
if x.ndim != 1 or y.ndim != 1 or x.size != y.size:
raise ValueError("Artifact geometry nodes must be matching vectors.")
if (
connectivity.ndim != _CONNECTIVITY_COLUMN_COUNT
or connectivity.shape[1] != _CONNECTIVITY_COLUMN_COUNT
):
raise ValueError("Artifact connectivity must have shape (n, 2).")
if np.any(connectivity < 0) or np.any(connectivity >= x.size):
raise ValueError(
"Artifact connectivity references invalid node indices."
)
x_start = x[connectivity[:, 0]]
y_start = y[connectivity[:, 0]]
x_end = x[connectivity[:, 1]]
y_end = y[connectivity[:, 1]]
dx = x_end - x_start
dy = y_end - y_start
length = np.sqrt(dx * dx + dy * dy)
if np.any(length <= _SMALL_NUMBER):
raise ValueError("Artifact geometry contains a degenerate panel.")
s_x = dx / length
s_y = dy / length
n_x = -s_y
n_y = s_x
collocation_fraction = artifact.collocation_fraction
x_col = x_start + collocation_fraction * dx
y_col = y_start + collocation_fraction * dy
if artifact.element_fraction is None:
return LinePanelGeometry2D(
geometry_id=artifact.geometry_id,
geometry_name=artifact.geometry_name,
x=x,
y=y,
connectivity=connectivity,
x_start=x_start,
y_start=y_start,
x_end=x_end,
y_end=y_end,
x_col=x_col,
y_col=y_col,
length=length,
s_x=s_x,
s_y=s_y,
n_x=n_x,
n_y=n_y,
body_panel_indices=body_panel_indices,
trailing_edge_panel_index=artifact.trailing_edge_panel_index,
)
x_element = x_start + artifact.element_fraction * dx
y_element = y_start + artifact.element_fraction * dy
return _PointElementLinePanelGeometry2D(
geometry_id=artifact.geometry_id,
geometry_name=artifact.geometry_name,
x=x,
y=y,
connectivity=connectivity,
x_start=x_start,
y_start=y_start,
x_end=x_end,
y_end=y_end,
x_col=x_col,
y_col=y_col,
length=length,
s_x=s_x,
s_y=s_y,
n_x=n_x,
n_y=n_y,
body_panel_indices=body_panel_indices,
trailing_edge_panel_index=artifact.trailing_edge_panel_index,
x_element=x_element,
y_element=y_element,
)
def _infer_uniform_collocation_fraction(
geometry: LinePanelGeometry2D,
) -> float | None:
"""Infer one uniform collocation fraction from runtime geometry."""
explicit_fraction = getattr(geometry, "collocation_fraction", None)
if explicit_fraction is not None:
return float(cast(float, explicit_fraction))
dx = geometry.x_end - geometry.x_start
dy = geometry.y_end - geometry.y_start
panel_length_sq = dx * dx + dy * dy
collocation_dx = geometry.x_col - geometry.x_start
collocation_dy = geometry.y_col - geometry.y_start
fractions = (collocation_dx * dx + collocation_dy * dy) / np.maximum(
panel_length_sq, _SMALL_NUMBER
)
if np.allclose(fractions, fractions[0]):
return float(fractions[0])
return None
def _infer_uniform_element_fraction(
geometry: LinePanelGeometry2D,
) -> float | None:
"""Infer one uniform point-element fraction from runtime geometry."""
explicit_fraction = getattr(geometry, "element_fraction", None)
if explicit_fraction is not None:
return float(cast(float, explicit_fraction))
if not isinstance(geometry, PointElementLineGeometry2D):
return None
x_element = np.asarray(geometry.x_element, dtype=np.float64)
y_element = np.asarray(geometry.y_element, dtype=np.float64)
dx = geometry.x_end - geometry.x_start
dy = geometry.y_end - geometry.y_start
panel_length_sq = dx * dx + dy * dy
element_dx = x_element - geometry.x_start
element_dy = y_element - geometry.y_start
fractions = (element_dx * dx + element_dy * dy) / np.maximum(
panel_length_sq, _SMALL_NUMBER
)
if np.allclose(fractions, fractions[0]):
return float(fractions[0])
return None
def _family_artifact_from_family(
family: ElementFamily,
) -> ElementFamilyArtifact:
"""Convert one runtime family into persisted family metadata."""
return ElementFamilyArtifact(
name=family.name,
support=family.support.name,
singularity=family.singularity.name,
basis=family.basis.name,
n_elements=family.n_elements,
panel_indices=[
int(index)
for index in np.asarray(
family.panel_indices, dtype=np.int32
).tolist()
],
dof_map=_dof_map_artifact_from_dof_map(family.dof_map),
metadata=family.metadata,
)
def _dof_map_artifact_from_dof_map(dof_map: DofMap) -> DofMapArtifact:
"""Convert one runtime DOF map into persisted metadata."""
if isinstance(dof_map, PerElementDofMap):
return DofMapArtifact(
kind="per_element",
global_indices=[
int(index)
for index in np.asarray(
dof_map.global_indices,
dtype=np.int32,
).tolist()
],
)
if isinstance(dof_map, SharedGlobalDofMap):
return DofMapArtifact(
kind="shared_global",
global_index=dof_map.global_index,
n_elements=dof_map.n_elements,
)
raise NotImplementedError(
f"Artifact persistence is not implemented for {type(dof_map).__name__}."
)
def _family_mapping(family: ElementFamilyArtifact) -> dict[str, object]:
"""Serialize one family artifact to a YAML-friendly mapping."""
dof_map: dict[str, object] = {"kind": family.dof_map.kind}
if family.dof_map.global_indices is not None:
dof_map["global_indices"] = family.dof_map.global_indices
if family.dof_map.global_index is not None:
dof_map["global_index"] = family.dof_map.global_index
if family.dof_map.n_elements is not None:
dof_map["n_elements"] = family.dof_map.n_elements
return {
"name": family.name,
"support": family.support,
"singularity": family.singularity,
"basis": family.basis,
"n_elements": family.n_elements,
"panel_indices": family.panel_indices,
"dof_map": dof_map,
"metadata": family.metadata,
}
def _family_from_mapping(data: Mapping[str, object]) -> ElementFamilyArtifact:
"""Build one family artifact from parsed YAML mapping data."""
dof_map = _require_mapping(data["dof_map"], "family.dof_map")
global_indices_value = dof_map.get("global_indices")
return ElementFamilyArtifact(
name=_as_str(data["name"], "family.name"),
support=_as_str(data["support"], "family.support"),
singularity=_as_str(data["singularity"], "family.singularity"),
basis=_as_str(data["basis"], "family.basis"),
n_elements=_as_int(data["n_elements"], "family.n_elements"),
panel_indices=_as_int_list(
data["panel_indices"], "family.panel_indices"
),
dof_map=DofMapArtifact(
kind=cast(
DofMapKind, _as_str(dof_map["kind"], "family.dof_map.kind")
),
global_indices=(
_as_int_list(
global_indices_value,
"family.dof_map.global_indices",
)
if global_indices_value is not None
else None
),
global_index=_optional_int(dof_map.get("global_index")),
n_elements=_optional_int(dof_map.get("n_elements")),
),
metadata=cast(dict[str, object] | None, data.get("metadata")),
)
def _family_from_artifact(
family: ElementFamilyArtifact,
) -> ElementFamily:
"""Rebuild one runtime family from persisted artifact metadata."""
return ElementFamily(
name=family.name,
support=_SUPPORT_BY_NAME[family.support],
singularity=_SINGULARITY_BY_NAME[family.singularity],
basis=_BASIS_BY_NAME[family.basis],
n_elements=family.n_elements,
panel_indices=np.asarray(family.panel_indices, dtype=np.int32),
dof_map=_dof_map_from_artifact(family.dof_map),
metadata=family.metadata,
)
def _dof_map_from_artifact(dof_map: DofMapArtifact) -> DofMap:
"""Rebuild one runtime DOF map from persisted artifact metadata."""
if dof_map.kind == "per_element":
if dof_map.global_indices is None:
raise ValueError("Per-element artifact DOF map needs indices.")
return PerElementDofMap(
global_indices=np.asarray(dof_map.global_indices, dtype=np.int32)
)
if dof_map.kind == "shared_global":
if dof_map.global_index is None or dof_map.n_elements is None:
raise ValueError("Shared-global artifact DOF map is incomplete.")
return SharedGlobalDofMap(
global_index=dof_map.global_index,
n_elements=dof_map.n_elements,
)
raise ValueError(f"Unsupported artifact DOF map kind {dof_map.kind!r}.")
def _optional_int(value: object | None) -> int | None:
"""Normalize optional integer-like parsed YAML values."""
return None if value is None else _as_int(value, "optional integer")
def _optional_float(value: object | None) -> float | None:
"""Normalize optional float-like parsed YAML values."""
return None if value is None else _as_float(value, "optional float")
def _optional_str(value: object | None) -> str | None:
"""Normalize optional string-like parsed YAML values."""
return None if value is None else _as_str(value, "optional string")
def _as_float(value: object, field_name: str) -> float:
"""Validate and convert one parsed YAML scalar to ``float``."""
if isinstance(value, int | float) and not isinstance(value, bool):
return float(value)
raise ValueError(f"{field_name} must be a numeric scalar.")
def _as_int(value: object, field_name: str) -> int:
"""Validate and convert one parsed YAML scalar to ``int``."""
if isinstance(value, int) and not isinstance(value, bool):
return value
raise ValueError(f"{field_name} must be an integer scalar.")
def _as_str(value: object, field_name: str) -> str:
"""Validate and convert one parsed YAML scalar to ``str``."""
if isinstance(value, str):
return value
raise ValueError(f"{field_name} must be a string.")
def _require_mapping(value: object, field_name: str) -> Mapping[str, object]:
"""Validate and return one parsed YAML mapping value."""
if isinstance(value, Mapping):
return cast(Mapping[str, object], value)
raise ValueError(f"{field_name} must be a mapping.")
def _require_list(value: object, field_name: str) -> list[object]:
"""Validate and return one parsed YAML list value."""
if isinstance(value, list):
return cast(list[object], value)
raise ValueError(f"{field_name} must be a list.")
def _as_float_list(value: object, field_name: str) -> list[float]:
"""Validate and convert one parsed YAML list to ``list[float]``."""
return [
_as_float(item, field_name) for item in _require_list(value, field_name)
]
def _as_int_list(value: object, field_name: str) -> list[int]:
"""Validate and convert one parsed YAML list to ``list[int]``."""
return [
_as_int(item, field_name) for item in _require_list(value, field_name)
]
def _require_supported_suffix(path: Path) -> None:
"""Validate that one artifact path uses the supported YAML suffixes."""
if path.suffix.lower() not in _SUPPORTED_SUFFIXES:
raise ValueError("Solved-case artifact files must use .yaml or .yml.")