Source code for mercurial.core.patterns

"""Informational pattern definition and operations (MERCURIAL A.3)."""

from dataclasses import dataclass, field
from typing import Callable, Dict, List, Optional, Tuple

import numpy as np

from mercurial.hierarchy.complexity import ComplexityMeasure


[docs] @dataclass class Constraints: """Constraint set C.""" equality: List[Callable[[np.ndarray], float]] = field(default_factory=list) inequality: List[Callable[[np.ndarray], float]] = field(default_factory=list)
[docs] def evaluate(self, v: np.ndarray) -> Dict[str, float]: results = {} for i, c in enumerate(self.equality): results[f"eq_{i}"] = c(v) for i, c in enumerate(self.inequality): results[f"ineq_{i}"] = c(v) return results
[docs] def violation_norm(self, v: np.ndarray) -> float: eq_violation = sum(c(v) ** 2 for c in self.equality) ineq_violation = sum(max(0, c(v)) ** 2 for c in self.inequality) return np.sqrt(eq_violation + ineq_violation)
[docs] @dataclass class StabilityRegime: """Stability regime R (MERCURIAL A.3.1).""" attractor_basin: List[np.ndarray] = field(default_factory=list) decay_rate: float = 1e-3 # γ [s⁻¹] resonance_threshold: float = 0.5 # θ_r lyapunov_exponents: Optional[np.ndarray] = None
[docs] def is_attractor(self, v: np.ndarray, tol: float = 1e-6) -> bool: if not self.attractor_basin: return False distances = [np.linalg.norm(v - a) for a in self.attractor_basin] return min(distances) < tol
[docs] class Pattern: """Informational pattern P = (V, C, R).""" def __init__( self, state_variables: np.ndarray, constraints: Constraints, stability: StabilityRegime, label: str = "", impression_intensity: float = 0.0, ): self.V = state_variables # set of admissible configurations self.C = constraints self.R = stability self.label = label self.impression_intensity = impression_intensity
[docs] def free_energy(self, temperature: float = 1.0) -> float: """ F(P) = E(P) - T * S_gen with E = constraint violation norm, S_gen from entropy module. """ from mercurial.core.entropy import GeneralizedEntropy # Internal energy = constraint violation norm E = self.C.violation_norm(self.V) # Generalized entropy (simplified: use information entropy as proxy) info_entropy = self.information_content() therm_entropy = 0.0 # placeholder – can be extended ph_disorder = 1.0 - self.coherence() entropy_calc = GeneralizedEntropy() S_gen = entropy_calc.compute(info_entropy, therm_entropy, ph_disorder) return E - temperature * S_gen
[docs] def information_content(self) -> float: """I(P) = -∫ ρ log₂ ρ dV.""" if len(self.V) == 0: return 0.0 try: # Multi-dimensional histogram hist, _ = np.histogramdd(self.V, bins=10, density=True) except (ValueError, IndexError): # Fallback to 1D hist, _ = np.histogram(self.V, bins=10, density=True) hist = hist[hist > 0] return -np.sum(hist * np.log2(hist))
[docs] def coherence(self) -> float: """ Pattern coherence metric (0 = noisy, 1 = perfectly coherent). Based on cross‑modal synchronization and temporal stability. """ # Simplified: use inverse of entropy (normalised) I = self.information_content() # Max possible info for this dimension (uniform distribution) max_info = np.log2(10) # 10 bins return 1.0 - min(1.0, I / max_info)
[docs] def persistence_probability(self, delta_t: float, k_eff: float = 1.0) -> float: """ P_persist = exp(-ΔS_gen / k_eff), with ΔS_gen ≥ 0. Uses a simple entropy estimate (variance of state variables) to guarantee non‑negativity. """ # Entropy estimate: variance of flattened state (always ≥ 0) S_initial = np.var(self.V) # Simulate a small increase in entropy over time (decay) S_final = S_initial * (1 + 0.01 * delta_t) delta_S = max(0.0, S_final - S_initial) return np.exp(-delta_S / k_eff)
[docs] @classmethod def simple_gaussian( cls, dimension: int, mean: float = 0.0, std: float = 1.0, label: str = "" ) -> "Pattern": """Create a test pattern with Gaussian state variables.""" V = np.random.normal(mean, std, size=(100, dimension)) constraints = Constraints() stability = StabilityRegime(decay_rate=0.001) return cls(V, constraints, stability, label)
[docs] def update_impression(self, dt: float, arousal: float, event_occurred: bool, focused: bool): from mercurial.impressions.dynamic_formation import DynamicImpressionFormation if not hasattr(self, "_impression_dynamics"): self._impression_dynamics = DynamicImpressionFormation() self._impression_dynamics.impression_intensity = self.impression_intensity intensity, _ = self._impression_dynamics.update(dt, arousal, event_occurred, focused) self.impression_intensity = intensity return intensity
[docs] def complexity(self, measure: Optional["ComplexityMeasure"] = None) -> float: """Compute Λ(P).""" from mercurial.hierarchy.complexity import ComplexityMeasure if measure is None: measure = ComplexityMeasure() return measure.compute(self.V)
[docs] def level(self) -> Tuple[int, str]: """Return LADDER level and name for this pattern.""" from mercurial.hierarchy.complexity import LevelClassifier comp = self.complexity() return LevelClassifier.classify(comp)
# ============================================================================ # Neural Pattern using Wilson‑Cowan dynamics # ============================================================================ from mercurial.core.wilson_cowan import WilsonCowanPopulation
[docs] class NeuralPattern(Pattern): """ Pattern defined by neural population activity (Wilson‑Cowan). """ def __init__(self, wc: WilsonCowanPopulation, label: str = ""): # Initialize with dummy state variables (will be overwritten) super().__init__(np.array([[0.0]]), Constraints(), StabilityRegime(), label) self.wc = wc self.E_history = None self.I_history = None self.current_time = 0.0
[docs] def evolve(self, dt: float, n_steps: int, P_ext: float = 0.0, Q_ext: float = 0.0): """Evolve the neural pattern.""" E, I = self.wc.evolve(dt, n_steps, P_ext, Q_ext, initial_state=self._get_last_state()) self.E_history = E self.I_history = I self.current_time += n_steps * dt # Update state variables for compatibility with pattern methods self.V = np.column_stack([E, I]) # each row is (E, I) at a time step
def _get_last_state(self) -> Optional[np.ndarray]: if self.E_history is not None and len(self.E_history) > 0: return np.array([self.E_history[-1], self.I_history[-1]]) return None
[docs] def information_content(self) -> float: """Compute spectral entropy of neural activity.""" if self.E_history is None or len(self.E_history) < 10: return 0.0 # Power spectrum via FFT fft_vals = np.fft.fft(self.E_history - np.mean(self.E_history)) power = np.abs(fft_vals[: len(fft_vals) // 2]) ** 2 power = power / (np.sum(power) + 1e-12) entropy = -np.sum(power * np.log2(power + 1e-12)) return entropy
[docs] def coherence(self) -> float: """Return synchrony measure (order parameter R if multiple populations, else 0).""" # For single population, use variance of activity as proxy if self.E_history is None: return 0.0 return 1.0 - np.var(self.E_history)
# ======================================================================== # External input update for Wilson‑Cowan (appended) # ========================================================================
[docs] def set_external_inputs(self, P_ext: float, Q_ext: float): """ Set external inputs to the Wilson‑Cowan population. Called by the simulation engine at each step (for biasing). """ self._P_ext = P_ext self._Q_ext = Q_ext
[docs] def get_external_inputs(self): return getattr(self, "_P_ext", 0.0), getattr(self, "_Q_ext", 0.0)