""" Composite interestingness metrics for WorldSmithAI. This module computes a generic, explainable interestingness score for arbitrary agent-based worlds. It deliberately avoids domain-specific assumptions about ecosystems, farms, civilizations, research labs, markets, transport systems, power grids, or fantasy worlds. Interestingness is modeled as a weighted composite of normalized signals: - diversity: many groups coexist. - entropy: group distribution is not overly concentrated. - stability: the world is not collapsing into extreme volatility. - change: the world evolves over time. - novelty: current distribution differs from a reference distribution. - activity: agents/resources/events are producing observable traces. - balance: no single group fully dominates. Example: metric = InterestingnessMetric() result = metric.compute(world) print(result.score) print(result.level) print(result.component_scores) Future extensibility: - Add narrative surprise and semantic novelty from narrator outputs. - Add event-system activity once events become first-class logs. - Add behavior execution traces from the scheduler. - Add God-Agent world evaluation using this score as one feature. - Add multi-run comparison and Pareto ranking. - Add automatic tuning of component weights. """ 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 from metrics.diversity import DiversityMetric, DiversityResult from metrics.entropy import EntropyMetric, EntropyResult from metrics.stability import StabilityAggregation, StabilityMetric, StabilityResult if TYPE_CHECKING: from core.world import World logger = logging.getLogger(__name__) _MISSING = object() _EPSILON = 1.0e-12 class InterestingnessCollection(str, Enum): """Collections over which interestingness may be computed.""" AGENTS = "agents" RESOURCES = "resources" BOTH = "both" class InterestingnessLevel(str, Enum): """Human-readable interestingness bands.""" EMPTY = "empty" LOW = "low" MEDIUM = "medium" HIGH = "high" EXCEPTIONAL = "exceptional" class InterestingnessComponent(str, Enum): """Supported normalized interestingness components.""" DIVERSITY = "diversity" ENTROPY = "entropy" STABILITY = "stability" CHANGE = "change" NOVELTY = "novelty" ACTIVITY = "activity" BALANCE = "balance" @dataclass(frozen=True) class InterestingnessResult: """Result produced by composite interestingness computation. Attributes: metric_name: Stable metric name. score: Final weighted score in [0, 1]. level: Human-readable score band. component_scores: Normalized score per component. component_weights: Effective normalized weight per component. raw_components: Raw metric outputs or intermediate values. explanations: Short explanation per component. collection: Collection analyzed. group_by_path: Path used to group world objects. item_count: Number of included items when available. step: Optional world step. metadata: Additional JSON-compatible details. """ metric_name: str score: float level: InterestingnessLevel component_scores: Mapping[str, float] = field(default_factory=dict) component_weights: Mapping[str, float] = field(default_factory=dict) raw_components: Mapping[str, Any] = field(default_factory=dict) explanations: Mapping[str, str] = field(default_factory=dict) collection: str = "agents" group_by_path: str = "type" item_count: int = 0 step: int | None = None metadata: Mapping[str, Any] = field(default_factory=dict) @property def is_empty(self) -> bool: """Return whether this result was computed from an empty world slice.""" return self.level is InterestingnessLevel.EMPTY or self.item_count == 0 def to_dict(self) -> dict[str, Any]: """Return a JSON-friendly representation of the result.""" return { "metric_name": self.metric_name, "score": self.score, "level": self.level.value, "component_scores": copy.deepcopy(dict(self.component_scores)), "component_weights": copy.deepcopy(dict(self.component_weights)), "raw_components": _json_safe(copy.deepcopy(dict(self.raw_components))), "explanations": copy.deepcopy(dict(self.explanations)), "collection": self.collection, "group_by_path": self.group_by_path, "item_count": self.item_count, "is_empty": self.is_empty, "step": self.step, "metadata": _json_safe(copy.deepcopy(dict(self.metadata))), } @dataclass class InterestingnessMetric: """Compute an explainable composite interestingness score. The default configuration computes interestingness over alive agents grouped by ``type``. This is generic: ``type`` can mean species, profession, role, faction, node class, market role, research discipline, or any DSL-defined label. The metric combines sub-metrics from diversity, entropy, and stability with optional temporal and reference-distribution signals. """ name: ClassVar[str] = "interestingness" collection: InterestingnessCollection | str = InterestingnessCollection.AGENTS group_by_path: str = "type" value_path: str | None = None weight_path: str | None = None alive_only: bool = True include_missing: bool = False missing_label: str = "unknown" include_activity: bool = True activity_paths: tuple[str, ...] = ( "memory.policy_decisions", "memory.bandit_decisions", "memory.market_trades", "memory.construction_history", "memory.adoption_history", "memory.planning_history", "memory.memory_history", "memory.inbox", "memory.outbox", ) activity_saturation: float = 25.0 component_weights: Mapping[str, float] = field( default_factory=lambda: { InterestingnessComponent.DIVERSITY.value: 0.20, InterestingnessComponent.ENTROPY.value: 0.20, InterestingnessComponent.STABILITY.value: 0.15, InterestingnessComponent.CHANGE.value: 0.20, InterestingnessComponent.NOVELTY.value: 0.15, InterestingnessComponent.ACTIVITY.value: 0.05, InterestingnessComponent.BALANCE.value: 0.05, } ) change_saturation: float = 10.0 novelty_reference_smoothing: float = 0.0 minimum_items_for_medium: int = 2 metadata: Mapping[str, Any] = field(default_factory=dict) def compute( self, world: World, *, previous_group_values: Mapping[str, float] | None = None, reference_distribution: Mapping[str, float] | None = None, ) -> InterestingnessResult: """Compute interestingness for the current world. Args: world: Runtime world object. previous_group_values: Optional previous grouped values used to compute temporal change. reference_distribution: Optional baseline distribution used to compute novelty. Returns: ``InterestingnessResult`` with final score and component breakdown. """ collection = _normalize_collection(self.collection) diversity = self._compute_diversity(world, collection) entropy = self._compute_entropy(world, collection) stability = self._compute_stability(world, collection, previous_group_values) item_count = max(diversity.item_count, entropy.item_count, stability.item_count) if item_count == 0: return self._empty_result(world, collection) current_distribution = _distribution_from_values(stability.group_values) previous_distribution = ( _distribution_from_values(previous_group_values) if previous_group_values else {} ) component_scores: dict[str, float] = {} raw_components: dict[str, Any] = {} explanations: dict[str, str] = {} diversity_score = _clamp_unit(diversity.gini_simpson_index) component_scores[InterestingnessComponent.DIVERSITY.value] = diversity_score raw_components[InterestingnessComponent.DIVERSITY.value] = diversity.to_dict() explanations[InterestingnessComponent.DIVERSITY.value] = ( f"Group richness is {diversity.richness}; " f"Gini-Simpson diversity is {diversity_score:.3f}." ) entropy_score = _clamp_unit(entropy.normalized_entropy) component_scores[InterestingnessComponent.ENTROPY.value] = entropy_score raw_components[InterestingnessComponent.ENTROPY.value] = entropy.to_dict() explanations[InterestingnessComponent.ENTROPY.value] = ( f"Normalized entropy is {entropy_score:.3f}; " f"dominant outcome is {entropy.dominant_outcome!r}." ) stability_score = _clamp_unit(stability.stability_score) component_scores[InterestingnessComponent.STABILITY.value] = stability_score raw_components[InterestingnessComponent.STABILITY.value] = stability.to_dict() explanations[InterestingnessComponent.STABILITY.value] = ( f"Stability score is {stability_score:.3f}; " f"coefficient of variation is {stability.coefficient_of_variation:.3f}." ) change_score = self._change_score(stability) component_scores[InterestingnessComponent.CHANGE.value] = change_score raw_components[InterestingnessComponent.CHANGE.value] = { "deltas": copy.deepcopy(dict(stability.deltas)), "total_absolute_delta": stability.total_absolute_delta, "mean_absolute_delta": stability.mean_absolute_delta, "mean_relative_delta": stability.mean_relative_delta, } explanations[InterestingnessComponent.CHANGE.value] = ( f"Temporal change score is {change_score:.3f}; " f"mean relative delta is {stability.mean_relative_delta:.3f}." ) novelty_score = self._novelty_score( current_distribution=current_distribution, previous_distribution=previous_distribution, reference_distribution=reference_distribution, ) component_scores[InterestingnessComponent.NOVELTY.value] = novelty_score raw_components[InterestingnessComponent.NOVELTY.value] = { "current_distribution": current_distribution, "previous_distribution": previous_distribution, "reference_distribution": copy.deepcopy(dict(reference_distribution or {})), "jensen_shannon_divergence": novelty_score, } explanations[InterestingnessComponent.NOVELTY.value] = ( f"Novelty score is {novelty_score:.3f} against the available baseline." ) activity_score, activity_raw = self._activity_score(world, collection) component_scores[InterestingnessComponent.ACTIVITY.value] = activity_score raw_components[InterestingnessComponent.ACTIVITY.value] = activity_raw explanations[InterestingnessComponent.ACTIVITY.value] = ( f"Activity score is {activity_score:.3f} from " f"{activity_raw.get('activity_count', 0)} observed activity record(s)." ) balance_score = self._balance_score(diversity, entropy) component_scores[InterestingnessComponent.BALANCE.value] = balance_score raw_components[InterestingnessComponent.BALANCE.value] = { "dominance": diversity.dominance, "dominant_group": diversity.dominant_group, "richness": diversity.richness, } explanations[InterestingnessComponent.BALANCE.value] = ( f"Balance score is {balance_score:.3f}; " f"dominance is {diversity.dominance:.3f}." ) effective_weights = _normalize_weights( self.component_weights, available_components=component_scores.keys(), ) score = _weighted_score(component_scores, effective_weights) if item_count < self.minimum_items_for_medium: score = min(score, 0.35) level = _interestingness_level(score, item_count=item_count) result = InterestingnessResult( metric_name=self.name, score=score, level=level, component_scores=component_scores, component_weights=effective_weights, raw_components=raw_components, explanations=explanations, collection=collection.value, group_by_path=self.group_by_path, item_count=item_count, step=_world_step(world), metadata={ **copy.deepcopy(dict(self.metadata)), "previous_values_provided": previous_group_values is not None, "reference_distribution_provided": reference_distribution is not None, }, ) logger.debug( "Computed interestingness over %s grouped by %s: score=%.3f level=%s", collection.value, self.group_by_path, result.score, result.level.value, ) return result def __call__( self, world: World, *, previous_group_values: Mapping[str, float] | None = None, reference_distribution: Mapping[str, float] | None = None, ) -> InterestingnessResult: """Compute interestingness, allowing metric instances to be called directly.""" return self.compute( world, previous_group_values=previous_group_values, reference_distribution=reference_distribution, ) def _compute_diversity( self, world: World, collection: InterestingnessCollection, ) -> DiversityResult: """Compute diversity sub-metric.""" metric = DiversityMetric( collection=collection.value, group_by_path=self.group_by_path, weight_path=self.weight_path, alive_only=self.alive_only, include_missing=self.include_missing, missing_label=self.missing_label, ) return metric.compute(world) def _compute_entropy( self, world: World, collection: InterestingnessCollection, ) -> EntropyResult: """Compute entropy sub-metric.""" metric = EntropyMetric( collection=collection.value, value_path=self.group_by_path, weight_path=self.weight_path, alive_only=self.alive_only, include_missing=self.include_missing, missing_label=self.missing_label, ) return metric.compute(world) def _compute_stability( self, world: World, collection: InterestingnessCollection, previous_group_values: Mapping[str, float] | None, ) -> StabilityResult: """Compute stability sub-metric.""" aggregation = ( StabilityAggregation.SUM if self.value_path is not None else StabilityAggregation.COUNT ) metric = StabilityMetric( collection=collection.value, group_by_path=self.group_by_path, value_path=self.value_path, weight_path=self.weight_path, aggregation=aggregation, alive_only=self.alive_only, include_missing=self.include_missing, missing_label=self.missing_label, ) return metric.compute(world, previous_values=previous_group_values) def _change_score(self, stability: StabilityResult) -> float: """Return normalized change score from temporal stability data.""" if not stability.deltas: return 0.0 if self.change_saturation <= _EPSILON: return _clamp_unit(stability.mean_relative_delta) combined_change = ( stability.mean_relative_delta + stability.mean_absolute_delta / max(float(self.change_saturation), _EPSILON) ) return _saturating_score(combined_change) def _novelty_score( self, *, current_distribution: Mapping[str, float], previous_distribution: Mapping[str, float], reference_distribution: Mapping[str, float] | None, ) -> float: """Return Jensen-Shannon novelty against reference or previous distribution.""" if reference_distribution: baseline = _distribution_from_values(reference_distribution) elif previous_distribution: baseline = previous_distribution else: return 0.0 if self.novelty_reference_smoothing > 0: baseline = _smooth_distribution( baseline, current_distribution, smoothing=float(self.novelty_reference_smoothing), ) return jensen_shannon_divergence(current_distribution, baseline) def _activity_score( self, world: World, collection: InterestingnessCollection, ) -> tuple[float, dict[str, Any]]: """Return activity score and raw activity diagnostics.""" if not self.include_activity: return 0.0, {"activity_count": 0, "enabled": False} items = _iter_world_items(world, collection) activity_count = 0 per_path: dict[str, int] = {} for item in items: if self.alive_only and _is_agent_like(item) and not _is_alive(item): continue for path in self.activity_paths: raw_value = _read_path(item, path, _MISSING) contribution = _activity_contribution(raw_value) if contribution <= 0: continue per_path[path] = per_path.get(path, 0) + contribution activity_count += contribution world_events = _activity_contribution(getattr(world, "events", None)) if world_events > 0: per_path["world.events"] = world_events activity_count += world_events score = _saturating_score(activity_count / max(float(self.activity_saturation), _EPSILON)) return score, { "activity_count": activity_count, "activity_by_path": per_path, "activity_saturation": self.activity_saturation, "enabled": True, } @staticmethod def _balance_score( diversity: DiversityResult, entropy: EntropyResult, ) -> float: """Return balance score from dominance and entropy.""" if diversity.is_empty: return 0.0 anti_dominance = 1.0 - _clamp_unit(diversity.dominance) entropy_balance = _clamp_unit(entropy.normalized_entropy) return _clamp_unit((anti_dominance + entropy_balance) / 2.0) def _empty_result( self, world: World, collection: InterestingnessCollection, ) -> InterestingnessResult: """Return an empty interestingness result.""" components = { component.value: 0.0 for component in InterestingnessComponent } weights = _normalize_weights(self.component_weights, available_components=components.keys()) return InterestingnessResult( metric_name=self.name, score=0.0, level=InterestingnessLevel.EMPTY, component_scores=components, component_weights=weights, raw_components={}, explanations={ component.value: "No world objects contributed to this component." for component in InterestingnessComponent }, collection=collection.value, group_by_path=self.group_by_path, item_count=0, step=_world_step(world), metadata=copy.deepcopy(dict(self.metadata)), ) @dataclass class InterestingnessTracker: """Track interestingness over time. The tracker stores previous grouped values so each update can compute temporal change and novelty without needing a separate history system. """ metric: InterestingnessMetric = field(default_factory=InterestingnessMetric) max_history: int = 1000 history: list[InterestingnessResult] = field(default_factory=list) previous_group_values: Mapping[str, float] = field(default_factory=dict) reference_distribution: Mapping[str, float] | None = None def update(self, world: World) -> InterestingnessResult: """Compute interestingness and append it to history.""" result = self.metric.compute( world, previous_group_values=self.previous_group_values, reference_distribution=self.reference_distribution, ) current_values = self._current_group_values(result) self.previous_group_values = current_values if self.reference_distribution is None and current_values: self.reference_distribution = _distribution_from_values(current_values) _append_bounded(self.history, result, self.max_history) return result def latest(self) -> InterestingnessResult | None: """Return the latest interestingness result, if any.""" if not self.history: return None return self.history[-1] def reset(self) -> None: """Clear tracked state.""" self.history.clear() self.previous_group_values = {} self.reference_distribution = None def to_series(self, value_key: str = "score") -> list[dict[str, Any]]: """Return a JSON-friendly time series for score or component values. Args: value_key: ``"score"`` or a component name such as ``"entropy"``. Returns: List of dictionaries with ``index``, ``step``, and ``value``. """ series: list[dict[str, Any]] = [] for index, result in enumerate(self.history): if value_key == "score": value = result.score else: value = result.component_scores.get(value_key) 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, }, "max_history": self.max_history, "previous_group_values": copy.deepcopy(dict(self.previous_group_values)), "reference_distribution": None if self.reference_distribution is None else copy.deepcopy(dict(self.reference_distribution)), "history": [result.to_dict() for result in self.history], } @staticmethod def _current_group_values(result: InterestingnessResult) -> dict[str, float]: """Extract current group values from a result's stability component.""" stability = result.raw_components.get(InterestingnessComponent.STABILITY.value) if not isinstance(stability, Mapping): return {} group_values = stability.get("group_values", {}) if not isinstance(group_values, Mapping): return {} return { str(key): float(value) for key, value in group_values.items() if _is_number(value) } def compute_interestingness( world: World, *, collection: InterestingnessCollection | str = InterestingnessCollection.AGENTS, group_by_path: str = "type", value_path: str | None = None, weight_path: str | None = None, previous_group_values: Mapping[str, float] | None = None, reference_distribution: Mapping[str, float] | None = None, ) -> InterestingnessResult: """Convenience function for computing world interestingness.""" metric = InterestingnessMetric( collection=collection, group_by_path=group_by_path, value_path=value_path, weight_path=weight_path, ) return metric.compute( world, previous_group_values=previous_group_values, reference_distribution=reference_distribution, ) def jensen_shannon_divergence( left_distribution: Mapping[str, float], right_distribution: Mapping[str, float], ) -> float: """Return normalized Jensen-Shannon divergence in [0, 1]. The computation uses log base 2, so the maximum divergence is 1. This makes it directly usable as a novelty score. """ left = _distribution_from_values(left_distribution) right = _distribution_from_values(right_distribution) if not left or not right: return 0.0 labels = sorted(set(left.keys()) | set(right.keys())) p = np.asarray([left.get(label, 0.0) for label in labels], dtype=float) q = np.asarray([right.get(label, 0.0) for label in labels], dtype=float) p = _normalize_probability_array(p) q = _normalize_probability_array(q) if p.size == 0 or q.size == 0: return 0.0 midpoint = 0.5 * (p + q) divergence = 0.5 * _kl_divergence(p, midpoint) + 0.5 * _kl_divergence(q, midpoint) return _clamp_unit(float(divergence)) def novelty_from_distributions( current_distribution: Mapping[str, float], reference_distribution: Mapping[str, float], ) -> float: """Return novelty score from two distributions.""" return jensen_shannon_divergence(current_distribution, reference_distribution) def weighted_component_score( component_scores: Mapping[str, float], component_weights: Mapping[str, float], ) -> float: """Return weighted score from normalized component scores and weights.""" weights = _normalize_weights(component_weights, available_components=component_scores.keys()) return _weighted_score(component_scores, weights) def _normalize_collection(value: InterestingnessCollection | str) -> InterestingnessCollection: """Normalize an interestingness collection value.""" if isinstance(value, InterestingnessCollection): return value return InterestingnessCollection(str(value)) def _interestingness_level(score: float, *, item_count: int) -> InterestingnessLevel: """Return a human-readable interestingness level for a score.""" if item_count <= 0: return InterestingnessLevel.EMPTY bounded_score = _clamp_unit(score) if bounded_score < 0.25: return InterestingnessLevel.LOW if bounded_score < 0.50: return InterestingnessLevel.MEDIUM if bounded_score < 0.75: return InterestingnessLevel.HIGH return InterestingnessLevel.EXCEPTIONAL def _normalize_weights( weights: Mapping[str, float], *, available_components: Iterable[str], ) -> dict[str, float]: """Normalize component weights over available components.""" available = tuple(str(component) for component in available_components) cleaned: dict[str, float] = {} for component in available: raw_weight = weights.get(component, 0.0) cleaned[component] = max(0.0, float(raw_weight)) if _is_number(raw_weight) else 0.0 total = sum(cleaned.values()) if total <= _EPSILON and available: equal = 1.0 / len(available) return {component: equal for component in available} if total <= _EPSILON: return {} return {component: weight / total for component, weight in cleaned.items()} def _weighted_score( component_scores: Mapping[str, float], component_weights: Mapping[str, float], ) -> float: """Return bounded weighted score.""" total = 0.0 for component, weight in component_weights.items(): score = component_scores.get(component, 0.0) if not _is_number(score): continue total += _clamp_unit(float(score)) * max(0.0, float(weight)) return _clamp_unit(total) def _distribution_from_values(values: Mapping[str, float] | None) -> dict[str, float]: """Normalize non-negative values into a probability distribution.""" if not values: return {} cleaned = { str(key): max(0.0, float(value)) for key, value in values.items() if _is_number(value) and math.isfinite(float(value)) and float(value) > 0.0 } total = sum(cleaned.values()) if total <= _EPSILON: return {} return { key: value / total for key, value in sorted(cleaned.items(), key=lambda item: item[0]) } def _smooth_distribution( baseline: Mapping[str, float], current: Mapping[str, float], *, smoothing: float, ) -> dict[str, float]: """Smooth a baseline distribution toward the current support.""" alpha = _clamp_unit(float(smoothing)) labels = sorted(set(baseline.keys()) | set(current.keys())) if not labels: return {} uniform = 1.0 / len(labels) smoothed = { label: (1.0 - alpha) * float(baseline.get(label, 0.0)) + alpha * uniform for label in labels } return _distribution_from_values(smoothed) def _normalize_probability_array(values: np.ndarray) -> np.ndarray: """Return a normalized non-negative probability array.""" cleaned = np.nan_to_num(values.astype(float), nan=0.0, posinf=0.0, neginf=0.0) cleaned = np.clip(cleaned, 0.0, None) total = float(np.sum(cleaned)) if total <= _EPSILON: return np.asarray([], dtype=float) return cleaned / total def _kl_divergence(p: np.ndarray, q: np.ndarray) -> float: """Return KL divergence with log base 2 and safe zero handling.""" mask = p > 0.0 if not np.any(mask): return 0.0 safe_p = p[mask] safe_q = np.clip(q[mask], _EPSILON, None) return float(np.sum(safe_p * np.log2(safe_p / safe_q))) def _saturating_score(value: float) -> float: """Return a smooth bounded score from a non-negative value.""" safe_value = max(0.0, float(value)) return float(1.0 - math.exp(-safe_value)) def _clamp_unit(value: float) -> float: """Clamp a numeric value to [0, 1].""" return min(max(float(value), 0.0), 1.0) 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 _iter_world_items(world: World, collection: InterestingnessCollection) -> tuple[Any, ...]: """Return world items for a configured interestingness collection.""" if collection is InterestingnessCollection.AGENTS: return _iter_collection(getattr(world, "agents", ())) if collection is InterestingnessCollection.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 _activity_contribution(value: Any) -> int: """Return a small generic activity count from a value.""" if value is _MISSING or value is None: return 0 if isinstance(value, Mapping): return len(value) if isinstance(value, Sequence) and not isinstance(value, (str, bytes)): return len(value) if isinstance(value, bool): return 1 if value else 0 if _is_number(value): return 1 if float(value) != 0.0 else 0 return 1 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 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 _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] def _json_safe(value: Any) -> Any: """Return a JSON-friendly copy of arbitrary metric data.""" if value is None or isinstance(value, (str, bool)): return value if _is_number(value): numeric_value = float(value) if not math.isfinite(numeric_value): return None if numeric_value.is_integer(): return int(numeric_value) return numeric_value if isinstance(value, Mapping): return {str(key): _json_safe(nested) for key, nested in value.items()} if isinstance(value, Sequence) and not isinstance(value, (str, bytes)): return [_json_safe(item) for item in value] if hasattr(value, "to_dict") and callable(value.to_dict): return _json_safe(value.to_dict()) return str(value) METRIC_REGISTRY: Mapping[str, type[InterestingnessMetric]] = MappingProxyType( { InterestingnessMetric.name: InterestingnessMetric, } ) __all__ = [ "InterestingnessCollection", "InterestingnessComponent", "InterestingnessLevel", "InterestingnessMetric", "InterestingnessResult", "InterestingnessTracker", "METRIC_REGISTRY", "compute_interestingness", "jensen_shannon_divergence", "novelty_from_distributions", "weighted_component_score", ]