solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Synthetic inputs and tolerances for the speaker differential harness.
5
6This module is intentionally self-contained. `scripts/build_core_fixtures.py`
7generates committed Rust drift-detector JSON under `core/fixtures/`; a re-pin
8there must not silently change this instrument's inputs. The differential also
9needs materially different shapes: multi-window log-prob matrices and embeddings
10that survive real AHC, not the monkeypatched clustering used by older unit tests.
11
12DRY still binds for production constants: thresholds and frame geometry are
13imported from their real homes. Only comparator tolerances are declared here,
14with enough room for float noise while staying far below branch margins.
15"""
16
17from __future__ import annotations
18
19from dataclasses import dataclass
20
21import numpy as np
22
23from solstone.observe.transcribe.diarize import (
24 FRAMES_PER_WINDOW,
25 MIN_INTERVAL_S,
26 SAMPLE_RATE,
27 SINGLE_SPEAKER_CLASSES,
28 WINDOW_S,
29)
30from solstone.observe.transcribe.main import EMBEDDER_NAME, MIN_STATEMENT_DURATION
31from solstone.observe.transcribe.overlap import OVERLAP_CLASSES, SpeakerWindowStats
32
33# Comparator tolerances. These are instrument tolerances, not production
34# thresholds. They allow small float drift while preserving all branch decisions.
35LOGPROB_MAX_ABS_TOLERANCE = 1e-4
36LOGPROB_MEDIAN_ABS_TOLERANCE = 1e-5
37LOGPROB_ARGMAX_AGREEMENT_MIN = 0.999
38EVIDENCE_FLOAT_ABS_TOLERANCE = 1e-6
39EMBEDDING_MAX_ABS_TOLERANCE = 1e-4
40EMBEDDING_MIN_COSINE_SIMILARITY = 0.9999
41EMBEDDING_MEDIAN_COSINE_SIMILARITY = 0.99999
42
43# Durations are seconds derived from sample counts; this is below one 16 kHz sample.
44STATEMENT_DURATION_ABS_TOLERANCE = 1e-6
45
46# Degenerate-norm floor for cosine math, not a comparison tolerance to tune.
47COSINE_NORM_FLOOR = 1e-12
48
49COMPARATOR_THRESHOLDS = {
50 "LOGPROB_MAX_ABS_TOLERANCE": LOGPROB_MAX_ABS_TOLERANCE,
51 "LOGPROB_MEDIAN_ABS_TOLERANCE": LOGPROB_MEDIAN_ABS_TOLERANCE,
52 "LOGPROB_ARGMAX_AGREEMENT_MIN": LOGPROB_ARGMAX_AGREEMENT_MIN,
53 "EVIDENCE_FLOAT_ABS_TOLERANCE": EVIDENCE_FLOAT_ABS_TOLERANCE,
54 "EMBEDDING_MAX_ABS_TOLERANCE": EMBEDDING_MAX_ABS_TOLERANCE,
55 "EMBEDDING_MIN_COSINE_SIMILARITY": EMBEDDING_MIN_COSINE_SIMILARITY,
56 "EMBEDDING_MEDIAN_COSINE_SIMILARITY": EMBEDDING_MEDIAN_COSINE_SIMILARITY,
57 "STATEMENT_DURATION_ABS_TOLERANCE": STATEMENT_DURATION_ABS_TOLERANCE,
58}
59
60
61@dataclass(frozen=True)
62class ModelFreeSpeakerCase:
63 audio: np.ndarray
64 statements: list[dict[str, object]]
65 avg_log_probs: np.ndarray
66 overlap_fraction: float
67 window_stats: tuple[SpeakerWindowStats, ...]
68 statement_embeddings: np.ndarray
69 statement_ids: np.ndarray
70 statement_durations_s: np.ndarray
71 interval_embeddings: np.ndarray
72
73
74def dominant_log_probs(classes: np.ndarray, *, seed: int = 20_260_724) -> np.ndarray:
75 """Return log-probs with a seeded dominant pyannote class per frame."""
76 class_count = max(max(SINGLE_SPEAKER_CLASSES), max(OVERLAP_CLASSES)) + 1
77 rng = np.random.default_rng(seed)
78 log_probs = rng.normal(
79 loc=-3.0, scale=0.05, size=(len(classes), class_count)
80 ).astype(np.float32)
81 log_probs[np.arange(len(classes)), classes] = rng.normal(
82 loc=2.0, scale=0.03, size=len(classes)
83 ).astype(np.float32)
84 return log_probs
85
86
87def separated_interval_embeddings() -> np.ndarray:
88 """Four interval embeddings in two well-separated clusters for real AHC."""
89 embs = np.zeros((4, 256), dtype=np.float32)
90 embs[0, 0] = 1.0
91 embs[1, 1] = 1.0
92 embs[2, 0] = 0.99
93 embs[2, 2] = 0.03
94 embs[3, 1] = 0.99
95 embs[3, 3] = -0.03
96 return embs
97
98
99def statement_embeddings(count: int) -> np.ndarray:
100 rng = np.random.default_rng(2_026_0724)
101 rows = rng.normal(size=(count, 256)).astype(np.float32)
102 norms = np.linalg.norm(rows, axis=1, keepdims=True)
103 return rows / np.where(norms > 1e-9, norms, 1.0)
104
105
106def model_free_case() -> ModelFreeSpeakerCase:
107 """Build a fully populated, model-free speaker differential input."""
108 single_a, single_b = sorted(SINGLE_SPEAKER_CLASSES)[:2]
109 silence = 0
110 classes = np.concatenate(
111 [
112 np.full(42, single_a, dtype=np.int64),
113 np.full(12, silence, dtype=np.int64),
114 np.full(42, single_b, dtype=np.int64),
115 np.full(12, silence, dtype=np.int64),
116 np.full(42, single_a, dtype=np.int64),
117 np.full(12, silence, dtype=np.int64),
118 np.full(42, single_b, dtype=np.int64),
119 ]
120 )
121 avg_log_probs = dominant_log_probs(classes, seed=30_001)
122 duration_s = float(len(classes) * WINDOW_S / FRAMES_PER_WINDOW + 0.2)
123 audio = np.zeros(int(duration_s * SAMPLE_RATE), dtype=np.float32)
124
125 frame_s = WINDOW_S / FRAMES_PER_WINDOW
126 assert 42 * frame_s >= MIN_INTERVAL_S
127 statements = [
128 {"id": 1, "start": 2 * frame_s, "end": 34 * frame_s, "text": "redacted"},
129 {"id": 2, "start": 58 * frame_s, "end": 88 * frame_s, "text": "redacted"},
130 {"id": 3, "start": 112 * frame_s, "end": 142 * frame_s, "text": "redacted"},
131 {"id": 4, "start": 166 * frame_s, "end": 196 * frame_s, "text": "redacted"},
132 ]
133 ids = np.array([int(stmt["id"]) for stmt in statements], dtype=np.int32)
134 durations = np.array(
135 [float(stmt["end"]) - float(stmt["start"]) for stmt in statements],
136 dtype=np.float32,
137 )
138 assert np.all(durations >= MIN_STATEMENT_DURATION)
139 return ModelFreeSpeakerCase(
140 audio=audio,
141 statements=statements,
142 avg_log_probs=avg_log_probs,
143 overlap_fraction=0.10,
144 window_stats=(SpeakerWindowStats(400, 2, 40),),
145 statement_embeddings=statement_embeddings(len(statements)),
146 statement_ids=ids,
147 statement_durations_s=durations,
148 interval_embeddings=separated_interval_embeddings(),
149 )
150
151
152def real_model_waveform(duration_s: float = 1.0, *, seed: int = 50_001) -> np.ndarray:
153 """Short synthetic waveform for real ONNX wiring tests."""
154 rng = np.random.default_rng(seed)
155 samples = int(duration_s * SAMPLE_RATE)
156 t = np.arange(samples, dtype=np.float32) / SAMPLE_RATE
157 voiced = (
158 0.025 * np.sin(2 * np.pi * 180.0 * t)
159 + 0.015 * np.sin(2 * np.pi * 310.0 * t)
160 + 0.004 * rng.standard_normal(samples)
161 )
162 envelope = np.minimum(1.0, np.linspace(0.0, 1.0, samples, dtype=np.float32) * 8.0)
163 return (voiced * envelope).astype(np.float32)
164
165
166__all__ = [
167 "COMPARATOR_THRESHOLDS",
168 "COSINE_NORM_FLOOR",
169 "EMBEDDING_MAX_ABS_TOLERANCE",
170 "EMBEDDING_MEDIAN_COSINE_SIMILARITY",
171 "EMBEDDING_MIN_COSINE_SIMILARITY",
172 "EVIDENCE_FLOAT_ABS_TOLERANCE",
173 "LOGPROB_ARGMAX_AGREEMENT_MIN",
174 "LOGPROB_MAX_ABS_TOLERANCE",
175 "LOGPROB_MEDIAN_ABS_TOLERANCE",
176 "MIN_INTERVAL_S",
177 "MIN_STATEMENT_DURATION",
178 "ModelFreeSpeakerCase",
179 "SAMPLE_RATE",
180 "STATEMENT_DURATION_ABS_TOLERANCE",
181 "WINDOW_S",
182 "EMBEDDER_NAME",
183 "model_free_case",
184 "real_model_waveform",
185]