"""Runtime validation helpers exposing canonical graph contracts.""" from __future__ import annotations import math from collections.abc import Mapping, MutableMapping, MutableSequence from typing import Any, cast from ..alias import ( get_attr, get_theta_attr, multi_recompute_abs_max, set_attr, set_attr_generic, set_theta, set_theta_attr, set_vf, ) from ..constants import DEFAULTS from ..mathematics.unified_numerical import np from ..types import ( ZERO_BEPI_STORAGE, NodeId, TNFRGraph, _is_scalar, ensure_bepi, serialize_bepi, ) from ..utils import clamp, ensure_collection from .base import ValidationOutcome, Validator from .graph import run_validators HistoryLog = MutableSequence[MutableMapping[str, object]] """Mutable history buffer storing clamp alerts for strict graphs.""" __all__ = ( "GraphCanonicalValidator", "apply_canonical_clamps", "validate_canon", ) def _max_bepi_magnitude(value: Any) -> float: element = ensure_bepi(value) mags = [ ( float(np.max(np.abs(element.f_continuous))) if element.f_continuous.size else 0.0 ), float(np.max(np.abs(element.a_discrete))) if element.a_discrete.size else 0.0, ] return float(max(mags)) if mags else 0.0 def _clamp_component(values: Any, lower: float, upper: float) -> np.ndarray: array = np.asarray(values, dtype=np.complex128) if array.size == 0: return array magnitudes = np.abs(array) result = array.copy() above = magnitudes > upper if np.any(above): result[above] = array[above] * (upper / magnitudes[above]) if lower > 0.0: below = (magnitudes < lower) & (magnitudes > 0.0) if np.any(below): result[below] = array[below] * (lower / magnitudes[below]) return result def _clamp_bepi(value: Any, lower: float, upper: float) -> Any: element = ensure_bepi(value) clamped_cont = _clamp_component(element.f_continuous, lower, upper) clamped_disc = _clamp_component(element.a_discrete, lower, upper) return ensure_bepi( {"continuous": clamped_cont, "discrete": clamped_disc, "grid": element.x_grid} ) def _log_clamp( hist: HistoryLog, node: NodeId | None, attr: str, value: float, lo: float, hi: float, ) -> None: if value < lo or value > hi: hist.append({"node": node, "attr": attr, "value": float(value)}) def apply_canonical_clamps( nd: MutableMapping[str, Any], G: TNFRGraph | None = None, node: NodeId | None = None, ) -> None: """Clamp nodal EPI, νf and θ according to canonical bounds.""" from ..dynamics.aliases import ALIAS_EPI, ALIAS_VF if G is not None: graph_dict = cast(MutableMapping[str, Any], G.graph) graph_data: Mapping[str, Any] = graph_dict else: graph_dict = None graph_data = DEFAULTS eps_min = float(graph_data.get("EPI_MIN", DEFAULTS["EPI_MIN"])) eps_max = float(graph_data.get("EPI_MAX", DEFAULTS["EPI_MAX"])) vf_min = float(graph_data.get("VF_MIN", DEFAULTS["VF_MIN"])) vf_max = float(graph_data.get("VF_MAX", DEFAULTS["VF_MAX"])) theta_wrap = bool(graph_data.get("THETA_WRAP", DEFAULTS["THETA_WRAP"])) raw_epi = get_attr(nd, ALIAS_EPI, ZERO_BEPI_STORAGE, conv=lambda obj: obj) was_scalar = _is_scalar(raw_epi) original_scalar_value = float(raw_epi) if was_scalar else None epi = ensure_bepi(raw_epi) vf = get_attr(nd, ALIAS_VF, 0.0) th_val = get_theta_attr(nd, 0.0) th = 0.0 if th_val is None else float(th_val) strict = bool( graph_data.get("VALIDATORS_STRICT", DEFAULTS.get("VALIDATORS_STRICT", False)) ) if strict and graph_dict is not None: history = cast(MutableMapping[str, Any], graph_dict.setdefault("history", {})) alerts = history.get("clamp_alerts") if alerts is None: hist = cast(HistoryLog, history.setdefault("clamp_alerts", [])) elif isinstance(alerts, MutableSequence): hist = cast(HistoryLog, alerts) else: materialized = ensure_collection(alerts, max_materialize=None) hist = cast(HistoryLog, list(materialized)) history["clamp_alerts"] = hist epi_mag = _max_bepi_magnitude(epi) _log_clamp(hist, node, "EPI", epi_mag, eps_min, eps_max) _log_clamp(hist, node, "VF", float(vf), vf_min, vf_max) clamped_epi = _clamp_bepi(epi, eps_min, eps_max) if was_scalar: clamped_value = float(clamp(original_scalar_value, eps_min, eps_max)) set_attr_generic(nd, ALIAS_EPI, clamped_value, conv=lambda obj: obj) else: set_attr_generic( nd, ALIAS_EPI, serialize_bepi(clamped_epi), conv=lambda obj: obj ) vf_val = float(clamp(vf, vf_min, vf_max)) if G is not None and node is not None: set_vf(G, node, vf_val, update_max=False) else: set_attr(nd, ALIAS_VF, vf_val) if theta_wrap: # Preserve the native centered chart exactly when already inside it. # Adding/subtracting pi needlessly erases tiny phases and shifts other # valid values; setters own alias/cache writes, not normalization. new_th = ( th if -math.pi <= th < math.pi else (th + math.pi) % (2 * math.pi) - math.pi ) if G is not None and node is not None: set_theta(G, node, new_th) else: set_theta_attr(nd, new_th) class GraphCanonicalValidator(Validator[TNFRGraph]): """Validator enforcing canonical runtime contracts on TNFR graphs.""" recompute_frequency_maxima: bool enforce_graph_validators: bool def __init__( self, *, recompute_frequency_maxima: bool = True, enforce_graph_validators: bool = True, ) -> None: self.recompute_frequency_maxima = bool(recompute_frequency_maxima) self.enforce_graph_validators = bool(enforce_graph_validators) def _diff_after_clamp( self, before: Mapping[str, float], after: Mapping[str, float], ) -> Mapping[str, Mapping[str, float]]: changes: dict[str, Mapping[str, float]] = {} for key, before_val in before.items(): after_val = after.get(key) if after_val is None: continue if math.isclose(before_val, after_val, rel_tol=0.0, abs_tol=1e-12): continue changes[key] = {"before": before_val, "after": after_val} return changes def validate(self, subject: TNFRGraph, /, **_: Any) -> ValidationOutcome[TNFRGraph]: """Clamp nodal attributes, refresh νf maxima and run graph validators.""" from ..dynamics.aliases import ALIAS_EPI, ALIAS_VF clamped_nodes: list[dict[str, Any]] = [] for node, data in subject.nodes(data=True): mapping = cast(MutableMapping[str, Any], data) before = { "EPI": _max_bepi_magnitude( get_attr( mapping, ALIAS_EPI, ZERO_BEPI_STORAGE, conv=lambda obj: obj ) ), "VF": float(get_attr(mapping, ALIAS_VF, 0.0)), "THETA": float(get_theta_attr(mapping, 0.0) or 0.0), } apply_canonical_clamps(mapping, subject, cast(NodeId, node)) after = { "EPI": _max_bepi_magnitude( get_attr( mapping, ALIAS_EPI, ZERO_BEPI_STORAGE, conv=lambda obj: obj ) ), "VF": float(get_attr(mapping, ALIAS_VF, 0.0)), "THETA": float(get_theta_attr(mapping, 0.0) or 0.0), } changes = self._diff_after_clamp(before, after) if changes: clamped_nodes.append({"node": node, "attributes": changes}) maxima: Mapping[str, float] = {} if self.recompute_frequency_maxima: maxima = multi_recompute_abs_max(subject, {"_vfmax": ALIAS_VF}) subject.graph.update(maxima) errors: list[str] = [] if self.enforce_graph_validators: try: run_validators(subject) except Exception as exc: # pragma: no cover - defensive errors.append(str(exc)) summary: dict[str, Any] = { "clamped": tuple(clamped_nodes), "maxima": dict(maxima), } if errors: summary["errors"] = tuple(errors) passed = not errors artifacts: dict[str, Any] = {} if clamped_nodes: artifacts["clamped_nodes"] = clamped_nodes if maxima: artifacts["maxima"] = dict(maxima) return ValidationOutcome( subject=subject, passed=passed, summary=summary, artifacts=artifacts or None, ) def report(self, outcome: ValidationOutcome[TNFRGraph]) -> str: """Return a concise textual summary of ``outcome``.""" if outcome.passed: clamped = outcome.summary.get("clamped") if clamped: return "Graph canonical validation applied clamps successfully." return "Graph canonical validation passed without adjustments." errors = outcome.summary.get("errors") if not errors: return "Graph canonical validation failed without diagnostic details." if isinstance(errors, (list, tuple)): return "Graph canonical validation errors: " + ", ".join(map(str, errors)) return f"Graph canonical validation error: {errors}" def validate_canon(G: TNFRGraph) -> ValidationOutcome[TNFRGraph]: """Validate ``G`` using :class:`GraphCanonicalValidator`.""" validator = GraphCanonicalValidator() return validator.validate(G)