diff --git a/sensorium/audio/__init__.py b/sensorium/audio/__init__.py new file mode 100644 index 00000000..9a360fb2 --- /dev/null +++ b/sensorium/audio/__init__.py @@ -0,0 +1,44 @@ +""" +sensorium.audio — CORE-native deterministic audio compiler (ADR-0181, PR-2). + +Audio enters CORE as a compiler, not an embedding bridge: canonical waveform +→ typed AudioIR → (32,) float32 Cl(4,1) versor, fully deterministic and +replayable. Each compiled chunk is one AudioCompilationUnit — the Delta-CRDT +delta the audio adapter writes into its thread-local arena (ADR-0181 §2.1). + +PR-2 ships the deterministic substrate only. Pack artifacts + the +AudioProjectionHead adapter land in PR-3; evals in PR-4; CRDT wiring in PR-5. +""" + +from sensorium.audio.compiler import AudioCompiler, compile_events +from sensorium.audio.operators import ( + DEFAULT_OPERATOR_REGISTRY, + AudioOperatorRegistry, + OperatorSpec, + build_elliptic_rotor, +) +from sensorium.audio.trace import audio_evidence_trace +from sensorium.audio.types import ( + AudioCompilationUnit, + AudioIR, + AudioSignal, + AudioToken, + AuditoryEvent, + PitchCandidate, +) + +__all__ = [ + "AudioCompiler", + "compile_events", + "AudioOperatorRegistry", + "OperatorSpec", + "DEFAULT_OPERATOR_REGISTRY", + "build_elliptic_rotor", + "audio_evidence_trace", + "AudioCompilationUnit", + "AudioIR", + "AudioSignal", + "AudioToken", + "AuditoryEvent", + "PitchCandidate", +] diff --git a/sensorium/audio/canonical.py b/sensorium/audio/canonical.py new file mode 100644 index 00000000..e6222dc6 --- /dev/null +++ b/sensorium/audio/canonical.py @@ -0,0 +1,74 @@ +""" +sensorium/audio/canonical.py — canonical signal formation (spec §3). + +Default: mono, 24 kHz, float32 internal-as-float64 compute. The original +source bytes are hashed for provenance (``source_sha256``); the canonical +float32 image is hashed as ``canonical_sha256``. Resampling, when needed, uses +the pinned polyphase FIR from the pack (PR-3); PR-2 supports the same-rate +passthrough path and resampling when explicit FIR taps are supplied. +""" + +from __future__ import annotations + +import numpy as np + +from sensorium.audio.checksum import sha256_array +from sensorium.audio.resample import needs_resample, resample_poly +from sensorium.audio.types import AudioSignal + +CANONICAL_SAMPLE_RATE = 24_000 + + +def to_mono(samples: np.ndarray) -> np.ndarray: + """Deterministic mono downmix: average across channels if multi-channel. + + Accepts (N,) mono, (N, C) or (C, N) interleaved/planar. Channel axis is + the smaller of the two dimensions (audio has more samples than channels). + """ + arr = np.asarray(samples, dtype=np.float64) + if arr.ndim == 1: + return arr + if arr.ndim != 2: + raise ValueError(f"expected 1-D or 2-D samples, got ndim={arr.ndim}") + channel_axis = 0 if arr.shape[0] < arr.shape[1] else 1 + return arr.mean(axis=channel_axis) + + +def canonicalize( + samples: np.ndarray, + sample_rate: int, + *, + target_sr: int = CANONICAL_SAMPLE_RATE, + fir: np.ndarray | None = None, + start_ms: int = 0, +) -> AudioSignal: + """Produce a canonical mono float32 ``AudioSignal`` with provenance hashes. + + ``source_sha256`` hashes the original input bytes exactly as received; + ``canonical_sha256`` hashes the canonical float32 image. Resampling to + ``target_sr`` requires explicit ``fir`` taps (the pinned pack artifact); + same-rate input is an exact passthrough. + """ + source_sha256 = sha256_array(np.asarray(samples, dtype=np.float32)) + mono = to_mono(samples) + + if needs_resample(sample_rate, target_sr): + if fir is None: + raise ValueError( + f"resampling {sample_rate}->{target_sr} requires explicit FIR taps " + "(pinned pack artifact, PR-3); none supplied" + ) + from math import gcd + g = gcd(target_sr, sample_rate) + mono = resample_poly(mono, up=target_sr // g, down=sample_rate // g, fir=fir) + + canonical = np.ascontiguousarray(mono, dtype=np.float32) + duration_ms = int(round(1000 * canonical.size / target_sr)) + return AudioSignal( + samples=canonical, + sample_rate=target_sr, + start_ms=start_ms, + end_ms=start_ms + duration_ms, + source_sha256=source_sha256, + canonical_sha256=sha256_array(canonical), + ) diff --git a/sensorium/audio/checksum.py b/sensorium/audio/checksum.py new file mode 100644 index 00000000..f45c4bb4 --- /dev/null +++ b/sensorium/audio/checksum.py @@ -0,0 +1,38 @@ +""" +sensorium/audio/checksum.py — the layered checksum chain (ADR-0181 §2.2, spec §6). + + source_sha256 → canonical_sha256 → token_stream_sha256 → ir_sha256 + → pack_manifest_sha256 → projection_sha256 + +Every link is content-addressed. The merge key +(canonical_sha256, ir_sha256, projection_sha256) is derived from these and is +what makes audio deltas idempotent under the Delta-CRDT join (ADR-0181 §2.2). + +Hashing arrays uses the *exact bytes that would be written to disk* — the same +discipline CLAUDE.md §Semantic Pack Discipline requires of manifest checksums. +Floats are hashed as canonical float32 bytes so the hash is stable across the +float64 internal compute path (spec §7: cast to float32 only at the boundary). +""" + +from __future__ import annotations + +import hashlib +import json +from typing import Any + +import numpy as np + + +def sha256_bytes(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def sha256_array(arr: np.ndarray) -> str: + """Hash an array by its canonical float32 byte image.""" + return sha256_bytes(np.ascontiguousarray(arr, dtype=np.float32).tobytes()) + + +def sha256_json(obj: Any) -> str: + """Hash a JSON-serialisable object with sorted keys / stable separators.""" + serialized = json.dumps(obj, sort_keys=True, ensure_ascii=False, separators=(",", ":")) + return sha256_bytes(serialized.encode("utf-8")) diff --git a/sensorium/audio/compiler.py b/sensorium/audio/compiler.py new file mode 100644 index 00000000..9e9b6e4c --- /dev/null +++ b/sensorium/audio/compiler.py @@ -0,0 +1,164 @@ +""" +sensorium/audio/compiler.py — the deterministic audio compiler (spec §1, §7, §9). + +Pipeline: canonical signal → frame grid → lexer → typed AudioIR → canonical +event ordering → rotor lowering → versor composition → AudioCompilationUnit. + +The ``compile_events`` fold is the **serialization barrier** of ADR-0181 §2.1: +it composes non-commutative rotors serially, in canonical order, and the only +thing that crosses into the Delta-CRDT merge layer is the order-invariant +AudioCompilationUnit (keyed by its content-addressed merge key). + +Strict invariant (spec §9 / ADR-0181 §4.3): same canonical bytes + same pack +⇒ same IR ⇒ same versor ⇒ same projection hash ⇒ same merge key. +""" + +from __future__ import annotations + +import numpy as np + +from algebra.cl41 import geometric_product +from algebra.versor import unitize_versor, versor_condition +from sensorium.audio.canonical import CANONICAL_SAMPLE_RATE, canonicalize +from sensorium.audio.checksum import sha256_array +from sensorium.audio.frames import frame_signal +from sensorium.audio.lexer import lex +from sensorium.audio.operators import ( + DEFAULT_OPERATOR_REGISTRY, + AudioOperatorRegistry, + build_elliptic_rotor, +) +from sensorium.audio.parser import parse +from sensorium.audio.types import ( + AudioCompilationUnit, + AudioIR, + AudioSignal, + AuditoryEvent, +) + +CL41_DIM = 32 +VERSOR_CONDITION_MAX = 1e-6 + +# Manifest event precedence (spec §6.1 [ordering]). +_PRECEDENCE = ("channel", "pause", "speech", "prosody", "turn", "non_speech", "content_anchor") +_PREFIX_TO_CATEGORY = { + "pause": "pause", "speech": "speech", "prosody": "prosody", + "turn": "turn", "nonspeech": "non_speech", "channel": "channel", +} + + +def _category(event_type: str) -> str: + prefix = event_type.split(".", 1)[0] + return _PREFIX_TO_CATEGORY.get(prefix, "content_anchor") + + +def canonical_event_order(ir: AudioIR) -> list[AuditoryEvent]: + """Flatten the IR into a single canonically-ordered event sequence. + + Stable key: (precedence rank, start_hop, end_hop, event_type). This is the + order ``compile_events`` folds in — deterministic for a fixed IR. + """ + events = [ + *ir.pause_spans, *ir.speech_spans, *ir.prosody_arcs, + *ir.turn_events, *ir.non_speech_events, *ir.content_anchors, + ] + rank = {name: i for i, name in enumerate(_PRECEDENCE)} + return sorted( + events, + key=lambda e: (rank.get(_category(e.event_type), len(_PRECEDENCE)), + e.start_hop, e.end_hop, e.event_type), + ) + + +def compile_events( + events: list[AuditoryEvent], + registry: AudioOperatorRegistry, +) -> tuple[np.ndarray, float]: + """SERIALIZATION BARRIER (ADR-0181 §2.1). + + Fold the canonical-ordered events into a single unit versor. Events whose + type has no operator are skipped (they contribute evidence to the IR but + no rotor). Returns (versor float32, versor_condition). + """ + v = np.zeros(CL41_DIM, dtype=np.float64) + v[0] = 1.0 + for ev in events: + if ev.event_type not in registry: + continue + spec = registry[ev.event_type] + theta_q = spec.theta_q_from_event(ev) + r = build_elliptic_rotor(spec.blade_index, theta_q) + v = geometric_product(v, r) + v = unitize_versor(v) + vc = float(versor_condition(v)) + if vc >= VERSOR_CONDITION_MAX: + raise ValueError( + f"audio compilation failed versor check: versor_condition={vc:.3e} " + f">= {VERSOR_CONDITION_MAX:.0e}" + ) + return v.astype(np.float32), vc + + +class AudioCompiler: + """Deterministic compiler from raw waveform to an AudioCompilationUnit.""" + + def __init__( + self, + registry: AudioOperatorRegistry = DEFAULT_OPERATOR_REGISTRY, + pack_id: str = "audio_core_v1", + *, + target_sr: int = CANONICAL_SAMPLE_RATE, + ) -> None: + self._registry = registry + self._pack_id = pack_id + self._target_sr = target_sr + self._manifest_sha256 = registry.manifest_sha256() + + def compile( + self, + samples: np.ndarray, + sample_rate: int, + *, + fir: np.ndarray | None = None, + ) -> AudioCompilationUnit: + signal = canonicalize(samples, sample_rate, target_sr=self._target_sr, fir=fir) + return self._compile_signal(signal) + + def compile_signal(self, signal: AudioSignal) -> AudioCompilationUnit: + """Compile an already-canonicalised signal.""" + return self._compile_signal(signal) + + def _compile_signal(self, signal: AudioSignal) -> AudioCompilationUnit: + frames = frame_signal(signal.samples, signal.sample_rate) + tokens = lex(frames, signal.sample_rate) + ir = parse(tokens, n_hops=frames.shape[0]) + + versor, vc = compile_events(canonical_event_order(ir), self._registry) + return AudioCompilationUnit( + canonical_sha256=signal.canonical_sha256, + ir_sha256=ir.ir_sha256, + pack_id=self._pack_id, + pack_manifest_sha256=self._manifest_sha256, + projection_sha256=sha256_array(versor), + versor=versor, + versor_condition=vc, + audio_ir=ir, + ) + + def compile_ir(self, ir: AudioIR) -> AudioCompilationUnit: + """Replay: recompile a stored IR back to a versor (spec §9 IR-replay). + + ``canonical_sha256`` is not available from the IR alone; replay equality + is asserted on the versor and ``ir_sha256`` (eval-plan §3.2). + """ + versor, vc = compile_events(canonical_event_order(ir), self._registry) + return AudioCompilationUnit( + canonical_sha256="", + ir_sha256=ir.ir_sha256, + pack_id=self._pack_id, + pack_manifest_sha256=self._manifest_sha256, + projection_sha256=sha256_array(versor), + versor=versor, + versor_condition=vc, + audio_ir=ir, + ) diff --git a/sensorium/audio/frames.py b/sensorium/audio/frames.py new file mode 100644 index 00000000..3c324ff2 --- /dev/null +++ b/sensorium/audio/frames.py @@ -0,0 +1,47 @@ +""" +sensorium/audio/frames.py — fixed frame grid (spec §4). + +Default 20 ms window / 10 ms hop. Deterministic: the last partial frame is +zero-padded to the full window length so the grid is a pure function of +(signal length, sample_rate, frame_ms, hop_ms). +""" + +from __future__ import annotations + +import numpy as np + +FRAME_MS = 20 +HOP_MS = 10 + + +def frame_signal( + samples: np.ndarray, + sample_rate: int, + *, + frame_ms: int = FRAME_MS, + hop_ms: int = HOP_MS, +) -> np.ndarray: + """Return a (n_frames, frame_len) float64 matrix of zero-padded frames. + + The number of frames is ``ceil((n - frame_len)/hop) + 1`` for n >= + frame_len, else 1 (a single zero-padded frame). Hop index i spans samples + [i*hop, i*hop+frame_len). + """ + x = np.asarray(samples, dtype=np.float64) + frame_len = max(1, int(round(sample_rate * frame_ms / 1000))) + hop_len = max(1, int(round(sample_rate * hop_ms / 1000))) + + if x.size <= frame_len: + n_frames = 1 + else: + n_frames = (x.size - frame_len) // hop_len + 1 + # Cover the tail with one more zero-padded frame when it spills over. + if (n_frames - 1) * hop_len + frame_len < x.size: + n_frames += 1 + + out = np.zeros((n_frames, frame_len), dtype=np.float64) + for i in range(n_frames): + start = i * hop_len + chunk = x[start:start + frame_len] + out[i, : chunk.size] = chunk + return out diff --git a/sensorium/audio/lexer.py b/sensorium/audio/lexer.py new file mode 100644 index 00000000..36c341b1 --- /dev/null +++ b/sensorium/audio/lexer.py @@ -0,0 +1,115 @@ +""" +sensorium/audio/lexer.py — acoustic lexer (spec §4). + +Operates on measured facts, not semantic guesses. Each frame yields quantized +descriptors so the token stream hashes deterministically (spec §7: quantize +before semantics). Emits AudioTokens in canonical hop order. + +Quantization regime (frozen here for v1): + - log energy : 1 dB bins (int dB) + - F0 : 25-cent bins, referenced to 55 Hz (A1) + - confidences : uint8 (0..255) + - spectral : fixed ordinal centroid bins +All thresholds are module constants so the lexer is a pure function of the +frame matrix. +""" + +from __future__ import annotations + +import math + +import numpy as np + +from sensorium.audio.types import AudioToken + +EPS = 1e-12 + +# Quantization / classification constants (v1, frozen). +SILENCE_DB = -55 # frames quieter than this are silence +VOICED_ZCR_MAX = 0.20 # voiced frames have low zero-crossing rate +VOICED_MIN_DB = -45 # and enough energy +F0_REF_HZ = 55.0 # A1 reference for cents +CENTS_BIN = 25 # 25-cent quantization +F0_MIN_HZ = 50.0 +F0_MAX_HZ = 500.0 +N_CENTROID_BINS = 16 +MAX_PITCH_CANDIDATES = 2 + + +def _log_energy_db(frame: np.ndarray) -> float: + rms = math.sqrt(float(np.mean(frame * frame)) + EPS) + return 20.0 * math.log10(rms + EPS) + + +def _zero_crossing_rate(frame: np.ndarray) -> float: + signs = np.signbit(frame) + return float(np.count_nonzero(signs[1:] != signs[:-1])) / max(1, frame.size - 1) + + +def _spectral_centroid_bin(frame: np.ndarray) -> int: + mag = np.abs(np.fft.rfft(frame * np.hanning(frame.size))) + total = float(mag.sum()) + EPS + bins = np.arange(mag.size, dtype=np.float64) + centroid = float((bins * mag).sum()) / total / max(1, mag.size - 1) # 0..1 + return int(min(N_CENTROID_BINS - 1, round(centroid * (N_CENTROID_BINS - 1)))) + + +def _hz_to_cents_q(hz: float) -> int: + cents = 1200.0 * math.log2(max(hz, EPS) / F0_REF_HZ) + return int(round(cents / CENTS_BIN)) + + +def _pitch_candidates_q(frame: np.ndarray, sample_rate: int) -> tuple[int, ...]: + """pYIN-style: keep the top autocorrelation peaks (cents_q, prob_q) pairs, + *before* any Viterbi smoothing (spec §4).""" + n = frame.size + ac = np.correlate(frame, frame, mode="full")[n - 1:] + if ac[0] <= EPS: + return () + ac = ac / ac[0] + lag_min = max(1, int(sample_rate / F0_MAX_HZ)) + lag_max = min(n - 1, int(sample_rate / F0_MIN_HZ)) + if lag_max <= lag_min: + return () + window = ac[lag_min:lag_max] + # local maxima + peaks = [ + lag_min + i + for i in range(1, window.size - 1) + if window[i] > window[i - 1] and window[i] >= window[i + 1] and window[i] > 0.3 + ] + peaks.sort(key=lambda lag: (-float(ac[lag]), lag)) + out: list[int] = [] + for lag in peaks[:MAX_PITCH_CANDIDATES]: + hz = sample_rate / lag + prob_q = int(min(255, max(0, round(float(ac[lag]) * 255)))) + out.extend((_hz_to_cents_q(hz), prob_q)) + return tuple(out) + + +def lex(frames: np.ndarray, sample_rate: int) -> tuple[AudioToken, ...]: + """Lower a frame matrix into a canonical-ordered tuple of AudioTokens. + + One primary classification token per hop (silence / voiced / unvoiced), + plus an energy_bin token and, for voiced frames, a pitch_candidates token. + """ + tokens: list[AudioToken] = [] + for i, frame in enumerate(frames): + db = _log_energy_db(frame) + db_q = int(round(db)) + zcr = _zero_crossing_rate(frame) + tokens.append(AudioToken("energy_bin", i, i + 1, (db_q,))) + + if db_q <= SILENCE_DB: + tokens.append(AudioToken("silence", i, i + 1, (db_q,))) + continue + + if zcr <= VOICED_ZCR_MAX and db_q >= VOICED_MIN_DB: + tokens.append(AudioToken("voiced", i, i + 1, (db_q, int(round(zcr * 255))))) + cands = _pitch_candidates_q(frame, sample_rate) + if cands: + tokens.append(AudioToken("pitch_candidates", i, i + 1, cands)) + else: + centroid_q = _spectral_centroid_bin(frame) + tokens.append(AudioToken("unvoiced", i, i + 1, (db_q, centroid_q))) + return tuple(tokens) diff --git a/sensorium/audio/operators.py b/sensorium/audio/operators.py new file mode 100644 index 00000000..5957e0dc --- /dev/null +++ b/sensorium/audio/operators.py @@ -0,0 +1,132 @@ +""" +sensorium/audio/operators.py — operator registry + rotor lowering (spec §6). + +Each auditory event lowers to a *declared rotor specification*, not an opaque +vector. v1 uses **elliptic bivector operators only** (square = -1), so every +rotor is the numerically well-behaved R = cos(θ/2) + B·sin(θ/2) and the +composition of any sequence is a unit versor (versor_condition < 1e-6 holds +without weakening the threshold — CLAUDE.md §Non-Negotiable Field Invariant). + +Elliptic planes in Cl(4,1) signature (+,+,+,+,-): a grade-2 blade e_a e_b +squares to -1 iff both a,b ∈ {e1..e4}. With the algebra's blade ordering +(combinations(range(5),2)), the elliptic grade-2 indices are: + + 6=(e1e2) 7=(e1e3) 8=(e1e4) 10=(e2e3) 11=(e2e4) 13=(e3e4) + +Indices 9,12,14,15 involve e5 and are hyperbolic — excluded from v1. The +alias→index assignment below is versioned pack data (frozen in PR-3's +manifest); here it is the in-code default the compiler ships with. +""" + +from __future__ import annotations + +import math +from dataclasses import dataclass, field + +import numpy as np + +from sensorium.audio.checksum import sha256_json +from sensorium.audio.types import AuditoryEvent + +CL41_DIM = 32 + +# The six elliptic grade-2 planes (square = -1). +ELLIPTIC_PLANES: tuple[int, ...] = (6, 7, 8, 10, 11, 13) + +# θ_q is an integer; the radian angle is θ_q * THETA_STEP. 1024 steps span +# [0, 2π), so a rotor angle is always representable and bounded. +THETA_STEP = math.pi / 512.0 + + +@dataclass(frozen=True, slots=True) +class OperatorSpec: + """Declared elliptic rotor spec for one event type (spec §6.2).""" + operator_id: str + event_type: str + blade_alias: str + blade_index: int + base_theta_q: int + gain_rules: tuple[tuple[str, int], ...] # (attr_name, gain) pairs + theta_clip_q: int + version: str = "1" + + def __post_init__(self) -> None: + if self.blade_index not in ELLIPTIC_PLANES: + raise ValueError( + f"operator '{self.operator_id}' uses non-elliptic blade " + f"{self.blade_index}; v1 permits only {ELLIPTIC_PLANES}" + ) + + def theta_q_from_event(self, event: AuditoryEvent) -> int: + """Deterministic θ_q from quantized event attrs. Inputs are ints + only (spec §7: quantized inputs only), so the result is an int.""" + attrs = dict(event.attrs) + theta_q = self.base_theta_q + for attr_name, gain in self.gain_rules: + value = attrs.get(attr_name, 0) + if isinstance(value, int): + theta_q += gain * value + return max(0, min(self.theta_clip_q, theta_q)) + + +def build_elliptic_rotor(blade_index: int, theta_q: int) -> np.ndarray: + """R = cos(θ/2) + B·sin(θ/2) for an elliptic plane B. θ = θ_q·THETA_STEP. + + Returns a float64 unit versor of shape (32,).""" + if blade_index not in ELLIPTIC_PLANES: + raise ValueError(f"non-elliptic blade {blade_index}") + out = np.zeros(CL41_DIM, dtype=np.float64) + half = (theta_q * THETA_STEP) / 2.0 + out[0] = math.cos(half) + out[blade_index] = math.sin(half) + return out + + +@dataclass(frozen=True, slots=True) +class AudioOperatorRegistry: + """Maps event_type → OperatorSpec. Frozen and content-addressable.""" + specs: dict[str, OperatorSpec] = field(default_factory=dict) + + def __getitem__(self, event_type: str) -> OperatorSpec: + return self.specs[event_type] + + def __contains__(self, event_type: str) -> bool: + return event_type in self.specs + + def manifest_sha256(self) -> str: + """Content hash over the registry's canonical serialization — the + ``pack_manifest_sha256`` link of the checksum chain (spec §6).""" + payload = [ + { + "operator_id": s.operator_id, + "event_type": s.event_type, + "blade_alias": s.blade_alias, + "blade_index": s.blade_index, + "base_theta_q": s.base_theta_q, + "gain_rules": [list(g) for g in s.gain_rules], + "theta_clip_q": s.theta_clip_q, + "version": s.version, + } + for s in sorted(self.specs.values(), key=lambda x: x.operator_id) + ] + return sha256_json({"basis_version": "audio-basis-v1", "operators": payload}) + + +def _spec(op_id, etype, alias, blade, base, gains, clip=768) -> OperatorSpec: + return OperatorSpec(op_id, etype, alias, blade, base, tuple(gains), clip) + + +# In-code default registry (PR-3 externalises this to operators.jsonl). Each +# atom family maps to one elliptic plane; planes are reused across families +# (only six exist) with distinct base angles. Full orthogonality is a later +# concern — lawfulness (elliptic, unit) is the PR-2 invariant. +DEFAULT_OPERATOR_REGISTRY = AudioOperatorRegistry({ + "pause.short": _spec("audio.pause.short.v1", "pause.short", "B_PAUSE_SHORT", 6, 48, [("dur_hops", 2)]), + "pause.long": _spec("audio.pause.long.v1", "pause.long", "B_PAUSE_LONG", 7, 96, [("dur_hops", 2)]), + "speech.voiced": _spec("audio.speech.voiced.v1", "speech.voiced", "B_SPEECH", 8, 64, [("dur_hops", 1)]), + "prosody.rise": _spec("audio.prosody.rise.v1", "prosody.rise", "B_PITCH_RISE", 10, 64, [("slope_q", 3)]), + "prosody.fall": _spec("audio.prosody.fall.v1", "prosody.fall", "B_PITCH_FALL", 11, 64, [("slope_q", 3)]), + "prosody.emphasis": _spec("audio.prosody.emph.v1", "prosody.emphasis", "B_EMPHASIS", 13, 32, [("delta_db_q", 4)]), + "turn.boundary": _spec("audio.turn.boundary.v1", "turn.boundary", "B_TURN", 6, 160, [("boundary_q", 2)]), + "nonspeech.noise": _spec("audio.nonspeech.noise.v1", "nonspeech.noise", "B_NOISE", 7, 200, [("noise_q", 2)]), +}) diff --git a/sensorium/audio/parser.py b/sensorium/audio/parser.py new file mode 100644 index 00000000..f2853921 --- /dev/null +++ b/sensorium/audio/parser.py @@ -0,0 +1,126 @@ +""" +sensorium/audio/parser.py — typed AudioIR parser (spec §5). + +Promotes the lexer's per-frame tokens into typed spans and events. The IR is +built from runs of like frames, never from individual mel/frame values. Output +event types match the operator registry keys so every event lowers to a rotor. + +Determinism: every numeric attr is a quantized int; events are emitted in a +stable per-category order; ``ir_sha256`` hashes the canonical serialization. +""" + +from __future__ import annotations + +from sensorium.audio.checksum import sha256_json +from sensorium.audio.types import AudioIR, AudioToken, AuditoryEvent + +LONG_PAUSE_HOPS = 30 # >= 300 ms (10 ms hop) is a long pause / turn +SLOPE_CENTS_THRESH = 1 # min |Δcents_q| to call a contour rise/fall +EMPHASIS_DB_THRESH = 6 # min intra-span energy delta (dB) for emphasis + + +def _runs(kinds: list[str | None]) -> list[tuple[str, int, int]]: + """Collapse a per-hop primary-kind list into (kind, start_hop, end_hop).""" + runs: list[tuple[str, int, int]] = [] + i = 0 + n = len(kinds) + while i < n: + k = kinds[i] + if k is None: + i += 1 + continue + j = i + while j < n and kinds[j] == k: + j += 1 + runs.append((k, i, j)) + i = j + return runs + + +def parse(tokens: tuple[AudioToken, ...], n_hops: int) -> AudioIR: + primary: list[str | None] = [None] * n_hops + energy_db: dict[int, int] = {} + pitch_cents: dict[int, int] = {} + + for tok in tokens: + h = tok.start_hop + if tok.kind == "energy_bin": + energy_db[h] = tok.value_q[0] + elif tok.kind in ("silence", "voiced", "unvoiced"): + primary[h] = tok.kind + elif tok.kind == "pitch_candidates" and tok.value_q: + pitch_cents[h] = tok.value_q[0] # top candidate's cents_q + + speech_spans: list[AuditoryEvent] = [] + pause_spans: list[AuditoryEvent] = [] + prosody_arcs: list[AuditoryEvent] = [] + turn_events: list[AuditoryEvent] = [] + non_speech_events: list[AuditoryEvent] = [] + + for kind, start, end in _runs(primary): + dur = end - start + if kind == "silence": + is_long = dur >= LONG_PAUSE_HOPS + etype = "pause.long" if is_long else "pause.short" + pause_spans.append(AuditoryEvent(etype, start, end, (("dur_hops", dur),), ())) + if is_long: + turn_events.append( + AuditoryEvent("turn.boundary", start, end, (("boundary_q", dur),), ()) + ) + elif kind == "voiced": + speech_spans.append( + AuditoryEvent("speech.voiced", start, end, (("dur_hops", dur),), ()) + ) + # Prosody arc from the final-contour F0 slope over the span. + cents = [pitch_cents[h] for h in range(start, end) if h in pitch_cents] + if len(cents) >= 2: + slope = cents[-1] - cents[0] + if slope >= SLOPE_CENTS_THRESH: + prosody_arcs.append( + AuditoryEvent("prosody.rise", start, end, (("slope_q", slope),), ()) + ) + elif slope <= -SLOPE_CENTS_THRESH: + prosody_arcs.append( + AuditoryEvent("prosody.fall", start, end, (("slope_q", -slope),), ()) + ) + # Emphasis from intra-span energy delta. + dbs = [energy_db[h] for h in range(start, end) if h in energy_db] + if dbs and (max(dbs) - min(dbs)) >= EMPHASIS_DB_THRESH: + prosody_arcs.append( + AuditoryEvent( + "prosody.emphasis", start, end, + (("delta_db_q", max(dbs) - min(dbs)),), (), + ) + ) + elif kind == "unvoiced": + non_speech_events.append( + AuditoryEvent("nonspeech.noise", start, end, (("noise_q", dur),), ()) + ) + + ir_payload = { + "speech": [_ev(e) for e in speech_spans], + "pause": [_ev(e) for e in pause_spans], + "prosody": [_ev(e) for e in prosody_arcs], + "turn": [_ev(e) for e in turn_events], + "non_speech": [_ev(e) for e in non_speech_events], + "content_anchor": [], + } + return AudioIR( + speech_spans=tuple(speech_spans), + pause_spans=tuple(pause_spans), + prosody_arcs=tuple(prosody_arcs), + turn_events=tuple(turn_events), + non_speech_events=tuple(non_speech_events), + content_anchors=(), + ir_sha256=sha256_json(ir_payload), + ) + + +def _ev(e: AuditoryEvent) -> dict: + return { + "event_type": e.event_type, + "start_hop": e.start_hop, + "end_hop": e.end_hop, + "attrs": [list(a) for a in e.attrs], + "evidence_ids": list(e.evidence_ids), + } diff --git a/sensorium/audio/resample.py b/sensorium/audio/resample.py new file mode 100644 index 00000000..9ac6e882 --- /dev/null +++ b/sensorium/audio/resample.py @@ -0,0 +1,53 @@ +""" +sensorium/audio/resample.py — deterministic polyphase FIR resampling (spec §3). + +SciPy is intentionally NOT a dependency (it is absent from the runtime and +CLAUDE.md forbids broad infrastructure). This is a pure-numpy polyphase FIR +upsample→filter→downsample, equivalent in form to scipy.signal.resample_poly +with explicit FIR coefficients. The FIR taps are a pack artifact in PR-3 +(`resample_fir_v1.npy`); this module only *applies* them, deterministically. + +Replayability requirements (spec §3 / §7): + - odd-length symmetric FIR → zero-phase (group delay = (len-1)/2 samples). + - float64 internal compute; the caller casts to float32 at the boundary. + - same-rate input is an exact passthrough (no filtering, no drift). +""" + +from __future__ import annotations + +from math import gcd + +import numpy as np + + +def resample_poly(x: np.ndarray, up: int, down: int, fir: np.ndarray) -> np.ndarray: + """Resample ``x`` by the rational factor ``up/down`` using explicit FIR taps. + + The FIR must be a low-pass designed for the ``up`` insertion rate. An + odd-length symmetric FIR yields zero-phase output (the group delay is + removed by centering). Deterministic for fixed (x, up, down, fir). + """ + if up < 1 or down < 1: + raise ValueError(f"up/down must be >= 1, got up={up}, down={down}") + if fir.ndim != 1 or fir.size % 2 == 0: + raise ValueError("FIR must be a 1-D odd-length (symmetric) array") + + g = gcd(up, down) + up, down = up // g, down // g + xf = np.asarray(x, dtype=np.float64) + + # Upsample by zero-insertion. + upsampled = np.zeros(xf.size * up, dtype=np.float64) + upsampled[::up] = xf + + # Zero-phase FIR via centered 'same' convolution, scaled by up to + # preserve amplitude through zero-insertion. + taps = np.asarray(fir, dtype=np.float64) * up + filtered = np.convolve(upsampled, taps, mode="same") + + # Downsample by decimation. + return filtered[::down] + + +def needs_resample(sample_rate: int, target_sr: int) -> bool: + return sample_rate != target_sr diff --git a/sensorium/audio/trace.py b/sensorium/audio/trace.py new file mode 100644 index 00000000..b854c9b5 --- /dev/null +++ b/sensorium/audio/trace.py @@ -0,0 +1,26 @@ +""" +sensorium/audio/trace.py — audio evidence trace (spec §6, ADR-0181 §3.1). + +Produces the trace-safe record of a compiled audio chunk: the layered +checksum chain, pack identity, and the content-addressed merge key — and +NEVER raw waveform bytes (ADR-0181 §4.2 A-6 / ADR-0180 §1.5.5). This is what +a CognitiveTurnResult / TurnEvent stores as audio evidence. +""" + +from __future__ import annotations + +from sensorium.audio.types import AudioCompilationUnit + + +def audio_evidence_trace(unit: AudioCompilationUnit) -> dict[str, object]: + """Trace-safe evidence dict for one compiled chunk. No PCM.""" + return { + "modality": "audio", + "pack_id": unit.pack_id, + "canonical_sha256": unit.canonical_sha256, + "ir_sha256": unit.ir_sha256, + "pack_manifest_sha256": unit.pack_manifest_sha256, + "projection_sha256": unit.projection_sha256, + "merge_key": list(unit.merge_key), + "versor_condition": unit.versor_condition, + } diff --git a/sensorium/audio/types.py b/sensorium/audio/types.py new file mode 100644 index 00000000..5724f14a --- /dev/null +++ b/sensorium/audio/types.py @@ -0,0 +1,95 @@ +""" +sensorium/audio/types.py — Typed AudioIR for the CORE-native audio compiler. + +ADR-0181 §2 / spec §2. The IR is built from typed spans and events, never +from raw frames or mel bins. Every dataclass is frozen and slotted so the +compiler path is immutable and hashable, matching CORE's trace-first +epistemology. + +A signal compiles to exactly one AudioCompilationUnit — the object the audio +adapter writes into its thread-local Delta-CRDT arena (ADR-0181 §2.1). The +unit carries no PCM: only the layered checksum chain, the (32,) versor, and +the content-addressed merge key (ADR-0181 §2.2). +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Literal + +import numpy as np + +TokenKind = Literal[ + "silence", "voiced", "unvoiced", "onset", + "energy_bin", "pitch_candidates", "spectral_bin", +] + + +@dataclass(frozen=True, slots=True) +class AudioSignal: + """Canonical mono float32 signal + provenance hashes (spec §3).""" + samples: np.ndarray # canonical mono float32 + sample_rate: int # canonical rate, e.g. 24_000 + start_ms: int + end_ms: int + source_sha256: str # hash of the original input bytes + canonical_sha256: str # hash of the canonical float32 bytes + + +@dataclass(frozen=True, slots=True) +class PitchCandidate: + cents_q: int # quantized cents (25-cent bins) + prob_q: int # 0..255 + + +@dataclass(frozen=True, slots=True) +class AudioToken: + kind: TokenKind + start_hop: int + end_hop: int + value_q: tuple[int, ...] # canonical quantized payload + + +@dataclass(frozen=True, slots=True) +class AuditoryEvent: + """A typed auditory event. ``attrs`` are quantized ints or short strings + so the event serializes deterministically into the IR hash.""" + event_type: str + start_hop: int + end_hop: int + attrs: tuple[tuple[str, int | str], ...] + evidence_ids: tuple[str, ...] + + +@dataclass(frozen=True, slots=True) +class AudioIR: + speech_spans: tuple[AuditoryEvent, ...] + pause_spans: tuple[AuditoryEvent, ...] + prosody_arcs: tuple[AuditoryEvent, ...] + turn_events: tuple[AuditoryEvent, ...] + non_speech_events: tuple[AuditoryEvent, ...] + content_anchors: tuple[AuditoryEvent, ...] + ir_sha256: str + + +@dataclass(frozen=True, slots=True) +class AudioCompilationUnit: + """One compiled chunk — the Delta-CRDT delta (ADR-0181 §2.1). + + ``versor`` is the (32,) float32 Cl(4,1) multivector that crosses the + ProjectionHead boundary. ``audio_ir`` is retained for deterministic + IR-replay (spec §9); it is evidence, never re-hashed into the projection. + """ + canonical_sha256: str + ir_sha256: str + pack_id: str + pack_manifest_sha256: str + projection_sha256: str + versor: np.ndarray # (32,) float32 + versor_condition: float + audio_ir: AudioIR + + @property + def merge_key(self) -> tuple[str, str, str]: + """Content-addressed CRDT merge / dedup key (ADR-0181 §2.2).""" + return (self.canonical_sha256, self.ir_sha256, self.projection_sha256) diff --git a/tests/test_audio_compiler.py b/tests/test_audio_compiler.py new file mode 100644 index 00000000..da0f36bf --- /dev/null +++ b/tests/test_audio_compiler.py @@ -0,0 +1,173 @@ +""" +ADR-0181 PR-2 — deterministic audio substrate tests. + +Covers the substrate's load-bearing invariants (the audio analogs of +ADR-0180 §1.5.4 T-1..T-4 named A-1..A-6 in ADR-0181 §4.2): + + A-1 determinism: same canonical bytes + same pack ⇒ byte-identical (32,) + A-4 serialization barrier: in-chunk compile_events is order-sensitive + A-5 versor condition: every emitted unit < 1e-6 (never weakened) + A-6 trace hygiene: no PCM in the evidence trace + + projection shape/dtype, IR-replay, elliptic-only operator guard. + +Fixtures are deterministic synthetic signals (silence, voiced tone with a +rising contour, broadband noise) at the canonical 24 kHz, so no resampling +FIR (a PR-3 pack artifact) is needed. +""" + +from __future__ import annotations + +import numpy as np +import pytest + +from sensorium.audio import ( + AudioCompiler, + AudioCompilationUnit, + audio_evidence_trace, + build_elliptic_rotor, +) +from sensorium.audio.compiler import compile_events +from sensorium.audio.operators import ( + DEFAULT_OPERATOR_REGISTRY, + ELLIPTIC_PLANES, + OperatorSpec, +) +from sensorium.audio.types import AuditoryEvent +from algebra.versor import versor_condition + +SR = 24_000 + + +# -------------------------------------------------------------------------- +# Synthetic fixtures +# -------------------------------------------------------------------------- + +def _silence(ms: int = 500) -> np.ndarray: + return np.zeros(int(SR * ms / 1000), dtype=np.float32) + + +def _tone(hz: float, ms: int, amp: float = 0.5, sweep: float = 0.0) -> np.ndarray: + n = int(SR * ms / 1000) + t = np.arange(n, dtype=np.float64) / SR + freq = hz + sweep * t / max(t[-1], 1e-9) # linear sweep over the span + phase = 2 * np.pi * np.cumsum(freq) / SR + return (amp * np.sin(phase)).astype(np.float32) + + +def _noise(ms: int, seed: int = 0) -> np.ndarray: + rng = np.random.default_rng(seed) + n = int(SR * ms / 1000) + return (0.3 * rng.standard_normal(n)).astype(np.float32) + + +def _compile(samples: np.ndarray) -> AudioCompilationUnit: + return AudioCompiler().compile(samples, SR) + + +# -------------------------------------------------------------------------- +# Shape / dtype / versor condition +# -------------------------------------------------------------------------- + +@pytest.mark.parametrize("signal", [ + _silence(300), + _tone(160.0, 400, sweep=80.0), + _noise(300), + np.concatenate([_tone(150.0, 300), _silence(400), _tone(150.0, 300, sweep=60.0)]), +]) +def test_projection_shape_dtype_and_versor_condition(signal): + unit = _compile(signal) + assert unit.versor.shape == (32,) # projection shape + assert unit.versor.dtype == np.float32 # projection dtype + assert unit.versor_condition < 1e-6 # A-5, never weakened + + +# -------------------------------------------------------------------------- +# A-1 — determinism +# -------------------------------------------------------------------------- + +def test_a1_compile_is_byte_identical_across_calls(): + sig = np.concatenate([_tone(160.0, 350, sweep=90.0), _silence(350), _noise(200, 7)]) + u1 = _compile(sig) + u2 = _compile(sig) + assert np.array_equal(u1.versor, u2.versor) + assert u1.merge_key == u2.merge_key + assert u1.ir_sha256 == u2.ir_sha256 + assert u1.projection_sha256 == u2.projection_sha256 + + +def test_a1_same_bytes_same_merge_key_idempotent(): + """The strict invariant tail (ADR-0181 §4.3): same canonical bytes ⇒ same + merge key ⇒ CRDT-idempotent.""" + sig = _tone(200.0, 300) + assert _compile(sig).merge_key == _compile(sig.copy()).merge_key + + +# -------------------------------------------------------------------------- +# A-4 — serialization barrier (compile_events is order-sensitive) +# -------------------------------------------------------------------------- + +def test_a4_compile_events_is_order_sensitive(): + """Swapping two events in the fold changes the versor — proving the barrier + is real (non-commutative composition). If this passes trivially (orders + equal), the substrate could be wrongly sharded (ADR-0181 §2.1).""" + e_speech = AuditoryEvent("speech.voiced", 0, 5, (("dur_hops", 5),), ()) + e_pause = AuditoryEvent("pause.short", 5, 9, (("dur_hops", 4),), ()) + ab, _ = compile_events([e_speech, e_pause], DEFAULT_OPERATOR_REGISTRY) + ba, _ = compile_events([e_pause, e_speech], DEFAULT_OPERATOR_REGISTRY) + assert not np.array_equal(ab, ba) + + +# -------------------------------------------------------------------------- +# IR replay (spec §9) +# -------------------------------------------------------------------------- + +def test_ir_replay_matches_original(): + sig = np.concatenate([_tone(150.0, 400, sweep=100.0), _silence(350)]) + unit = _compile(sig) + replay = AudioCompiler().compile_ir(unit.audio_ir) + assert np.array_equal(unit.versor, replay.versor) + assert unit.ir_sha256 == replay.ir_sha256 + assert unit.projection_sha256 == replay.projection_sha256 + + +# -------------------------------------------------------------------------- +# A-6 — trace hygiene +# -------------------------------------------------------------------------- + +def test_a6_evidence_trace_has_no_pcm(): + sig = _tone(180.0, 300) + unit = _compile(sig) + trace = audio_evidence_trace(unit) + # No ndarray / raw-bytes payloads — only hashes, ids, scalars. + for value in trace.values(): + assert not isinstance(value, (np.ndarray, bytes, bytearray)) + assert trace["merge_key"] == list(unit.merge_key) + assert "samples" not in trace + + +# -------------------------------------------------------------------------- +# Operator lawfulness — elliptic planes only +# -------------------------------------------------------------------------- + +def test_default_registry_uses_only_elliptic_planes(): + for spec in DEFAULT_OPERATOR_REGISTRY.specs.values(): + assert spec.blade_index in ELLIPTIC_PLANES + + +def test_non_elliptic_operator_is_rejected(): + with pytest.raises(ValueError): + OperatorSpec("bad", "x", "B_BAD", 9, 64, (), 768) # blade 9 = e1e5 (hyperbolic) + + +def test_build_elliptic_rotor_is_unit_versor(): + for plane in ELLIPTIC_PLANES: + r = build_elliptic_rotor(plane, theta_q=137) + assert versor_condition(r) < 1e-6 + + +def test_empty_event_stream_yields_identity_versor(): + v, vc = compile_events([], DEFAULT_OPERATOR_REGISTRY) + assert vc < 1e-6 + expected = np.zeros(32, dtype=np.float32) + expected[0] = 1.0 + assert np.array_equal(v, expected)