Wire phase confidence to active probes

audio_field/active_probe.py

@@ -0,0 +1,159 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+import json
+import math
+from pathlib import Path
+import time
+
+import numpy as np
+from scipy import signal
+from scipy.io import wavfile
+
+from audio_field.phase_meaning import PhaseFieldMeaning
+from audio_field.probe_optimizer import ActiveProbeOptimizer, ProbeRequest
+from audio_field.runtime_sync import SyncState
+
+
+@dataclass(frozen=True)
+class EmittedProbe:
+    request: ProbeRequest
+    sample_rate: int
+    start_hz: float
+    end_hz: float
+    duration_seconds: float
+    channel: int
+    path: Path
+    emitted_monotonic_ns: int
+
+
+class ActiveConfidenceMaintainer:
+    """Turns low phase-field confidence into bounded active chirplet probes."""
+
+    def __init__(
+        self,
+        optimizer: ActiveProbeOptimizer,
+        *,
+        sample_rate: int,
+        output_dir: Path,
+        duration_seconds: float = 0.03,
+        start_hz: float = 1800.0,
+        end_hz: float = 9000.0,
+        channels: int = 2,
+        channel: int = 0,
+    ):
+        self.optimizer = optimizer
+        self.sample_rate = int(sample_rate)
+        self.output_dir = Path(output_dir)
+        self.duration_seconds = float(duration_seconds)
+        self.start_hz = float(start_hz)
+        self.end_hz = float(end_hz)
+        self.channels = int(channels)
+        self.channel = int(channel)
+        if self.channels <= 0:
+            raise ValueError("channels must be positive")
+        if self.channel < 0 or self.channel >= self.channels:
+            raise ValueError("channel must be inside output channel count")
+        self.output_dir.mkdir(parents=True, exist_ok=True)
+
+    def update(self, meaning: PhaseFieldMeaning, reference_block: np.ndarray, *, force_masked: bool = False) -> EmittedProbe | None:
+        states = sync_states_from_phase_meaning(meaning)
+        masked = force_masked or is_masked_window(reference_block)
+        request = self.optimizer.choose_probe(meaning.frame_id, states, masked_window=masked)
+        if request is None:
+            return None
+        waveform = make_probe_chirplet(
+            self.sample_rate,
+            duration_seconds=self.duration_seconds,
+            start_hz=self.start_hz,
+            end_hz=self.end_hz,
+            level_dbfs=request.level_dbfs,
+            channels=self.channels,
+            channel=self.channel,
+        )
+        path = self.output_dir / f"probe-{meaning.frame_id:08d}-{request.source_id}.wav"
+        wavfile.write(path, self.sample_rate, waveform.astype(np.float32))
+        emitted = EmittedProbe(
+            request=request,
+            sample_rate=self.sample_rate,
+            start_hz=self.start_hz,
+            end_hz=self.end_hz,
+            duration_seconds=self.duration_seconds,
+            channel=self.channel,
+            path=path,
+            emitted_monotonic_ns=time.monotonic_ns(),
+        )
+        append_probe_manifest(self.output_dir / "active-probes.jsonl", emitted)
+        return emitted
+
+
+def sync_states_from_phase_meaning(meaning: PhaseFieldMeaning) -> dict[str, SyncState]:
+    return {
+        source.source_id: SyncState(
+            delay_samples=source.smoothed_delay_samples,
+            phase_radians=0.0,
+            rate_scale=1.0,
+            confidence=source.confidence,
+            last_frame_index=meaning.frame_id,
+        )
+        for source in meaning.sources
+    }
+
+
+def is_masked_window(reference_block: np.ndarray, *, min_rms_dbfs: float = -42.0, crest_factor_limit: float = 8.0) -> bool:
+    samples = np.asarray(reference_block, dtype=np.float32).reshape(-1)
+    if samples.size == 0:
+        return False
+    rms = float(np.sqrt(np.mean(samples.astype(np.float64) ** 2)))
+    if rms <= db_to_amp(min_rms_dbfs):
+        return False
+    peak = float(np.max(np.abs(samples)))
+    crest = peak / max(rms, 1e-12)
+    return crest <= float(crest_factor_limit)
+
+
+def make_probe_chirplet(
+    sample_rate: int,
+    *,
+    duration_seconds: float,
+    start_hz: float,
+    end_hz: float,
+    level_dbfs: float,
+    channels: int = 2,
+    channel: int = 0,
+) -> np.ndarray:
+    frames = max(8, int(round(float(duration_seconds) * int(sample_rate))))
+    t = np.arange(frames, dtype=np.float64) / float(sample_rate)
+    chirp = signal.chirp(t, f0=float(start_hz), f1=float(end_hz), t1=float(duration_seconds), method="linear")
+    taper = signal.windows.tukey(frames, alpha=0.5)
+    chirp = chirp * taper
+    peak = float(np.max(np.abs(chirp)))
+    if peak > 0.0:
+        chirp = chirp / peak
+    chirp = chirp * db_to_amp(level_dbfs)
+    out = np.zeros((frames, int(channels)), dtype=np.float32)
+    out[:, int(channel)] = chirp.astype(np.float32)
+    return out
+
+
+def db_to_amp(db: float) -> float:
+    return float(math.pow(10.0, float(db) / 20.0))
+
+
+def append_probe_manifest(path: Path, emitted: EmittedProbe) -> None:
+    row = {
+        "frameIndex": emitted.request.frame_index,
+        "sourceId": emitted.request.source_id,
+        "reason": emitted.request.reason,
+        "levelDbfs": emitted.request.level_dbfs,
+        "urgency": emitted.request.urgency,
+        "sampleRate": emitted.sample_rate,
+        "startHz": emitted.start_hz,
+        "endHz": emitted.end_hz,
+        "durationSeconds": emitted.duration_seconds,
+        "channel": emitted.channel,
+        "path": str(emitted.path),
+        "emittedMonotonicNs": emitted.emitted_monotonic_ns,
+    }
+    with path.open("a", encoding="utf-8") as handle:
+        handle.write(json.dumps(row, sort_keys=True) + "\n")

