""" Generic stability metrics for WorldSmithAI. This module computes stability over arbitrary world collections. It deliberately avoids hardcoded species, farm, research, market, transport, or civilization assumptions. A stability metric can measure: - population stability by agent type - resource amount stability by resource type - goal distribution stability - market-role stability - infrastructure-status stability - strategy/adoption stability - any DSL-defined grouped numeric path Example: metric = StabilityMetric(collection="agents", group_by_path="type") result = metric.compute(world) print(result.variance) print(result.stability_score) Temporal tracking: tracker = StabilityTracker(metric) for _ in range(100): world.step() tracker.update(world) Future extensibility: - Add rolling-window variance and volatility. - Add autocorrelation and trend stability. - Add equilibrium detection. - Add collapse and boom detection. - Add stability contributions per group. - Feed stability signals into interestingness and God-Agent metrics. """ from __future__ import annotations import copy import logging import math from collections.abc import Iterable, Mapping, MutableSequence, Sequence from dataclasses import dataclass, field from enum import Enum from numbers import Real from types import MappingProxyType from typing import TYPE_CHECKING, Any, ClassVar import numpy as np if TYPE_CHECKING: from core.world import World logger = logging.getLogger(__name__) _MISSING = object() _EPSILON = 1.0e-12 class StabilityCollection(str, Enum): """Collections over which stability may be computed.""" AGENTS = "agents" RESOURCES = "resources" BOTH = "both" class StabilityAggregation(str, Enum): """Aggregation modes used to convert world objects into group values.""" COUNT = "count" SUM = "sum" MEAN = "mean" @dataclass(frozen=True) class StabilityResult: """Result produced by stability metric computation. Attributes: metric_name: Stable metric name. collection: Collection analyzed. group_by_path: Path used to group objects. value_path: Optional numeric path used for sum or mean aggregation. weight_path: Optional numeric path used as contribution weight. aggregation: Aggregation mode. item_count: Number of included objects. group_count: Number of groups after aggregation. group_values: Current value per group. previous_group_values: Previous value per group when supplied. deltas: Current minus previous value per group when supplied. total_value: Sum of current group values. mean_value: Mean group value. variance: Population variance across current group values. standard_deviation: Population standard deviation across group values. coefficient_of_variation: Standard deviation divided by absolute mean. min_value: Minimum group value. max_value: Maximum group value. value_range: Difference between max and min. normalized_range: Range divided by absolute mean. temporal_variance: Variance of group deltas when previous values exist. temporal_standard_deviation: Standard deviation of deltas. total_absolute_delta: Sum of absolute group deltas. mean_absolute_delta: Mean absolute group delta. mean_relative_delta: Mean relative group delta. stability_score: Bounded score in [0, 1], where higher is more stable. step: Optional world step. metadata: Additional JSON-compatible details. """ metric_name: str collection: str group_by_path: str value_path: str | None weight_path: str | None aggregation: str item_count: int group_count: int group_values: Mapping[str, float] = field(default_factory=dict) previous_group_values: Mapping[str, float] = field(default_factory=dict) deltas: Mapping[str, float] = field(default_factory=dict) total_value: float = 0.0 mean_value: float = 0.0 variance: float = 0.0 standard_deviation: float = 0.0 coefficient_of_variation: float = 0.0 min_value: float = 0.0 max_value: float = 0.0 value_range: float = 0.0 normalized_range: float = 0.0 temporal_variance: float = 0.0 temporal_standard_deviation: float = 0.0 total_absolute_delta: float = 0.0 mean_absolute_delta: float = 0.0 mean_relative_delta: float = 0.0 stability_score: float = 1.0 step: int | None = None metadata: Mapping[str, Any] = field(default_factory=dict) @property def is_empty(self) -> bool: """Return whether no objects contributed to this result.""" return self.item_count == 0 or self.group_count == 0 @property def is_temporal(self) -> bool: """Return whether this result includes previous-state comparison.""" return bool(self.previous_group_values) def to_dict(self) -> dict[str, Any]: """Return a JSON-friendly representation of the stability result.""" return { "metric_name": self.metric_name, "collection": self.collection, "group_by_path": self.group_by_path, "value_path": self.value_path, "weight_path": self.weight_path, "aggregation": self.aggregation, "item_count": self.item_count, "group_count": self.group_count, "group_values": copy.deepcopy(dict(self.group_values)), "previous_group_values": copy.deepcopy(dict(self.previous_group_values)), "deltas": copy.deepcopy(dict(self.deltas)), "total_value": self.total_value, "mean_value": self.mean_value, "variance": self.variance, "standard_deviation": self.standard_deviation, "coefficient_of_variation": self.coefficient_of_variation, "min_value": self.min_value, "max_value": self.max_value, "value_range": self.value_range, "normalized_range": self.normalized_range, "temporal_variance": self.temporal_variance, "temporal_standard_deviation": self.temporal_standard_deviation, "total_absolute_delta": self.total_absolute_delta, "mean_absolute_delta": self.mean_absolute_delta, "mean_relative_delta": self.mean_relative_delta, "stability_score": self.stability_score, "is_empty": self.is_empty, "is_temporal": self.is_temporal, "step": self.step, "metadata": copy.deepcopy(dict(self.metadata)), } @dataclass class StabilityMetric: """Compute generic stability over grouped world objects. The default configuration computes population stability over alive agents grouped by ``type``. This can represent species, roles, professions, factions, node classes, strategies, or any DSL-defined type label. Examples: Agent population stability: metric = StabilityMetric() result = metric.compute(world) Resource amount stability: metric = StabilityMetric( collection="resources", group_by_path="type", value_path="amount", aggregation="sum", alive_only=False, ) result = metric.compute(world) Goal distribution stability: metric = StabilityMetric( collection="agents", group_by_path="state.current_goal", include_missing=True, ) result = metric.compute(world) """ name: ClassVar[str] = "stability" collection: StabilityCollection | str = StabilityCollection.AGENTS group_by_path: str = "type" value_path: str | None = None weight_path: str | None = None aggregation: StabilityAggregation | str = StabilityAggregation.COUNT alive_only: bool = True include_missing: bool = False missing_label: str = "unknown" include_group_values: tuple[str, ...] = () exclude_group_values: tuple[str, ...] = () case_sensitive: bool = True strip_labels: bool = True include_collection_prefix: bool = False minimum_weight: float = 0.0 metadata: Mapping[str, Any] = field(default_factory=dict) def compute( self, world: World, *, previous_values: Mapping[str, float] | None = None, ) -> StabilityResult: """Compute stability for the configured world collection. Args: world: Runtime world object. previous_values: Optional previous group values. When supplied, temporal stability statistics are included. Returns: ``StabilityResult`` with snapshot and optional temporal statistics. """ collection = _normalize_collection(self.collection) aggregation = _normalize_aggregation(self.aggregation) items = _iter_world_items(world, collection) group_values, included_count, skipped_count = self._group_values(items, collection, aggregation) result = stability_from_values( group_values, previous_values=previous_values, metric_name=self.name, collection=collection.value, group_by_path=self.group_by_path, value_path=self.value_path, weight_path=self.weight_path, aggregation=aggregation.value, item_count=included_count, step=_world_step(world), metadata={ **copy.deepcopy(dict(self.metadata)), "skipped_count": skipped_count, "alive_only": self.alive_only, "include_missing": self.include_missing, "minimum_weight": self.minimum_weight, }, ) logger.debug( "Computed stability over %s grouped by %s: variance=%.6f score=%.6f", collection.value, self.group_by_path, result.variance, result.stability_score, ) return result def __call__( self, world: World, *, previous_values: Mapping[str, float] | None = None, ) -> StabilityResult: """Compute stability, allowing metric instances to be called directly.""" return self.compute(world, previous_values=previous_values) def _group_values( self, items: Sequence[Any], collection: StabilityCollection, aggregation: StabilityAggregation, ) -> tuple[dict[str, float], int, int]: """Aggregate world objects into group values.""" group_sums: dict[str, float] = {} group_weights: dict[str, float] = {} included_count = 0 skipped_count = 0 include_values = { _normalize_label(value, case_sensitive=self.case_sensitive, strip=True) for value in self.include_group_values } exclude_values = { _normalize_label(value, case_sensitive=self.case_sensitive, strip=True) for value in self.exclude_group_values } for item in items: if self.alive_only and _is_agent_like(item) and not _is_alive(item): skipped_count += 1 continue label = self._label_for(item, collection) if label is None: skipped_count += 1 continue normalized_label = _normalize_label( label, case_sensitive=self.case_sensitive, strip=self.strip_labels, ) if include_values and normalized_label not in include_values: skipped_count += 1 continue if exclude_values and normalized_label in exclude_values: skipped_count += 1 continue weight = self._weight_for(item) if weight <= max(0.0, float(self.minimum_weight)): skipped_count += 1 continue value = self._value_for(item, aggregation) if value is None: skipped_count += 1 continue if aggregation is StabilityAggregation.COUNT: contribution = weight denominator = 1.0 elif aggregation is StabilityAggregation.SUM: contribution = value * weight denominator = 1.0 else: contribution = value * weight denominator = weight group_sums[normalized_label] = group_sums.get(normalized_label, 0.0) + contribution group_weights[normalized_label] = group_weights.get(normalized_label, 0.0) + denominator included_count += 1 if aggregation is StabilityAggregation.MEAN: group_values = { label: group_sums[label] / max(group_weights.get(label, 0.0), _EPSILON) for label in group_sums } else: group_values = group_sums return dict(sorted(group_values.items(), key=lambda item: item[0])), included_count, skipped_count def _label_for(self, item: Any, collection: StabilityCollection) -> str | None: """Return the group label for an item.""" raw_label = _read_path(item, self.group_by_path, _MISSING) if raw_label is _MISSING or raw_label is None or str(raw_label) == "": if not self.include_missing: return None raw_label = self.missing_label label = _stable_label(raw_label) if self.include_collection_prefix and collection is StabilityCollection.BOTH: return f"{_item_collection_name(item)}:{label}" return label def _weight_for(self, item: Any) -> float: """Return the non-negative item weight.""" if self.weight_path is None: return 1.0 raw_weight = _read_path(item, self.weight_path, 0.0) if not _is_number(raw_weight): return 0.0 return max(0.0, float(raw_weight)) def _value_for(self, item: Any, aggregation: StabilityAggregation) -> float | None: """Return the numeric value contributed by an item.""" if aggregation is StabilityAggregation.COUNT: return 1.0 if self.value_path is None: return 1.0 raw_value = _read_path(item, self.value_path, _MISSING) if not _is_number(raw_value): return None return float(raw_value) @dataclass class StabilityTracker: """Track stability over time. ``StabilityMetric`` can compute one snapshot. ``StabilityTracker`` stores previous group values and produces temporal delta statistics at each update. """ metric: StabilityMetric = field(default_factory=StabilityMetric) max_history: int = 1000 history: list[StabilityResult] = field(default_factory=list) previous_values: Mapping[str, float] = field(default_factory=dict) def update(self, world: World) -> StabilityResult: """Compute stability and append the result to history.""" result = self.metric.compute(world, previous_values=self.previous_values) self.previous_values = copy.deepcopy(dict(result.group_values)) _append_bounded(self.history, result, self.max_history) return result def latest(self) -> StabilityResult | None: """Return the latest stability result, if any.""" if not self.history: return None return self.history[-1] def reset(self) -> None: """Clear tracked history and previous values.""" self.history.clear() self.previous_values = {} def to_series(self, value_key: str = "stability_score") -> list[dict[str, Any]]: """Return a JSON-friendly time series for one result field. Args: value_key: Name of a ``StabilityResult`` attribute to extract. Returns: List of dictionaries with ``index``, ``step``, and ``value``. """ series: list[dict[str, Any]] = [] for index, result in enumerate(self.history): value = getattr(result, value_key, None) if not _is_number(value): continue series.append( { "index": index, "step": result.step, "value": float(value), } ) return series def to_dict(self) -> dict[str, Any]: """Return a JSON-friendly tracker representation.""" return { "metric": { "name": self.metric.name, "collection": _normalize_collection(self.metric.collection).value, "group_by_path": self.metric.group_by_path, "value_path": self.metric.value_path, "weight_path": self.metric.weight_path, "aggregation": _normalize_aggregation(self.metric.aggregation).value, }, "max_history": self.max_history, "previous_values": copy.deepcopy(dict(self.previous_values)), "history": [result.to_dict() for result in self.history], } def compute_stability( world: World, *, collection: StabilityCollection | str = StabilityCollection.AGENTS, group_by_path: str = "type", value_path: str | None = None, weight_path: str | None = None, aggregation: StabilityAggregation | str = StabilityAggregation.COUNT, alive_only: bool = True, include_missing: bool = False, previous_values: Mapping[str, float] | None = None, ) -> StabilityResult: """Convenience function for computing generic stability.""" metric = StabilityMetric( collection=collection, group_by_path=group_by_path, value_path=value_path, weight_path=weight_path, aggregation=aggregation, alive_only=alive_only, include_missing=include_missing, ) return metric.compute(world, previous_values=previous_values) def compute_agent_population_stability( world: World, *, alive_only: bool = True, include_missing: bool = False, previous_values: Mapping[str, float] | None = None, ) -> StabilityResult: """Compute population stability over agent ``type`` labels.""" return compute_stability( world, collection=StabilityCollection.AGENTS, group_by_path="type", aggregation=StabilityAggregation.COUNT, alive_only=alive_only, include_missing=include_missing, previous_values=previous_values, ) def compute_resource_amount_stability( world: World, *, include_missing: bool = False, previous_values: Mapping[str, float] | None = None, ) -> StabilityResult: """Compute resource amount stability over resource ``type`` labels.""" return compute_stability( world, collection=StabilityCollection.RESOURCES, group_by_path="type", value_path="amount", aggregation=StabilityAggregation.SUM, alive_only=False, include_missing=include_missing, previous_values=previous_values, ) def stability_from_values( group_values: Mapping[str, float], *, previous_values: Mapping[str, float] | None = None, metric_name: str = "stability", collection: str = "custom", group_by_path: str = "group", value_path: str | None = None, weight_path: str | None = None, aggregation: str = "count", item_count: int | None = None, step: int | None = None, metadata: Mapping[str, Any] | None = None, ) -> StabilityResult: """Compute stability statistics from current and optional previous values. Args: group_values: Current group values. previous_values: Optional previous group values. metric_name: Name stored in the result. collection: Collection label stored in the result. group_by_path: Grouping path stored in the result. value_path: Optional numeric value path stored in the result. weight_path: Optional weight path stored in the result. aggregation: Aggregation label stored in the result. item_count: Optional number of contributing items. step: Optional world step. metadata: Optional result metadata. Returns: ``StabilityResult`` with cross-sectional and temporal statistics. """ cleaned_values = { str(label): float(value) for label, value in group_values.items() if _is_number(value) and math.isfinite(float(value)) } sorted_values = dict(sorted(cleaned_values.items(), key=lambda item: item[0])) if not sorted_values: return StabilityResult( metric_name=metric_name, collection=collection, group_by_path=group_by_path, value_path=value_path, weight_path=weight_path, aggregation=aggregation, item_count=0 if item_count is None else int(item_count), group_count=0, step=step, metadata=metadata or {}, ) values_array = np.asarray(list(sorted_values.values()), dtype=float) total_value = float(np.sum(values_array)) mean_value = float(np.mean(values_array)) variance = float(np.var(values_array)) standard_deviation = float(np.sqrt(variance)) coefficient_of_variation = ( float(standard_deviation / abs(mean_value)) if abs(mean_value) > _EPSILON else 0.0 ) min_value = float(np.min(values_array)) max_value = float(np.max(values_array)) value_range = max_value - min_value normalized_range = ( float(value_range / abs(mean_value)) if abs(mean_value) > _EPSILON else 0.0 ) previous_cleaned = _clean_previous_values(previous_values) deltas = _compute_deltas(sorted_values, previous_cleaned) temporal_statistics = _temporal_statistics(deltas, previous_cleaned) stability_score = _stability_score( coefficient_of_variation=coefficient_of_variation, normalized_range=normalized_range, mean_relative_delta=temporal_statistics["mean_relative_delta"], ) return StabilityResult( metric_name=metric_name, collection=collection, group_by_path=group_by_path, value_path=value_path, weight_path=weight_path, aggregation=aggregation, item_count=len(sorted_values) if item_count is None else int(item_count), group_count=len(sorted_values), group_values=sorted_values, previous_group_values=previous_cleaned, deltas=deltas, total_value=total_value, mean_value=mean_value, variance=variance, standard_deviation=standard_deviation, coefficient_of_variation=coefficient_of_variation, min_value=min_value, max_value=max_value, value_range=value_range, normalized_range=normalized_range, temporal_variance=temporal_statistics["temporal_variance"], temporal_standard_deviation=temporal_statistics["temporal_standard_deviation"], total_absolute_delta=temporal_statistics["total_absolute_delta"], mean_absolute_delta=temporal_statistics["mean_absolute_delta"], mean_relative_delta=temporal_statistics["mean_relative_delta"], stability_score=stability_score, step=step, metadata=metadata or {}, ) def population_variance(values: Sequence[float]) -> float: """Return population variance for a numeric sequence.""" array = _numeric_array(values) if array.size == 0: return 0.0 return float(np.var(array)) def coefficient_of_variation(values: Sequence[float]) -> float: """Return coefficient of variation for a numeric sequence.""" array = _numeric_array(values) if array.size == 0: return 0.0 mean_value = float(np.mean(array)) if abs(mean_value) <= _EPSILON: return 0.0 return float(np.std(array) / abs(mean_value)) def relative_delta(current: float, previous: float) -> float: """Return a bounded relative delta magnitude. If the previous value is zero and the current value is positive, this returns ``1.0``. If both are zero, it returns ``0.0``. """ current_value = float(current) previous_value = float(previous) if abs(previous_value) <= _EPSILON: return 0.0 if abs(current_value) <= _EPSILON else 1.0 return abs(current_value - previous_value) / abs(previous_value) def _clean_previous_values(previous_values: Mapping[str, float] | None) -> dict[str, float]: """Return cleaned previous values.""" if previous_values is None: return {} return { str(label): float(value) for label, value in previous_values.items() if _is_number(value) and math.isfinite(float(value)) } def _compute_deltas( current_values: Mapping[str, float], previous_values: Mapping[str, float], ) -> dict[str, float]: """Return current-minus-previous deltas over the union of group labels.""" if not previous_values: return {} labels = sorted(set(current_values.keys()) | set(previous_values.keys())) return { label: float(current_values.get(label, 0.0) - previous_values.get(label, 0.0)) for label in labels } def _temporal_statistics( deltas: Mapping[str, float], previous_values: Mapping[str, float], ) -> dict[str, float]: """Return temporal statistics from group deltas.""" if not deltas: return { "temporal_variance": 0.0, "temporal_standard_deviation": 0.0, "total_absolute_delta": 0.0, "mean_absolute_delta": 0.0, "mean_relative_delta": 0.0, } delta_array = np.asarray(list(deltas.values()), dtype=float) absolute_deltas = np.abs(delta_array) labels = list(deltas.keys()) relative_deltas = [ relative_delta( current=previous_values.get(label, 0.0) + deltas[label], previous=previous_values.get(label, 0.0), ) for label in labels ] temporal_variance = float(np.var(delta_array)) temporal_standard_deviation = float(np.sqrt(temporal_variance)) return { "temporal_variance": temporal_variance, "temporal_standard_deviation": temporal_standard_deviation, "total_absolute_delta": float(np.sum(absolute_deltas)), "mean_absolute_delta": float(np.mean(absolute_deltas)), "mean_relative_delta": float(np.mean(np.asarray(relative_deltas, dtype=float))) if relative_deltas else 0.0, } def _stability_score( *, coefficient_of_variation: float, normalized_range: float, mean_relative_delta: float, ) -> float: """Return a bounded stability score in [0, 1].""" instability_pressure = ( max(0.0, float(coefficient_of_variation)) + max(0.0, float(normalized_range)) + max(0.0, float(mean_relative_delta)) ) return float(1.0 / (1.0 + instability_pressure)) def _numeric_array(values: Sequence[float]) -> np.ndarray: """Convert a sequence to a finite numeric array.""" array = np.asarray(values, dtype=float).reshape(-1) array = np.nan_to_num(array, nan=0.0, posinf=0.0, neginf=0.0) return array def _normalize_collection(value: StabilityCollection | str) -> StabilityCollection: """Normalize a stability collection value.""" if isinstance(value, StabilityCollection): return value return StabilityCollection(str(value)) def _normalize_aggregation(value: StabilityAggregation | str) -> StabilityAggregation: """Normalize a stability aggregation value.""" if isinstance(value, StabilityAggregation): return value return StabilityAggregation(str(value)) def _world_step(world: World) -> int | None: """Return the current world step if available.""" value = getattr(world, "step_count", None) if isinstance(value, Real) and not isinstance(value, bool): return int(value) return None def _is_number(value: Any) -> bool: """Return whether a value is a real numeric scalar, excluding booleans.""" return isinstance(value, (Real, np.integer, np.floating)) and not isinstance(value, bool) def _is_alive(item: Any) -> bool: """Return whether an item is alive when it exposes an ``alive`` field.""" return bool(getattr(item, "alive", True)) def _is_agent_like(item: Any) -> bool: """Return whether an item looks like a runtime agent.""" return hasattr(item, "alive") or hasattr(item, "behaviors") or hasattr(item, "policy") def _item_collection_name(item: Any) -> str: """Return a best-effort collection name for an item.""" if _is_agent_like(item): return StabilityCollection.AGENTS.value if hasattr(item, "amount"): return StabilityCollection.RESOURCES.value return "items" def _iter_world_items(world: World, collection: StabilityCollection) -> tuple[Any, ...]: """Return world items for a configured stability collection.""" if collection is StabilityCollection.AGENTS: return _iter_collection(getattr(world, "agents", ())) if collection is StabilityCollection.RESOURCES: return _iter_collection(getattr(world, "resources", ())) agents = _iter_collection(getattr(world, "agents", ())) resources = _iter_collection(getattr(world, "resources", ())) return agents + resources def _iter_collection(raw_collection: Any) -> tuple[Any, ...]: """Return items from a mapping-backed or sequence-backed collection.""" if raw_collection is None: return () if isinstance(raw_collection, Mapping): values = raw_collection.values() elif isinstance(raw_collection, Iterable) and not isinstance(raw_collection, (str, bytes)): values = raw_collection else: values = (raw_collection,) return tuple(item for item in values if item is not None) def _split_path(path: str) -> tuple[str, ...]: """Split a dot-separated path into components.""" return tuple(part for part in str(path).split(".") if part) def _get_mapping_path(container: Mapping[str, Any], path: str, default: Any = _MISSING) -> Any: """Read a nested mapping value using dot notation.""" parts = _split_path(path) if not parts: return default current: Any = container for part in parts: if not isinstance(current, Mapping) or part not in current: return default current = current[part] return current def _get_object_path(root: Any, path: str, default: Any = _MISSING) -> Any: """Read nested values from mappings, sequences, or object attributes.""" parts = _split_path(path) if not parts: return root current: Any = root for part in parts: if isinstance(current, Mapping): if part not in current: return default current = current[part] continue if isinstance(current, Sequence) and not isinstance(current, (str, bytes)) and part.isdigit(): index = int(part) if index >= len(current): return default current = current[index] continue if not hasattr(current, part): return default current = getattr(current, part) return current def _read_path(item: Any, path: str, default: Any = _MISSING) -> Any: """Read a path from an item. Supported special prefixes: - ``state.foo`` - ``state:foo`` - ``memory.foo`` - ``memory:foo`` - ``metadata.foo`` - ``metadata:foo`` Without a prefix, object attributes and mapping keys are read directly. """ normalized_path = str(path) if normalized_path.startswith("state."): state = getattr(item, "state", {}) return _get_mapping_path( state if isinstance(state, Mapping) else {}, normalized_path.removeprefix("state."), default, ) if normalized_path.startswith("state:"): state = getattr(item, "state", {}) return _get_mapping_path( state if isinstance(state, Mapping) else {}, normalized_path.removeprefix("state:"), default, ) if normalized_path.startswith("memory."): memory = getattr(item, "memory", {}) return _get_mapping_path( memory if isinstance(memory, Mapping) else {}, normalized_path.removeprefix("memory."), default, ) if normalized_path.startswith("memory:"): memory = getattr(item, "memory", {}) return _get_mapping_path( memory if isinstance(memory, Mapping) else {}, normalized_path.removeprefix("memory:"), default, ) if normalized_path.startswith("metadata."): metadata = getattr(item, "metadata", {}) return _get_mapping_path( metadata if isinstance(metadata, Mapping) else {}, normalized_path.removeprefix("metadata."), default, ) if normalized_path.startswith("metadata:"): metadata = getattr(item, "metadata", {}) return _get_mapping_path( metadata if isinstance(metadata, Mapping) else {}, normalized_path.removeprefix("metadata:"), default, ) return _get_object_path(item, normalized_path, default) def _stable_label(value: Any) -> str: """Return a stable string label for a group value.""" if value is None: return "none" if isinstance(value, bool): return "true" if value else "false" if _is_number(value): numeric_value = float(value) if numeric_value.is_integer(): return str(int(numeric_value)) return str(numeric_value) if isinstance(value, Mapping): parts = [f"{key}={_stable_label(value[key])}" for key in sorted(value.keys(), key=str)] return "{" + ",".join(parts) + "}" if isinstance(value, Sequence) and not isinstance(value, (str, bytes)): return "[" + ",".join(_stable_label(item) for item in value) + "]" return str(value) def _normalize_label(value: Any, *, case_sensitive: bool, strip: bool) -> str: """Normalize a group label for deterministic comparisons.""" label = _stable_label(value) if strip: label = label.strip() if not case_sensitive: label = label.lower() return label def _append_bounded(items: MutableSequence[Any], value: Any, max_items: int) -> None: """Append an item while enforcing an optional maximum history length.""" items.append(value) if max_items > 0 and len(items) > max_items: del items[: len(items) - max_items] METRIC_REGISTRY: Mapping[str, type[StabilityMetric]] = MappingProxyType( { StabilityMetric.name: StabilityMetric, } ) __all__ = [ "METRIC_REGISTRY", "StabilityAggregation", "StabilityCollection", "StabilityMetric", "StabilityResult", "StabilityTracker", "coefficient_of_variation", "compute_agent_population_stability", "compute_resource_amount_stability", "compute_stability", "population_variance", "relative_delta", "stability_from_values", ]