"""Data management for Buffalo Panel simulations in the TUI."""
from __future__ import annotations
from collections.abc import Sequence
from typing import Any, Literal, TypeGuard, cast
from buffalo_panel.config import (
SchemaFieldTree,
panel_case_from_mapping,
)
from buffalo_panel.config.internal.loading import (
airfoil_mapping_source,
)
type SimulationScalar = bool | int | float | str | None
type SimulationValue = (
SimulationScalar | list["SimulationValue"] | dict[str, "SimulationValue"]
)
type NodePath = tuple[str | int, ...]
type AirfoilSource = Literal["designation", "params"]
type PendingApproxAirfoilSwitch = tuple[
NodePath,
AirfoilSource,
dict[str, SimulationValue],
str,
]
AIRFOIL_ENTRY_FIELD_NAME = "airfoil"
AIRFOIL_ENTRY_PATH_LENGTH = 4
AIRFOIL_SOURCE_FIELD_NAME = "__source__"
AIRFOIL_SWITCH_SOURCE_COUNT = 2
AIRFOIL_TYPE_PATH_LENGTH = 5
TWO_DIMENSIONAL_TUPLE_LENGTH = 2
[docs]
class SimulationDataManager:
"""Encapsulates raw data manipulation and validation logic.
This class decouples the TUI widgets from the underlying dictionary
traversal and schema validation.
"""
def __init__(self, data: dict[str, SimulationValue]) -> None:
self.data = data
[docs]
def get_value(self, path: NodePath) -> SimulationValue:
"""Return the current simulation value at ``path``."""
if self.is_airfoil_source_path(path):
airfoil_value = self.get_value(path[:-1])
if not isinstance(airfoil_value, dict):
raise TypeError("Airfoil source nodes require mapping values.")
source = airfoil_mapping_source(
cast(dict[str, object], airfoil_value)
)
if source is None:
raise ValueError(
"Airfoil mapping must use designation or params."
)
return source
current: SimulationValue = self.data
for key in path:
if isinstance(current, dict):
current = current[str(key)]
elif isinstance(current, list):
if not isinstance(key, int):
raise TypeError("List paths must use integer indices.")
current = current[key]
else:
raise TypeError("Scalar values do not have children.")
return current
[docs]
def set_value(self, path: NodePath, value: SimulationScalar) -> None:
"""Update one scalar field and validate the full case mapping."""
parent = self._parent_container(path)
last_key = path[-1]
if isinstance(parent, dict):
old_value = parent[str(last_key)]
parent[str(last_key)] = value
else:
if not isinstance(last_key, int):
raise TypeError("List assignments require integer indices.")
index = last_key
old_value = parent[index]
parent[index] = value
try:
panel_case_from_mapping(cast(dict[str, object], self.data))
except Exception:
if isinstance(parent, dict):
parent[str(last_key)] = old_value
else:
if not isinstance(last_key, int):
raise TypeError(
"List assignments require integer indices."
) from None
parent[last_key] = old_value
raise
def _parent_container(
self, path: NodePath
) -> dict[str, SimulationValue] | list[SimulationValue]:
"""Return the mutable parent container for one path."""
current: SimulationValue = self.data
for key in path[:-1]:
if isinstance(current, dict):
current = current[str(key)]
elif isinstance(current, list):
if not isinstance(key, int):
raise TypeError("List paths must use integer indices.")
current = current[key]
else:
raise TypeError("Scalar values do not have children.")
if isinstance(current, dict | list):
return current
raise TypeError("Leaf values do not have child containers.")
[docs]
@staticmethod
def coerce_value(
value_text: str, schema: SchemaFieldTree
) -> SimulationScalar:
"""Coerce input text to the strictest scalar type allowed by schema."""
stripped_value = value_text.strip()
if schema.optional and stripped_value == "":
return None
metadata = schema.metadata
value_kind = metadata.get("value_kind")
result: SimulationScalar
if value_kind == "int":
result = int(stripped_value)
elif value_kind in {"float", "number"}:
result = float(stripped_value)
elif value_kind == "bool":
normalized = stripped_value.lower()
if normalized in {"true", "1", "yes", "on"}:
result = True
elif normalized in {"false", "0", "no", "off"}:
result = False
else:
raise ValueError("Boolean fields accept true/false values.")
else:
result = stripped_value
# Range validation
if isinstance(result, int | float) and not isinstance(result, bool):
if (
min_val := metadata.get("minimum")
) is not None and result < cast(float, min_val):
raise ValueError(f"Value must be >= {min_val}.")
if (
max_val := metadata.get("maximum")
) is not None and result > cast(float, max_val):
raise ValueError(f"Value must be <= {max_val}.")
if (
ex_min := metadata.get("exclusive_minimum")
) is not None and result <= cast(float, ex_min):
raise ValueError(f"Value must be > {ex_min}.")
if (
ex_max := metadata.get("exclusive_maximum")
) is not None and result >= cast(float, ex_max):
raise ValueError(f"Value must be < {ex_max}.")
# Choices validation
raw_choices = metadata.get("choices") or metadata.get("item_choices")
if raw_choices is not None and isinstance(raw_choices, tuple | list):
valid_values: list[SimulationScalar] = []
for c in cast(Sequence[Any], raw_choices):
if _is_dict(c):
valid_values.append(cast(SimulationScalar, c.get("value")))
elif hasattr(c, "value"):
valid_values.append(cast(SimulationScalar, c.value))
else:
valid_values.append(cast(SimulationScalar, c))
if result not in valid_values:
raise ValueError(f"Value must be one of {valid_values!r}.")
return result
[docs]
@staticmethod
def is_airfoil_source_path(path: NodePath) -> bool:
"""Return whether one path targets the synthetic source field."""
return bool(path) and path[-1] == AIRFOIL_SOURCE_FIELD_NAME
[docs]
def airfoil_entry_type(self, path: NodePath) -> str | None:
"""Return the airfoil family for one airfoil-entry path."""
if not _is_airfoil_entry_path(path):
return None
airfoil_value = self.get_value(path)
if not isinstance(airfoil_value, dict):
return None
airfoil_type = airfoil_value.get("type")
if isinstance(airfoil_type, str):
return airfoil_type
return None
[docs]
def apply_airfoil_mapping(
self, path: NodePath, new_mapping: SimulationValue
) -> None:
"""Replace one airfoil mapping and validate the full case payload."""
airfoil_path = path[:-1]
parent = self._parent_container(airfoil_path)
airfoil_key = airfoil_path[-1]
if not isinstance(parent, dict):
raise TypeError("Airfoil entries must live inside one mapping.")
old_value = parent[str(airfoil_key)]
parent[str(airfoil_key)] = new_mapping
try:
panel_case_from_mapping(cast(dict[str, object], self.data))
except Exception:
parent[str(airfoil_key)] = old_value
raise
[docs]
@staticmethod
def is_branch_value(value: SimulationValue) -> bool:
"""Return whether one value should render as a branch node."""
return isinstance(value, dict | list)
def _is_dict(value: object) -> TypeGuard[dict[object, object]]:
"""
Return whether ``value`` is a dictionary.
Parameters
----------
value : object
The value to check.
Returns
-------
TypeGuard[Dict[object, object]]
True if the value is a dictionary.
"""
return isinstance(value, dict)
def _is_airfoil_entry_path(path: NodePath) -> bool:
"""Return whether one path points to an embedded body airfoil mapping."""
return (
len(path) == AIRFOIL_ENTRY_PATH_LENGTH
and path[0] == "geometry"
and path[1] == "bodies"
and isinstance(path[2], int)
and path[3] == AIRFOIL_ENTRY_FIELD_NAME
)