diff --git a/solstone/observe/hear.py b/solstone/observe/hear.py index b34eca7b1..7f7061c85 100644 --- a/solstone/observe/hear.py +++ b/solstone/observe/hear.py @@ -223,7 +223,7 @@ def format_audio( pass # Add metadata fields (excluding special fields) - skip_fields = {"error", "raw", "imported", "_solstone_processing"} + skip_fields = {"error", "raw", "imported", "_solstone_processing", "sound_tags"} for key, value in metadata.items(): if key in skip_fields: diff --git a/solstone/observe/transcribe/audio.schema.json b/solstone/observe/transcribe/audio.schema.json index 902e96201..c7b305f5c 100644 --- a/solstone/observe/transcribe/audio.schema.json +++ b/solstone/observe/transcribe/audio.schema.json @@ -21,7 +21,24 @@ "warning": {"type": "string"}, "imported": {"type": "object", "additionalProperties": true}, "facet": {"type": "string"}, - "id": {"type": "string"} + "id": {"type": "string"}, + "sound_tags": { + "type": "object", + "additionalProperties": true, + "required": ["engine", "model", "threshold", "window_s", "agg", "windows", "tags"], + "properties": { + "engine": {"type": "string"}, + "model": {"type": "string"}, + "threshold": {"type": "number"}, + "window_s": {"type": "number"}, + "agg": {"type": "string"}, + "windows": {"type": "integer"}, + "tags": { + "type": "object", + "additionalProperties": {"type": "number"} + } + } + } } }, "record": { diff --git a/solstone/observe/transcribe/main.py b/solstone/observe/transcribe/main.py index e8e1b6401..5e696b8cd 100644 --- a/solstone/observe/transcribe/main.py +++ b/solstone/observe/transcribe/main.py @@ -88,6 +88,7 @@ from solstone.observe.transcribe.resource import ( select_stt_backend, stt_local_floor_bytes, ) +from solstone.observe.transcribe.sound_tags import is_salient, tag_audio from solstone.observe.utils import ( SAMPLE_RATE, AudioDecodeError, @@ -471,6 +472,7 @@ def _statements_to_jsonl( overlap_fraction: float | None = None, overlap_detector: str | None = None, processing_record: dict | None = None, + sound_tags: dict | None = None, ) -> list[str]: """Convert statements to JSONL lines. @@ -487,6 +489,10 @@ def _statements_to_jsonl( segment_meta: Optional metadata dict from SEGMENT_META env var (facet, setting, host, platform, etc.). Setting overrides enrichment. backend: Optional STT backend name (e.g., "parakeet", "revai") + overlap_fraction: Optional fraction of speech containing overlapping speakers + overlap_detector: Optional overlap detector identifier + processing_record: Optional _solstone_processing record + sound_tags: Optional ambient sound-tag metadata Returns: List of JSON strings (metadata line first, then entries) @@ -536,6 +542,8 @@ def _statements_to_jsonl( if processing_record is not None: metadata["_solstone_processing"] = processing_record + if sound_tags is not None: + metadata["sound_tags"] = sound_tags lines = [json.dumps(metadata)] @@ -589,6 +597,7 @@ def _write_empty_processing_jsonl( vad_result: VadResult | None, segment_meta: dict | None, backend: str | None, + sound_tags: dict | None = None, ) -> None: record = build_processing_record( state=STATE_EMPTY, @@ -606,6 +615,7 @@ def _write_empty_processing_jsonl( segment_meta=segment_meta, backend=backend, processing_record=record, + sound_tags=sound_tags, ) write_text(jsonl_path, "\n".join(lines) + "\n") @@ -628,6 +638,7 @@ def _write_failed_processing_jsonl( datetime.datetime.min, {}, processing_record=record, + sound_tags=None, ) write_text(jsonl_path, "\n".join(lines) + "\n") @@ -642,6 +653,8 @@ def process_audio( reduced_audio: np.ndarray | None = None, backend: str | None = None, entity_names: list[str] | None = None, + *, + sound_tags: dict | None = None, ) -> None: """Process a raw audio file with pre-computed VAD. @@ -662,6 +675,7 @@ def process_audio( reduced_audio: Optional reduced audio buffer (used if reduction provided) backend: STT backend name. If omitted, uses DEFAULT_BACKEND. entity_names: Optional list of entity names for STT and enrichment context + sound_tags: Optional ambient sound-tag metadata computed from full audio """ start_time = time.time() resolved_backend = backend or DEFAULT_BACKEND @@ -745,12 +759,30 @@ def process_audio( vad_result=vad_result, segment_meta=segment_meta, backend=resolved_backend, + sound_tags=sound_tags, ) logging.info( f"No speech detected in {raw_path}, preserving file " f"(preserve_all=true, VAD: {vad_result.speech_duration:.1f}s " f"of {vad_result.duration:.1f}s)" ) + elif sound_tags is not None and is_salient(sound_tags["tags"]): + event["outcome"] = "filtered" + _write_empty_processing_jsonl( + raw_path, + jsonl_path, + model_info=model_info, + observer=observer, + vad_result=vad_result, + segment_meta=segment_meta, + backend=resolved_backend, + sound_tags=sound_tags, + ) + logging.info( + "No speech detected in %s, wrote salient sound metadata before removing file", + raw_path, + ) + raw_path.unlink() else: event["outcome"] = "filtered" logging.info(f"No speech detected in {raw_path}, removing file") @@ -866,6 +898,7 @@ def process_audio( overlap_fraction=overlap_fraction_value, overlap_detector=OVERLAP_DETECTOR_ID, processing_record=processing_record, + sound_tags=sound_tags, ) # Write JSONL @@ -1001,6 +1034,16 @@ def _process_one( # Stage 1: Run VAD to detect speech (lightweight, before loading STT model) vad_result = run_vad(audio_buffer, min_speech_seconds=min_speech_seconds) + try: + sound_tags = tag_audio(audio_buffer, SAMPLE_RATE) + except Exception as exc: + logging.warning( + "sound tagging failed for %s: %s", + audio_path, + exc, + exc_info=True, + ) + sound_tags = None # Early exit if no speech detected (skip loading heavy STT model) if not vad_result.has_speech: @@ -1018,12 +1061,30 @@ def _process_one( vad_result=vad_result, segment_meta=None, backend=None, + sound_tags=sound_tags, ) logging.info( f"Insufficient speech in {audio_path}, preserving file " f"(preserve_all=true, VAD: {vad_result.speech_duration:.1f}s " f"of {vad_result.duration:.1f}s, threshold: {min_speech_seconds:.1f}s)" ) + elif sound_tags is not None and is_salient(sound_tags["tags"]): + event["outcome"] = "filtered" + _write_empty_processing_jsonl( + audio_path, + _get_jsonl_path(audio_path), + model_info={}, + observer=observer, + vad_result=vad_result, + segment_meta=None, + backend=None, + sound_tags=sound_tags, + ) + logging.info( + "Insufficient speech in %s, wrote salient sound metadata before removing file", + audio_path, + ) + audio_path.unlink() else: event["outcome"] = "filtered" logging.info( @@ -1142,6 +1203,7 @@ def _process_one( reduced_audio=reduced_audio, backend=backend, entity_names=entity_names, + sound_tags=sound_tags, ) diff --git a/solstone/observe/transcribe/sound_tags.py b/solstone/observe/transcribe/sound_tags.py new file mode 100644 index 000000000..883c25978 --- /dev/null +++ b/solstone/observe/transcribe/sound_tags.py @@ -0,0 +1,337 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Ambient sound tagging for transcribed audio segments. + +ced.cpp v0.1.0 C API reference: + +```c +int ced_capi_abi_version(void); +ced_ctx* ced_capi_load(const char* gguf_path); +void ced_capi_free(ced_ctx* ctx); +const char* ced_capi_last_error(const ced_ctx* ctx); +int ced_capi_num_classes(const ced_ctx* ctx); +const char* ced_capi_label(const ced_ctx* ctx, int index); +int ced_capi_sample_rate(const ced_ctx* ctx); +char* ced_capi_classify_pcm_json(ced_ctx* ctx, const float* samples, int n_samples, + int sample_rate, int top_k); +char* ced_capi_classify_path_json(ced_ctx* ctx, const char* wav_path, int top_k); +int ced_capi_classify_pcm(ced_ctx* ctx, const float* samples, int n_samples, + int sample_rate, ced_tag* out, int max_tags); +void ced_capi_free_string(char* s); +``` + +Only the symbols used by this runtime path are bound. +""" + +from __future__ import annotations + +import atexit +import json +import logging +from collections.abc import Callable +from pathlib import Path +from typing import Any + +LOG = logging.getLogger(__name__) + +SCORE_FLOOR = 0.1 +WINDOW_S = 10 +MIN_TAIL_S = 1 +SALIENCE_THRESHOLD = 0.2 +SILENCE_FAMILY = frozenset({"Silence", "White noise"}) +ENGINE = "ced.cpp v0.1.0" +MODEL = "ced-tiny-q8_0" +AGG = "max" +ABI_VERSION = 1 +CLASSIFY_SAMPLE_RATE = 16000 + +_disabled = False +_lib: Any | None = None +_ctx: Any | None = None +_np: Any | None = None +_ctypes: Any | None = None +_cleanup_registered = False + + +def parse_classify_json(raw: str) -> dict[str, float]: + data = json.loads(raw) + if not isinstance(data, list): + raise ValueError("ced classify JSON must be an array") + + tags: dict[str, float] = {} + for item in data: + if not isinstance(item, dict): + raise ValueError("ced classify JSON entries must be objects") + label = item.get("label") + score = item.get("score") + if not isinstance(label, str) or not label: + raise ValueError("ced classify JSON entry label must be a non-empty string") + if not isinstance(score, (int, float)) or isinstance(score, bool): + raise ValueError("ced classify JSON entry score must be numeric") + score_float = float(score) + if label not in tags or score_float > tags[label]: + tags[label] = score_float + return tags + + +def _window_spans(n_samples: int, sample_rate: int) -> list[tuple[int, int]]: + if sample_rate <= 0: + raise ValueError("sample_rate must be positive") + if n_samples <= 0: + return [] + + window_samples = WINDOW_S * sample_rate + min_tail_samples = MIN_TAIL_S * sample_rate + spans: list[tuple[int, int]] = [] + + full_windows = n_samples // window_samples + for i in range(full_windows): + start = i * window_samples + spans.append((start, start + window_samples)) + + tail_start = full_windows * window_samples + tail_samples = n_samples - tail_start + if tail_samples >= min_tail_samples: + spans.append((tail_start, n_samples)) + return spans + + +def aggregate(per_window: list[dict[str, float]]) -> dict[str, float]: + max_scores: dict[str, float] = {} + for tags in per_window: + for label, score in tags.items(): + score_float = float(score) + if label not in max_scores or score_float > max_scores[label]: + max_scores[label] = score_float + + kept = [ + (label, score) for label, score in max_scores.items() if score > SCORE_FLOOR + ] + kept.sort(key=lambda item: (-item[1], item[0])) + return {label: round(score, 3) for label, score in kept} + + +def is_salient(tags: dict[str, float]) -> bool: + return any( + label not in SILENCE_FAMILY and score >= SALIENCE_THRESHOLD + for label, score in tags.items() + ) + + +def _disable(reason: str) -> None: + global _disabled + if _disabled: + return + _disabled = True + LOG.warning("sound tagger disabled: %s", reason) + + +def _require_nonempty(path: Path, label: str) -> None: + if not path.is_file(): + raise RuntimeError(f"{label} missing: {path}") + if path.stat().st_size <= 0: + raise RuntimeError(f"{label} is empty: {path}") + + +def _check_asset_sizes() -> None: + from solstone.think.providers import ced_install + + model = ced_install.model_path() + if not model.is_file(): + raise RuntimeError(f"ced model missing: {model}") + actual_size = model.stat().st_size + expected_size = ced_install.CED_MODEL_SPEC.size_bytes + if actual_size != expected_size: + raise RuntimeError( + f"ced model size mismatch: expected {expected_size}, got {actual_size}" + ) + + _require_nonempty(ced_install.engine_lib_path(), "ced engine library") + _require_nonempty(ced_install.engine_header_path(), "ced C API header") + + +def _last_error(lib: Any, ctx: Any | None) -> str | None: + try: + raw = lib.ced_capi_last_error(ctx) + except Exception: + return None + if not raw: + return None + return raw.decode("utf-8", "replace") + + +def _cleanup_context() -> None: + global _ctx + if _lib is None or _ctx is None: + return + try: + _lib.ced_capi_free(_ctx) + except Exception: + return + _ctx = None + + +def _bind_symbols(lib: Any, ctypes: Any) -> None: + lib.ced_capi_abi_version.restype = ctypes.c_int + lib.ced_capi_abi_version.argtypes = [] + + lib.ced_capi_load.restype = ctypes.c_void_p + lib.ced_capi_load.argtypes = [ctypes.c_char_p] + + lib.ced_capi_free.restype = None + lib.ced_capi_free.argtypes = [ctypes.c_void_p] + + lib.ced_capi_last_error.restype = ctypes.c_char_p + lib.ced_capi_last_error.argtypes = [ctypes.c_void_p] + + lib.ced_capi_classify_pcm_json.restype = ctypes.c_void_p + lib.ced_capi_classify_pcm_json.argtypes = [ + ctypes.c_void_p, + ctypes.POINTER(ctypes.c_float), + ctypes.c_int, + ctypes.c_int, + ctypes.c_int, + ] + + lib.ced_capi_free_string.restype = None + lib.ced_capi_free_string.argtypes = [ctypes.c_void_p] + + +def _load() -> tuple[Any, Any, Any, Any] | None: + global _cleanup_registered, _ctx, _ctypes, _lib, _np + + if _disabled: + return None + if ( + _lib is not None + and _ctx is not None + and _np is not None + and _ctypes is not None + ): + return _lib, _ctx, _np, _ctypes + + try: + _check_asset_sizes() + + import ctypes + + import numpy as np + + from solstone.think.providers import ced_install + + lib_path = ced_install.engine_lib_path() + model_path = ced_install.model_path() + lib = ctypes.CDLL(str(lib_path)) + _bind_symbols(lib, ctypes) + + abi_version = lib.ced_capi_abi_version() + if abi_version != ABI_VERSION: + _disable( + f"ced C API ABI mismatch: expected {ABI_VERSION}, got {abi_version}" + ) + return None + + ctx = lib.ced_capi_load(str(model_path).encode("utf-8")) + if not ctx: + reason = _last_error(lib, None) or "ced_capi_load returned NULL" + _disable(f"ced model load failed: {reason}") + return None + + _lib = lib + _ctx = ctx + _np = np + _ctypes = ctypes + if not _cleanup_registered: + atexit.register(_cleanup_context) + _cleanup_registered = True + return _lib, _ctx, _np, _ctypes + except Exception as exc: + _disable(str(exc)) + return None + + +def _ensure_loaded() -> tuple[Any, Any, Any, Any] | None: + if _disabled: + return None + return _load() + + +def _classify_pcm_json(window: Any, _sample_rate: int) -> dict[str, float]: + loaded = _ensure_loaded() + if loaded is None: + raise RuntimeError("sound tagger disabled") + lib, ctx, np, ctypes = loaded + + contiguous = np.ascontiguousarray(window, dtype=np.float32) + samples = contiguous.ctypes.data_as(ctypes.POINTER(ctypes.c_float)) + result_ptr = lib.ced_capi_classify_pcm_json( + ctx, + samples, + int(contiguous.shape[0]), + CLASSIFY_SAMPLE_RATE, + 0, + ) + if not result_ptr: + reason = _last_error(lib, ctx) or "ced_capi_classify_pcm_json returned NULL" + raise RuntimeError(reason) + + try: + raw = ctypes.cast(result_ptr, ctypes.c_char_p).value + if raw is None: + raise RuntimeError("ced_capi_classify_pcm_json returned no bytes") + return parse_classify_json(raw.decode("utf-8")) + finally: + lib.ced_capi_free_string(result_ptr) + + +def tag_audio( + buffer: Any, + sample_rate: int, + *, + classify: Callable[[Any, int], dict[str, float]] | None = None, +) -> dict[str, Any] | None: + if _disabled: + return None + + try: + spans = _window_spans(len(buffer), sample_rate) + if not spans: + return None + + classifier = classify + if classifier is None: + if _ensure_loaded() is None: + return None + classifier = _classify_pcm_json + + per_window: list[dict[str, float]] = [] + failures: list[Exception] = [] + for i, (start, end) in enumerate(spans): + try: + per_window.append(classifier(buffer[start:end], sample_rate)) + except Exception as exc: + failures.append(exc) + LOG.debug("sound tagger window %d failed: %s", i, exc) + + if not per_window: + cause = failures[0] if failures else "no successful windows" + LOG.warning("sound tagger failed for all windows: %s", cause) + return None + + tags = aggregate(per_window) + if not tags: + return None + + return { + "engine": ENGINE, + "model": MODEL, + "threshold": SCORE_FLOOR, + "window_s": WINDOW_S, + "agg": AGG, + "windows": len(per_window), + "tags": tags, + } + except Exception as exc: + _disable(str(exc)) + return None diff --git a/solstone/talent/journal/contract/bundle.json b/solstone/talent/journal/contract/bundle.json index c0fe4d57f..6bc2987e6 100644 --- a/solstone/talent/journal/contract/bundle.json +++ b/solstone/talent/journal/contract/bundle.json @@ -82,6 +82,45 @@ "setting": { "type": "string" }, + "sound_tags": { + "additionalProperties": true, + "properties": { + "agg": { + "type": "string" + }, + "engine": { + "type": "string" + }, + "model": { + "type": "string" + }, + "tags": { + "additionalProperties": { + "type": "number" + }, + "type": "object" + }, + "threshold": { + "type": "number" + }, + "window_s": { + "type": "number" + }, + "windows": { + "type": "integer" + } + }, + "required": [ + "engine", + "model", + "threshold", + "window_s", + "agg", + "windows", + "tags" + ], + "type": "object" + }, "topics": { "type": "array" }, diff --git a/tests/test_formatters.py b/tests/test_formatters.py index 611b736a2..eaf0d25af 100644 --- a/tests/test_formatters.py +++ b/tests/test_formatters.py @@ -451,6 +451,31 @@ class TestFormatAudio: assert "solstone.processing" not in header assert len(chunks) == 1 + def test_format_audio_skips_sound_tags(self): + from solstone.observe.hear import format_audio + + entries = [ + { + "raw": "audio.flac", + "sound_tags": { + "engine": "ced.cpp v0.1.0", + "model": "ced-tiny-q8_0", + "threshold": 0.1, + "window_s": 10, + "agg": "max", + "windows": 1, + "tags": {"Music": 0.201}, + }, + }, + {"start": "00:00:01", "text": "Test"}, + ] + chunks, meta = format_audio(entries) + header = meta.get("header", "") + assert "sound_tags" not in header + assert "ced.cpp" not in header + assert "Music" not in header + assert len(chunks) == 1 + def test_format_audio_imported_metadata(self): """Test audio formatting with imported metadata.""" from solstone.observe.hear import format_audio diff --git a/tests/test_journal_contract.py b/tests/test_journal_contract.py index 23d5dc3ef..d22e53178 100644 --- a/tests/test_journal_contract.py +++ b/tests/test_journal_contract.py @@ -86,6 +86,26 @@ def test_contract_validator_accepts_producer_headers_with_raw() -> None: assert journal.validate_contract_file("audio.jsonl", audio, audio_schema) == [] +def test_contract_validator_accepts_audio_sound_tags_header() -> None: + bundle = journal.build_bundle() + audio_schema = bundle["schemas"]["audio-jsonl"]["schema"] + header = { + "raw": "audio.flac", + "sound_tags": { + "engine": "ced.cpp v0.1.0", + "model": "ced-tiny-q8_0", + "threshold": 0.1, + "window_s": 10, + "agg": "max", + "windows": 2, + "tags": {"Speech": 0.872, "Music": 0.201}, + }, + } + audio = json.dumps(header).encode("utf-8") + b'\n{"start":"00:00:00","text":"hi"}\n' + + assert journal.validate_contract_file("audio.jsonl", audio, audio_schema) == [] + + def test_contract_validator_still_rejects_non_raw_floor_violations() -> None: bundle = journal.build_bundle() screen_schema = bundle["schemas"]["screen-jsonl"]["schema"] diff --git a/tests/test_sound_tags.py b/tests/test_sound_tags.py new file mode 100644 index 000000000..296cc6d6d --- /dev/null +++ b/tests/test_sound_tags.py @@ -0,0 +1,216 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +import logging + +import numpy as np +import pytest + +from solstone.observe.transcribe import sound_tags +from solstone.observe.utils import SAMPLE_RATE + +SINE_WAVE_JSON = ( + '[{"index": 501, "score": 0.906418, "label": "Sine wave"}, ' + '{"index": 503, "score": 0.027465, "label": "Chirp tone"}, ' + '{"index": 394, "score": 0.01928, "label": "Busy signal"}]' +) + + +@pytest.fixture(autouse=True) +def reset_sound_tagger(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr(sound_tags, "_disabled", False) + monkeypatch.setattr(sound_tags, "_lib", None) + monkeypatch.setattr(sound_tags, "_ctx", None) + monkeypatch.setattr(sound_tags, "_np", None) + monkeypatch.setattr(sound_tags, "_ctypes", None) + + +def _buffer(seconds: int | float) -> np.ndarray: + return np.zeros(int(seconds * SAMPLE_RATE), dtype=np.float32) + + +def test_parse_classify_json_reads_verbatim_sine_wave_bytes() -> None: + assert sound_tags.parse_classify_json(SINE_WAVE_JSON) == { + "Sine wave": 0.906418, + "Chirp tone": 0.027465, + "Busy signal": 0.01928, + } + + +def test_parse_classify_json_keeps_max_for_duplicate_labels() -> None: + raw = ( + '[{"score": 0.1, "label": "Music"}, ' + '{"score": 0.4, "label": "Music"}, ' + '{"score": 0.2, "label": "Speech"}]' + ) + + assert sound_tags.parse_classify_json(raw) == {"Music": 0.4, "Speech": 0.2} + + +def test_window_spans_full_windows() -> None: + assert sound_tags._window_spans(20 * SAMPLE_RATE, SAMPLE_RATE) == [ + (0, 10 * SAMPLE_RATE), + (10 * SAMPLE_RATE, 20 * SAMPLE_RATE), + ] + + +def test_window_spans_includes_tail_exactly_one_second() -> None: + assert sound_tags._window_spans(21 * SAMPLE_RATE, SAMPLE_RATE) == [ + (0, 10 * SAMPLE_RATE), + (10 * SAMPLE_RATE, 20 * SAMPLE_RATE), + (20 * SAMPLE_RATE, 21 * SAMPLE_RATE), + ] + + +def test_window_spans_excludes_tail_under_one_second() -> None: + assert sound_tags._window_spans(int(20.5 * SAMPLE_RATE), SAMPLE_RATE) == [ + (0, 10 * SAMPLE_RATE), + (10 * SAMPLE_RATE, 20 * SAMPLE_RATE), + ] + + +def test_window_spans_short_audio_has_no_windows() -> None: + assert sound_tags._window_spans(int(0.99 * SAMPLE_RATE), SAMPLE_RATE) == [] + + +def test_aggregate_max_floor_rounding_and_order() -> None: + result = sound_tags.aggregate( + [ + { + "Speech": 0.1999, + "Music": 0.1004, + "Inside, small room": 0.1, + "Alpha": 0.3334, + }, + { + "Speech": 0.2, + "Music": 0.099, + "Beta": 0.3334, + }, + ] + ) + + assert list(result.items()) == [ + ("Alpha", 0.333), + ("Beta", 0.333), + ("Speech", 0.2), + ("Music", 0.1), + ] + + +def test_salience_excludes_silence_family() -> None: + assert not sound_tags.is_salient({"Silence": 0.9, "White noise": 0.8}) + + +def test_salience_includes_exact_threshold_for_non_silence() -> None: + assert sound_tags.is_salient({"Music": 0.2}) + + +def test_salience_rejects_below_threshold() -> None: + assert not sound_tags.is_salient({"Music": 0.199}) + + +def test_tag_audio_success_header_with_stub_classifier() -> None: + calls: list[tuple[int, int]] = [] + responses = [ + {"Speech": 0.872}, + {"Music": 0.201}, + {"Silence": 0.5}, + ] + + def classify(window: np.ndarray, sample_rate: int) -> dict[str, float]: + calls.append((len(window), sample_rate)) + return responses[len(calls) - 1] + + result = sound_tags.tag_audio(_buffer(21), SAMPLE_RATE, classify=classify) + + assert result == { + "engine": "ced.cpp v0.1.0", + "model": "ced-tiny-q8_0", + "threshold": 0.1, + "window_s": 10, + "agg": "max", + "windows": 3, + "tags": {"Speech": 0.872, "Silence": 0.5, "Music": 0.201}, + } + assert calls == [ + (10 * SAMPLE_RATE, SAMPLE_RATE), + (10 * SAMPLE_RATE, SAMPLE_RATE), + (SAMPLE_RATE, SAMPLE_RATE), + ] + + +def test_tag_audio_partial_window_failures_count_successes( + caplog: pytest.LogCaptureFixture, +) -> None: + calls = 0 + + def classify(_window: np.ndarray, _sample_rate: int) -> dict[str, float]: + nonlocal calls + calls += 1 + if calls == 2: + raise RuntimeError("bad window") + return {"Music": 0.3 + calls / 100} + + caplog.set_level(logging.DEBUG, logger=sound_tags.LOG.name) + + result = sound_tags.tag_audio(_buffer(21), SAMPLE_RATE, classify=classify) + + assert result is not None + assert result["windows"] == 2 + assert result["tags"] == {"Music": 0.33} + assert any( + "sound tagger window 1 failed" in record.message for record in caplog.records + ) + assert not [ + record for record in caplog.records if record.levelno >= logging.WARNING + ] + + +def test_tag_audio_all_windows_fail_returns_none_with_one_warning( + caplog: pytest.LogCaptureFixture, +) -> None: + def classify(_window: np.ndarray, _sample_rate: int) -> dict[str, float]: + raise RuntimeError("ced classify failed") + + caplog.set_level(logging.WARNING, logger=sound_tags.LOG.name) + + assert sound_tags.tag_audio(_buffer(20), SAMPLE_RATE, classify=classify) is None + + warnings = [ + record + for record in caplog.records + if record.levelno == logging.WARNING + and "sound tagger failed for all windows" in record.message + ] + assert len(warnings) == 1 + + +def test_disable_latch_logs_once(caplog: pytest.LogCaptureFixture) -> None: + caplog.set_level(logging.WARNING, logger=sound_tags.LOG.name) + + sound_tags._disable("missing assets") + sound_tags._disable("lib load failed") + + warnings = [ + record + for record in caplog.records + if record.levelno == logging.WARNING + and "sound tagger disabled" in record.message + ] + assert len(warnings) == 1 + assert warnings[0].message == "sound tagger disabled: missing assets" + + +def test_tag_audio_no_labels_over_floor_returns_none_without_warning( + caplog: pytest.LogCaptureFixture, +) -> None: + def classify(_window: np.ndarray, _sample_rate: int) -> dict[str, float]: + return {"Inside, small room": 0.1, "Quiet": 0.099} + + caplog.set_level(logging.WARNING, logger=sound_tags.LOG.name) + + assert sound_tags.tag_audio(_buffer(10), SAMPLE_RATE, classify=classify) is None + assert caplog.records == [] diff --git a/tests/test_transcribe_empty_result.py b/tests/test_transcribe_empty_result.py index 396248afb..095a8dfa4 100644 --- a/tests/test_transcribe_empty_result.py +++ b/tests/test_transcribe_empty_result.py @@ -5,6 +5,7 @@ import argparse import json +import logging from unittest.mock import MagicMock, patch import numpy as np @@ -14,6 +15,26 @@ from solstone.observe.processing_record import REASON_NO_DECODABLE_AUDIO, STATE_ from solstone.observe.utils import SAMPLE_RATE from solstone.observe.vad import VadResult +SOUND_TAGS = { + "engine": "ced.cpp v0.1.0", + "model": "ced-tiny-q8_0", + "threshold": 0.1, + "window_s": 10, + "agg": "max", + "windows": 1, + "tags": {"Music": 0.201, "Silence": 0.5}, +} + +SILENCE_SOUND_TAGS = { + "engine": "ced.cpp v0.1.0", + "model": "ced-tiny-q8_0", + "threshold": 0.1, + "window_s": 10, + "agg": "max", + "windows": 1, + "tags": {"Silence": 0.7, "White noise": 0.3}, +} + @pytest.fixture def raw_path(tmp_path): @@ -38,15 +59,79 @@ def vad_result(): ) -def test_empty_statements_filter_path(raw_path, audio_buffer, vad_result): - from solstone.observe.transcribe.main import process_audio +@pytest.fixture +def no_speech_vad_result(): + return VadResult( + duration=10.0, + speech_duration=0.0, + has_speech=False, + speech_segments=[], + ) + +def _backend_module() -> MagicMock: backend_module = MagicMock() backend_module.get_model_info.return_value = { "model": "medium.en", "device": "cpu", "compute_type": "int8", } + return backend_module + + +def _read_header(jsonl_path): + return json.loads(jsonl_path.read_text(encoding="utf-8").splitlines()[0]) + + +def test_process_audio_speech_writes_sound_tags_and_keeps_audio( + raw_path, + audio_buffer, + vad_result, +): + from solstone.observe.transcribe.main import process_audio + + statements = [{"id": 0, "start": 0.0, "end": 1.0, "text": "hi"}] + + with ( + patch( + "solstone.observe.transcribe.main.get_config", + return_value={"transcribe": {"preserve_all": False, "enrich": False}}, + ), + patch( + "solstone.observe.transcribe.main.get_journal", + return_value=str(raw_path.parents[4]), + ), + patch( + "solstone.observe.transcribe.main.stt_transcribe", return_value=statements + ), + patch( + "solstone.observe.transcribe.main.get_backend", + return_value=_backend_module(), + ), + patch("solstone.observe.transcribe.main._embed_statements", return_value=None), + patch( + "solstone.observe.transcribe.overlap.compute_overlap_and_logprobs", + return_value=(0.0, np.zeros((589, 7), dtype=np.float32)), + ), + patch("solstone.observe.transcribe.main.callosum_send") as mock_send, + ): + process_audio( + raw_path, + audio_buffer, + vad_result, + {}, + backend="parakeet", + sound_tags=SOUND_TAGS, + ) + + assert raw_path.exists() + header = _read_header(raw_path.with_suffix(".jsonl")) + assert header["sound_tags"] == SOUND_TAGS + assert mock_send.call_args.kwargs["outcome"] == "transcribed" + + +def test_empty_statements_filter_path(raw_path, audio_buffer, vad_result): + from solstone.observe.transcribe.main import process_audio with ( patch( @@ -59,7 +144,8 @@ def test_empty_statements_filter_path(raw_path, audio_buffer, vad_result): ), patch("solstone.observe.transcribe.main.stt_transcribe", return_value=[]), patch( - "solstone.observe.transcribe.main.get_backend", return_value=backend_module + "solstone.observe.transcribe.main.get_backend", + return_value=_backend_module(), ), patch("solstone.observe.transcribe.main.callosum_send") as mock_send, ): @@ -74,13 +160,6 @@ def test_empty_statements_filter_path(raw_path, audio_buffer, vad_result): def test_empty_statements_preserve_path(raw_path, audio_buffer, vad_result): from solstone.observe.transcribe.main import process_audio - backend_module = MagicMock() - backend_module.get_model_info.return_value = { - "model": "medium.en", - "device": "cpu", - "compute_type": "int8", - } - with ( patch( "solstone.observe.transcribe.main.get_config", @@ -92,11 +171,19 @@ def test_empty_statements_preserve_path(raw_path, audio_buffer, vad_result): ), patch("solstone.observe.transcribe.main.stt_transcribe", return_value=[]), patch( - "solstone.observe.transcribe.main.get_backend", return_value=backend_module + "solstone.observe.transcribe.main.get_backend", + return_value=_backend_module(), ), patch("solstone.observe.transcribe.main.callosum_send") as mock_send, ): - process_audio(raw_path, audio_buffer, vad_result, {}, backend="parakeet") + process_audio( + raw_path, + audio_buffer, + vad_result, + {}, + backend="parakeet", + sound_tags=SOUND_TAGS, + ) assert raw_path.exists() jsonl_path = raw_path.with_suffix(".jsonl") @@ -104,21 +191,137 @@ def test_empty_statements_preserve_path(raw_path, audio_buffer, vad_result): header = json.loads(lines[0]) record = header["_solstone_processing"] assert len(lines) == 1 + assert header["sound_tags"] == SOUND_TAGS assert record["state"] == STATE_EMPTY assert record["reason_code"] == REASON_NO_DECODABLE_AUDIO assert mock_send.call_args.args[:2] == ("observe", "transcribed") assert mock_send.call_args.kwargs["outcome"] == "preserved" -def test_vad_no_speech_preserve_path_writes_empty_record(raw_path): +def test_empty_statements_salient_writes_empty_jsonl_then_deletes_audio( + raw_path, + audio_buffer, + vad_result, +): + from solstone.observe.transcribe.main import process_audio + + with ( + patch( + "solstone.observe.transcribe.main.get_config", + return_value={"transcribe": {"preserve_all": False}}, + ), + patch( + "solstone.observe.transcribe.main.get_journal", + return_value=str(raw_path.parents[4]), + ), + patch("solstone.observe.transcribe.main.stt_transcribe", return_value=[]), + patch( + "solstone.observe.transcribe.main.get_backend", + return_value=_backend_module(), + ), + patch("solstone.observe.transcribe.main.callosum_send") as mock_send, + ): + process_audio( + raw_path, + audio_buffer, + vad_result, + {}, + backend="parakeet", + sound_tags=SOUND_TAGS, + ) + + jsonl_path = raw_path.with_suffix(".jsonl") + assert not raw_path.exists() + assert jsonl_path.exists() + assert _read_header(jsonl_path)["sound_tags"] == SOUND_TAGS + assert mock_send.call_args.kwargs["outcome"] == "filtered" + + +def test_empty_statements_non_salient_deletes_without_jsonl( + raw_path, + audio_buffer, + vad_result, +): + from solstone.observe.transcribe.main import process_audio + + with ( + patch( + "solstone.observe.transcribe.main.get_config", + return_value={"transcribe": {"preserve_all": False}}, + ), + patch( + "solstone.observe.transcribe.main.get_journal", + return_value=str(raw_path.parents[4]), + ), + patch("solstone.observe.transcribe.main.stt_transcribe", return_value=[]), + patch( + "solstone.observe.transcribe.main.get_backend", + return_value=_backend_module(), + ), + patch("solstone.observe.transcribe.main.callosum_send") as mock_send, + ): + process_audio( + raw_path, + audio_buffer, + vad_result, + {}, + backend="parakeet", + sound_tags=SILENCE_SOUND_TAGS, + ) + + assert not raw_path.exists() + assert not raw_path.with_suffix(".jsonl").exists() + assert mock_send.call_args.kwargs["outcome"] == "filtered" + + +def test_empty_statements_write_failure_preserves_audio( + raw_path, + audio_buffer, + vad_result, +): + from solstone.observe.transcribe.main import process_audio + + with ( + patch( + "solstone.observe.transcribe.main.get_config", + return_value={"transcribe": {"preserve_all": False}}, + ), + patch( + "solstone.observe.transcribe.main.get_journal", + return_value=str(raw_path.parents[4]), + ), + patch("solstone.observe.transcribe.main.stt_transcribe", return_value=[]), + patch( + "solstone.observe.transcribe.main.get_backend", + return_value=_backend_module(), + ), + patch( + "solstone.observe.transcribe.main.write_text", + side_effect=RuntimeError("disk full"), + ), + patch("solstone.observe.transcribe.main.callosum_send") as mock_send, + ): + with pytest.raises(SystemExit): + process_audio( + raw_path, + audio_buffer, + vad_result, + {}, + backend="parakeet", + sound_tags=SOUND_TAGS, + ) + + assert raw_path.exists() + assert not raw_path.with_suffix(".jsonl").exists() + assert mock_send.call_args.kwargs["outcome"] == "failed" + + +def test_vad_no_speech_preserve_path_writes_empty_record( + raw_path, + no_speech_vad_result, +): from solstone.observe.transcribe.main import _process_one - vad_result = VadResult( - duration=10.0, - speech_duration=0.0, - has_speech=False, - speech_segments=[], - ) args = argparse.Namespace(backend=None, cpu=False, model=None, redo=False) with ( @@ -126,7 +329,8 @@ def test_vad_no_speech_preserve_path_writes_empty_record(raw_path): "solstone.observe.transcribe.main.load_audio", return_value=np.zeros(10 * SAMPLE_RATE, dtype=np.float32), ), - patch("solstone.observe.vad.run_vad", return_value=vad_result), + patch("solstone.observe.vad.run_vad", return_value=no_speech_vad_result), + patch("solstone.observe.transcribe.main.tag_audio", return_value=SOUND_TAGS), patch("solstone.observe.transcribe.main.callosum_send") as mock_send, ): _process_one( @@ -143,6 +347,7 @@ def test_vad_no_speech_preserve_path_writes_empty_record(raw_path): header = json.loads(lines[0]) record = header["_solstone_processing"] assert len(lines) == 1 + assert header["sound_tags"] == SOUND_TAGS assert record["state"] == STATE_EMPTY assert record["reason_code"] == REASON_NO_DECODABLE_AUDIO assert header["backend"] == "unknown" @@ -150,6 +355,147 @@ def test_vad_no_speech_preserve_path_writes_empty_record(raw_path): assert mock_send.call_args.kwargs["outcome"] == "preserved" +def test_vad_no_speech_salient_writes_empty_jsonl_then_deletes_audio( + raw_path, + no_speech_vad_result, +): + from solstone.observe.transcribe.main import _process_one + + args = argparse.Namespace(backend=None, cpu=False, model=None, redo=False) + + with ( + patch( + "solstone.observe.transcribe.main.load_audio", + return_value=np.zeros(10 * SAMPLE_RATE, dtype=np.float32), + ), + patch("solstone.observe.vad.run_vad", return_value=no_speech_vad_result), + patch("solstone.observe.transcribe.main.tag_audio", return_value=SOUND_TAGS), + patch("solstone.observe.transcribe.main.callosum_send") as mock_send, + ): + _process_one( + raw_path, + args, + {"preserve_all": False}, + "parakeet", + [], + ) + + jsonl_path = raw_path.with_suffix(".jsonl") + assert not raw_path.exists() + assert jsonl_path.exists() + assert _read_header(jsonl_path)["sound_tags"] == SOUND_TAGS + assert mock_send.call_args.kwargs["outcome"] == "filtered" + + +def test_vad_no_speech_non_salient_deletes_without_jsonl( + raw_path, + no_speech_vad_result, +): + from solstone.observe.transcribe.main import _process_one + + args = argparse.Namespace(backend=None, cpu=False, model=None, redo=False) + + with ( + patch( + "solstone.observe.transcribe.main.load_audio", + return_value=np.zeros(10 * SAMPLE_RATE, dtype=np.float32), + ), + patch("solstone.observe.vad.run_vad", return_value=no_speech_vad_result), + patch( + "solstone.observe.transcribe.main.tag_audio", + return_value=SILENCE_SOUND_TAGS, + ), + patch("solstone.observe.transcribe.main.callosum_send") as mock_send, + ): + _process_one( + raw_path, + args, + {"preserve_all": False}, + "parakeet", + [], + ) + + assert not raw_path.exists() + assert not raw_path.with_suffix(".jsonl").exists() + assert mock_send.call_args.kwargs["outcome"] == "filtered" + + +def test_vad_no_speech_write_failure_preserves_audio( + raw_path, + no_speech_vad_result, +): + from solstone.observe.transcribe.main import _process_one + + args = argparse.Namespace(backend=None, cpu=False, model=None, redo=False) + + with ( + patch( + "solstone.observe.transcribe.main.load_audio", + return_value=np.zeros(10 * SAMPLE_RATE, dtype=np.float32), + ), + patch("solstone.observe.vad.run_vad", return_value=no_speech_vad_result), + patch("solstone.observe.transcribe.main.tag_audio", return_value=SOUND_TAGS), + patch( + "solstone.observe.transcribe.main.write_text", + side_effect=RuntimeError("disk full"), + ), + patch("solstone.observe.transcribe.main.callosum_send"), + ): + with pytest.raises(RuntimeError, match="disk full"): + _process_one( + raw_path, + args, + {"preserve_all": False}, + "parakeet", + [], + ) + + assert raw_path.exists() + assert not raw_path.with_suffix(".jsonl").exists() + + +def test_vad_no_speech_tagger_raise_degrades_to_existing_filter_path( + raw_path, + no_speech_vad_result, + caplog: pytest.LogCaptureFixture, +): + from solstone.observe.transcribe.main import _process_one + + args = argparse.Namespace(backend=None, cpu=False, model=None, redo=False) + caplog.set_level(logging.WARNING) + + with ( + patch( + "solstone.observe.transcribe.main.load_audio", + return_value=np.zeros(10 * SAMPLE_RATE, dtype=np.float32), + ), + patch("solstone.observe.vad.run_vad", return_value=no_speech_vad_result), + patch( + "solstone.observe.transcribe.main.tag_audio", + side_effect=RuntimeError("tagger bug"), + ), + patch("solstone.observe.transcribe.main.callosum_send") as mock_send, + ): + _process_one( + raw_path, + args, + {"preserve_all": False}, + "parakeet", + [], + ) + + assert not raw_path.exists() + assert not raw_path.with_suffix(".jsonl").exists() + assert mock_send.call_args.kwargs["outcome"] == "filtered" + warnings = [ + record + for record in caplog.records + if record.levelno == logging.WARNING + and "sound tagging failed" in record.message + ] + assert len(warnings) == 1 + + def test_backend_raise_propagates(raw_path, audio_buffer, vad_result): from solstone.observe.transcribe.main import process_audio