|
| 1 | +"""Emergent blend detection. |
| 2 | +
|
| 3 | +Observes emotional states over time. When two or more emotions co-occur at |
| 4 | +high intensity repeatedly, the detector records them as a named blend. |
| 5 | +Names can be assigned later (once the shape is recognised). |
| 6 | +
|
| 7 | +Design per spec Section 5.2 (blend sub-module). Threshold tunable — current |
| 8 | +defaults: intensity ≥5 each, ≥5 co-occurrences to register. |
| 9 | +""" |
| 10 | + |
| 11 | +from __future__ import annotations |
| 12 | + |
| 13 | +from collections.abc import Iterable |
| 14 | +from dataclasses import dataclass, field |
| 15 | +from itertools import combinations |
| 16 | +from typing import Any |
| 17 | + |
| 18 | +from brain.emotion.state import EmotionalState |
| 19 | + |
| 20 | + |
| 21 | +@dataclass |
| 22 | +class DetectedBlend: |
| 23 | + """A repeatedly-observed co-occurrence of high-intensity emotions. |
| 24 | +
|
| 25 | + Attributes: |
| 26 | + components: Tuple of emotion names (sorted alphabetically for stable hashing). |
| 27 | + count: How many times this combination has been observed above threshold. |
| 28 | + name: Optional human-readable label, assigned via BlendDetector.name_blend(). |
| 29 | + """ |
| 30 | + |
| 31 | + components: tuple[str, ...] |
| 32 | + count: int |
| 33 | + name: str | None = None |
| 34 | + |
| 35 | + |
| 36 | +@dataclass |
| 37 | +class BlendDetector: |
| 38 | + """Tracks emotional co-occurrences to surface emergent patterns. |
| 39 | +
|
| 40 | + Attributes: |
| 41 | + intensity_threshold: Minimum per-emotion intensity to count an observation. |
| 42 | + detection_threshold: Minimum observations to register the pattern. |
| 43 | + _observations: Internal count map (components tuple → count). |
| 44 | + _names: Internal name map (components tuple → name). |
| 45 | + """ |
| 46 | + |
| 47 | + intensity_threshold: float = 5.0 |
| 48 | + detection_threshold: int = 5 |
| 49 | + # Private state populated by observe() / name_blend() / from_dict(). |
| 50 | + # init=False keeps the public constructor surface minimal — callers |
| 51 | + # shouldn't be able to inject arbitrary counts via kwargs. |
| 52 | + _observations: dict[tuple[str, ...], int] = field(default_factory=dict, init=False) |
| 53 | + _names: dict[tuple[str, ...], str | None] = field(default_factory=dict, init=False) |
| 54 | + |
| 55 | + def observe(self, state: EmotionalState) -> None: |
| 56 | + """Record the high-intensity emotion combinations from the given state.""" |
| 57 | + high = tuple( |
| 58 | + sorted( |
| 59 | + name |
| 60 | + for name, intensity in state.emotions.items() |
| 61 | + if intensity >= self.intensity_threshold |
| 62 | + ) |
| 63 | + ) |
| 64 | + if len(high) < 2: |
| 65 | + return |
| 66 | + |
| 67 | + # Track every pair and every triple. Cap subset size at 3 — with |
| 68 | + # typically ≤6 high-intensity emotions at once, C(6,3)=20 but |
| 69 | + # C(6,4)=15, so 4-way tracking is affordable, but 3-way is where |
| 70 | + # the meaningful emergent patterns ("building_love", "creative_feral") |
| 71 | + # actually live. Higher orders are diluted. Bump if that assumption |
| 72 | + # breaks as real data accrues. |
| 73 | + for size in (2, 3): |
| 74 | + if size > len(high): |
| 75 | + break |
| 76 | + for combo in combinations(high, size): |
| 77 | + self._observations[combo] = self._observations.get(combo, 0) + 1 |
| 78 | + |
| 79 | + def detected(self) -> list[DetectedBlend]: |
| 80 | + """Return every combination that has crossed the detection threshold.""" |
| 81 | + result = [] |
| 82 | + for components, count in self._observations.items(): |
| 83 | + if count >= self.detection_threshold: |
| 84 | + result.append( |
| 85 | + DetectedBlend( |
| 86 | + components=components, |
| 87 | + count=count, |
| 88 | + name=self._names.get(components), |
| 89 | + ) |
| 90 | + ) |
| 91 | + return result |
| 92 | + |
| 93 | + def name_blend(self, components: Iterable[str], name: str) -> None: |
| 94 | + """Assign a human-readable name to a previously-detected blend. |
| 95 | +
|
| 96 | + Raises: |
| 97 | + KeyError: if the given components haven't been detected yet. |
| 98 | + """ |
| 99 | + key = tuple(sorted(components)) |
| 100 | + if key not in self._observations or self._observations[key] < self.detection_threshold: |
| 101 | + raise KeyError(f"Blend {key!r} has not been detected yet") |
| 102 | + self._names[key] = name |
| 103 | + |
| 104 | + def to_dict(self) -> dict[str, Any]: |
| 105 | + """Serialise to a plain-dict form suitable for JSON.""" |
| 106 | + return { |
| 107 | + "intensity_threshold": self.intensity_threshold, |
| 108 | + "detection_threshold": self.detection_threshold, |
| 109 | + "observations": [ |
| 110 | + {"components": list(k), "count": v} for k, v in self._observations.items() |
| 111 | + ], |
| 112 | + "names": [{"components": list(k), "name": v} for k, v in self._names.items()], |
| 113 | + } |
| 114 | + |
| 115 | + @classmethod |
| 116 | + def from_dict(cls, data: dict[str, Any]) -> BlendDetector: |
| 117 | + """Restore from a dict produced by to_dict.""" |
| 118 | + detector = cls( |
| 119 | + intensity_threshold=float(data.get("intensity_threshold", 5.0)), |
| 120 | + detection_threshold=int(data.get("detection_threshold", 5)), |
| 121 | + ) |
| 122 | + # Sort component lists into canonical tuple order on the way in — |
| 123 | + # observe() and name_blend() both operate on sorted keys, so any |
| 124 | + # caller-supplied JSON with unsorted components would otherwise |
| 125 | + # land at a non-matching key and cause spurious KeyErrors later. |
| 126 | + for entry in data.get("observations", []): |
| 127 | + key = tuple(sorted(entry["components"])) |
| 128 | + detector._observations[key] = int(entry["count"]) |
| 129 | + for entry in data.get("names", []): |
| 130 | + name = entry.get("name") |
| 131 | + if name is None: |
| 132 | + continue |
| 133 | + key = tuple(sorted(entry["components"])) |
| 134 | + detector._names[key] = name |
| 135 | + return detector |
0 commit comments