docs/audio-field.md

@@ -132,6 +132,14 @@ Publish live extracted phase-field meaning from an aligned field and known progr
 
 This is not a phase oscilloscope. Raw phase bands stay internal. The live cache publishes consequences: per-source delay correction, distance-equivalent delta, coherence, fit error, confidence, reference-bleed estimate, suppression weight, correction energy, and whether the active probe optimizer should spend chirps.
 
+Close the confidence loop by letting the phase-field stream schedule strategic active probes:
+
+```powershell
+.\.venv\Scripts\python.exe .\scripts\stream_phase_field.py --field .\calibration\runs\<run-folder>\field-program-cleaned.wav --reference .\calibration\runs\<run-folder>\ground_truth_loopback.wav --cache .\calibration\runs\audio-phase-field.msgpack --maintain-confidence --ultrasonic-probes --play-probes
+```
+
+`--maintain-confidence` converts low-confidence phase-field state into `ActiveProbeOptimizer` requests, writes each emitted chirplet to `calibration/runs/active-probes`, and appends `active-probes.jsonl`. `--play-probes` sends the emitted chirplet to the default output device through `sounddevice`; omit it for dry-run scheduling. `--ultrasonic-probes` uses the highest safe band under Nyquist, roughly `18.5-22 kHz` at 48 kHz. This can fail silently in the physical world if the speaker, mic, driver, or resampler refuses to carry that band; the confidence feedback is the judge, not the command line's self-esteem.
+
 Record local mics while playing one speaker sweep. Use the output rate that the device probe says is real:
 
 ```powershell
@@ -214,6 +222,7 @@ The `play-record` and `record-field` commands remain for a `shared-input-device`
 - Sweep arrival estimates should use chirplet refinement when calibration signal quality allows it; plain matched-filter peaks are a fallback, not the precision path.
 - Runtime sync updates delay/SRO/phase estimates from known speaker chirplets every block/frame, gated by confidence so a bad observation freezes rather than poisons the field.
 - Active chirplet probes are scheduled by an optimization loop, not by panic. It should prefer masked windows, respect minimum spacing and level caps, and maximize expected confidence gain per audible intrusion.
+- Ultrasonic probes are allowed only as a bounded optimization strategy. They must stay below Nyquist, remain level-capped, and prove themselves by raising measured confidence; hardware may distort or discard them.
 - FOA output is AmbiX: ACN channel order, SN3D normalization, channels `W,Y,Z,X`.
 - Speaker calibration is a measurement path, not a generic monitor output.
 - Source-event geometry is published beside the AmbiX bed; render effects do not have to infer transient positions from mixed audio.
@@ -230,6 +239,7 @@ The audio code is split so the hard parts can be tested without the room, driver
 - `audio_field.pipeline` wires those ports together without knowing whether the source is a real Focusrite, an SRT receiver, a WAV fixture, or a test double.
 - The next runtime estimator module should own chirplet observations, delay/SRO/phase state, confidence gates, and phase-field updates. It should feed the aligner; it should not be hidden inside the FOA encoder.
 - The active probe optimizer owns whether to emit extra chirplets. It reads sync confidence and probe budget, then decides when and where an intentional chirplet is worth the cost.
+- `audio_field.active_probe` owns the runtime join between phase-field confidence and emitted chirplet probes. It converts extracted source confidence into sync states, applies masking/spacing policy, writes probe WAVs plus a manifest, and optionally plays them.
 - `audio_field.phase_meaning` owns live extraction of meaning from phase/chirplet evidence. It consumes known reference audio plus aligned mic windows, keeps the smoothed phase/frequency state internal, and emits only actionable state for alignment, suppression, and probe control.
 - Room suppression owns stream-side cleanup: dialogue anchors define desired direct energy, spatial/context mics act as witnesses for room and transient clutter, and the cleaned aligned field feeds FOA/Aquarium/Faust.
 - Source-event analysis owns non-vocal spatial facts: dialogue focus weights and localized transient vectors are published beside the AmbiX bed, not hidden inside it.

docs/implementation-plan.md

@@ -41,6 +41,7 @@
   - `audio_field.phase_meaning` extracts actionable delay, coherence, confidence, suppression, correction-energy, and active-probe need from internal phase/chirplet evidence.
   - `localcast.audio.phase_field` is a typed CultCache document at `calibration/runs/audio-phase-field.msgpack`.
   - `scripts/stream_phase_field.py` publishes the live meaning document from an aligned mic field plus known program/loopback reference without exposing raw phase bands as the renderer API.
+  - `audio_field.active_probe` wires low phase-field confidence into `ActiveProbeOptimizer`, emits bounded chirplet WAVs plus `active-probes.jsonl`, and can play probes through the default output device.
 
 ## Temporary
 
@@ -66,4 +67,4 @@
 13. Replace synthetic live-fusion observations with PS3 Eye detector observations and calibrated surface-feature tracks.
 14. Keep the neighbor loopback leg running through the scheduled interactive WASAPI capture path and use it as the timing witness once actual neighbor app audio is present.
 15. Let the co-streamer surface delay drive the shared presentation buffer horizon for audio stems, AmbiX, and remote video.
-16. Wire `stream_phase_field.py` to the real live aligned-field producer instead of replaying WAV artifacts, then feed its low-confidence state into the active chirplet probe optimizer.
+16. Wire `stream_phase_field.py` to the real live aligned-field producer instead of replaying WAV artifacts, then feed emitted active chirps back through loopback/mic capture so confidence maintenance closes against fresh observations instead of dry-run probe manifests.

notes/current-system-map.md

@@ -70,7 +70,7 @@ Ownership:
 - `audio_field/` owns unit-testable buffering, bounded-latency convergence, injectable port protocols, and pipeline orchestration.
 - Runtime sync owns per-block chirplet observations from known speaker output and updates delay/SRO/phase estimates with confidence gates before alignment.
 - `audio_field.phase_meaning` owns extracting usable meaning from internal phase/chirplet evidence. The live cache publishes delay correction, coherence, confidence, distance-equivalent delta, reference-bleed/suppression estimates, correction energy, and active-probe need; raw phase bands stay inside the estimator.
-- Active probe optimization owns extra chirplet emission when confidence drops, bounded by level/spacing/audibility budget.
+- Active probe optimization owns extra chirplet emission when confidence drops, bounded by level/spacing/audibility budget. `audio_field.active_probe` is the runtime join that converts phase-field confidence into emitted chirplet WAVs/manifests and optional default-device playback.
 - The camera/sensor-fusion pipeline may publish world poses later; it does not own audio clocks or channel timing.
 - OBS may ingest rendered output later; it is not the authority for the Ambisonic field.
 

scripts/live_status.py

@@ -14,6 +14,15 @@ def read_json(path: Path):
     return json.loads(path.read_text(encoding="utf-8"))
 
 
+def read_last_jsonl(path: Path):
+    if not path.exists():
+        return None
+    lines = [line for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]
+    if not lines:
+        return None
+    return json.loads(lines[-1])
+
+
 def msgpack_head(path: Path):
     if not path.exists():
         return None
@@ -101,6 +110,7 @@ def main() -> None:
             "visual-stream-status.msgpack",
             "audio-stream-status.msgpack",
             "audio-phase-field.msgpack",
+            "active-probes/active-probes.jsonl",
             "stream-spout-status.json",
             "av-sync-status.json",
         ]
@@ -114,6 +124,7 @@ def main() -> None:
         "audioHead": msgpack_head(runs / "audio-state.msgpack"),
         "eventsHead": msgpack_head(runs / "audio-events.msgpack"),
         "phaseFieldHead": msgpack_head(runs / "audio-phase-field.msgpack"),
+        "lastActiveProbe": read_last_jsonl(runs / "active-probes" / "active-probes.jsonl"),
     }
     print(json.dumps(payload, indent=2, default=str))
 

scripts/stream_phase_field.py

@@ -12,7 +12,9 @@ if str(ROOT) not in sys.path:
     sys.path.insert(0, str(ROOT))
 
 from audio_field.cultcache_audio import make_audio_phase_field, put_live_audio_phase_field  # noqa: E402
+from audio_field.active_probe import ActiveConfidenceMaintainer  # noqa: E402
 from audio_field.phase_meaning import LivePhaseMeaningExtractor  # noqa: E402
+from audio_field.probe_optimizer import ActiveProbeOptimizer, ProbePolicy  # noqa: E402
 
 
 def read_float_wav(path: Path) -> tuple[int, np.ndarray]:
@@ -74,6 +76,19 @@ def main() -> None:
     parser.add_argument("--duration", type=float)
     parser.add_argument("--loop", action="store_true")
     parser.add_argument("--realtime", action="store_true")
+    parser.add_argument("--maintain-confidence", action="store_true")
+    parser.add_argument("--target-confidence", type=float, default=0.65)
+    parser.add_argument("--trigger-confidence", type=float, default=0.35)
+    parser.add_argument("--min-probe-interval-frames", type=int, default=45)
+    parser.add_argument("--probe-level-dbfs", type=float, default=-18.0)
+    parser.add_argument("--probe-output-dir", type=Path, default=ROOT / "calibration" / "runs" / "active-probes")
+    parser.add_argument("--probe-seconds", type=float, default=0.03)
+    parser.add_argument("--probe-band", default="1800:9000")
+    parser.add_argument("--ultrasonic-probes", action="store_true", help="Use a near-ultrasonic probe band below Nyquist.")
+    parser.add_argument("--probe-channel", type=int, default=0)
+    parser.add_argument("--probe-output-channels", type=int, default=2)
+    parser.add_argument("--probe-unmasked", action="store_true", help="Allow probes even when the reference block is not masking them.")
+    parser.add_argument("--play-probes", action="store_true", help="Play scheduled probes through the default output device.")
     args = parser.parse_args()
 
     sample_rate, field = read_float_wav(args.field)
@@ -88,6 +103,27 @@ def main() -> None:
         parse_frequencies(args.frequency),
         reference_id=args.reference_id,
     )
+    maintainer = None
+    if args.maintain_confidence:
+        band_start, band_end = ultrasonic_probe_band(sample_rate) if args.ultrasonic_probes else parse_probe_band(args.probe_band)
+        maintainer = ActiveConfidenceMaintainer(
+            ActiveProbeOptimizer(
+                ProbePolicy(
+                    target_confidence=args.target_confidence,
+                    trigger_confidence=args.trigger_confidence,
+                    min_interval_frames=args.min_probe_interval_frames,
+                    max_probe_level_dbfs=args.probe_level_dbfs,
+                    prefer_masked_windows=not args.probe_unmasked,
+                )
+            ),
+            sample_rate=sample_rate,
+            output_dir=args.probe_output_dir,
+            duration_seconds=args.probe_seconds,
+            start_hz=band_start,
+            end_hz=band_end,
+            channels=args.probe_output_channels,
+            channel=args.probe_channel,
+        )
 
     start_monotonic = time.monotonic()
     audio_origin_ns = time.monotonic_ns()
@@ -122,6 +158,10 @@ def main() -> None:
                 active_probe_reason=meaning.active_probe_reason,
             ),
         )
+        if maintainer is not None:
+            emitted = maintainer.update(meaning, block_ref, force_masked=args.probe_unmasked)
+            if emitted is not None and args.play_probes:
+                play_probe_file(emitted.path)
         frame_id += 1
         cursor += args.hop_frames
         if cursor + args.chunk_frames > max_samples and args.loop:
@@ -133,5 +173,28 @@ def main() -> None:
                 time.sleep(sleep_for)
 
 
+def parse_probe_band(value: str) -> tuple[float, float]:
+    parts = value.split(":")
+    if len(parts) != 2:
+        raise SystemExit(f"probe band must be start:end Hz, got {value!r}")
+    return float(parts[0]), float(parts[1])
+
+
+def ultrasonic_probe_band(sample_rate: int) -> tuple[float, float]:
+    nyquist = 0.5 * float(sample_rate)
+    end_hz = min(22_000.0, nyquist - 1200.0)
+    start_hz = min(18_500.0, end_hz - 2500.0)
+    if end_hz <= 16_000.0 or start_hz <= 0.0:
+        raise SystemExit(f"sample rate {sample_rate} leaves no useful ultrasonic probe band")
+    return start_hz, end_hz
+
+
+def play_probe_file(path: Path) -> None:
+    import sounddevice as sd
+
+    rate, data = wavfile.read(path)
+    sd.play(data.astype(np.float32), int(rate), blocking=True)
+
+
 if __name__ == "__main__":
     main()

state/map.yaml

@@ -132,8 +132,11 @@ research:
       `audio_field.phase_meaning` keeps raw phase internal and publishes
       `localcast.audio.phase_field` with delay correction, coherence,
       confidence, distance-equivalent delta, reference-bleed/suppression
-      weights, correction energy, and active-probe need. This is learned
-      audio-field state, not a phase display.
+      weights, correction energy, and active-probe need. `audio_field.active_probe`
+      closes the next loop by feeding that confidence state into
+      ActiveProbeOptimizer and emitting bounded chirplet WAVs/manifests, with
+      optional default-device playback. This is learned audio-field state, not
+      a phase display.
   sensor_fusion_architecture:
     summary: docs/sensor-fusion-architecture.md
     key_decision: >

tests/test_active_probe.py

@@ -0,0 +1,64 @@
+import tempfile
+import unittest
+from pathlib import Path
+
+import numpy as np
+
+from audio_field.active_probe import ActiveConfidenceMaintainer, is_masked_window, make_probe_chirplet
+from audio_field.phase_meaning import PhaseFieldMeaning, SourcePhaseMeaning
+from audio_field.probe_optimizer import ActiveProbeOptimizer, ProbePolicy
+from scripts.stream_phase_field import ultrasonic_probe_band
+
+
+class ActiveProbeTests(unittest.TestCase):
+    def test_low_confidence_phase_meaning_emits_probe_manifest(self):
+        meaning = PhaseFieldMeaning(
+            frame_id=10,
+            audio_time_ns=1,
+            sample_rate=48000,
+            start_sample=0,
+            frame_count=1024,
+            reference_id="program",
+            sources=(
+                SourcePhaseMeaning("good", 0, 0.0, 0.0, 0.0, 0.0, 0.9, 0.0, 0.8, 0.1, 0.0, 0.0),
+                SourcePhaseMeaning("weak", 1, 0.0, 0.0, 0.0, 0.0, 0.2, 1.0, 0.1, 0.1, 0.0, 0.0),
+            ),
+            global_confidence=0.45,
+            needs_active_probe=True,
+            active_probe_reason="weak",
+        )
+        reference = np.full(1024, 0.05, dtype=np.float32)
+
+        with tempfile.TemporaryDirectory() as tmp:
+            maintainer = ActiveConfidenceMaintainer(
+                ActiveProbeOptimizer(ProbePolicy(trigger_confidence=0.35, min_interval_frames=1)),
+                sample_rate=48000,
+                output_dir=Path(tmp),
+            )
+            emitted = maintainer.update(meaning, reference, force_masked=True)
+
+            self.assertIsNotNone(emitted)
+            self.assertEqual("weak", emitted.request.source_id)
+            self.assertTrue(emitted.path.exists())
+            self.assertTrue((Path(tmp) / "active-probes.jsonl").exists())
+
+    def test_probe_chirplet_respects_level_and_channel(self):
+        chirp = make_probe_chirplet(48000, duration_seconds=0.02, start_hz=1000, end_hz=4000, level_dbfs=-20, channels=2, channel=1)
+
+        self.assertEqual((960, 2), chirp.shape)
+        self.assertAlmostEqual(0.0, float(np.max(np.abs(chirp[:, 0]))), delta=1e-9)
+        self.assertLessEqual(float(np.max(np.abs(chirp[:, 1]))), 0.101)
+
+    def test_masked_window_requires_enough_reference_energy(self):
+        self.assertFalse(is_masked_window(np.zeros(1024, dtype=np.float32)))
+        self.assertTrue(is_masked_window(np.full(1024, 0.05, dtype=np.float32)))
+
+    def test_ultrasonic_probe_band_stays_below_nyquist(self):
+        start_hz, end_hz = ultrasonic_probe_band(48000)
+
+        self.assertGreaterEqual(start_hz, 18000.0)
+        self.assertLess(end_hz, 24000.0)
+
+
+if __name__ == "__main__":
+    unittest.main()