tests/test_silence_sensitivity.py

from __future__ import annotations

import json
import math
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / "src"))

try:
    import torch
except ImportError:  # pragma: no cover
    torch = None

from scripts import analyze_silence_sensitivity as analysis  # noqa: E402


class AggregateSummaryTest(unittest.TestCase):
    def test_paired_shift_and_flip_accounting(self) -> None:
        labels = [0, 0, 1, 1]
        baseline = [0.2, 0.8, 0.8, 0.2]
        condition = [0.6, 0.4, 0.4, 0.7]

        shift = analysis._probability_shift(labels, baseline, condition)
        flips = analysis._decision_flips(labels, baseline, condition, threshold=0.5)

        self.assertAlmostEqual(shift["mean"], 0.025)
        self.assertAlmostEqual(shift["mean_absolute"], 0.425)
        self.assertEqual(shift["increased_count"], 2)
        self.assertEqual(shift["decreased_count"], 2)
        self.assertEqual(flips["count"], 4)
        self.assertEqual(flips["HOLD_to_END"], 2)
        self.assertEqual(flips["END_to_HOLD"], 2)
        self.assertEqual(flips["false_interruptions_introduced"], 1)
        self.assertEqual(flips["false_interruptions_resolved"], 1)
        self.assertEqual(flips["missed_ends_introduced"], 1)
        self.assertEqual(flips["missed_ends_resolved"], 1)

    @unittest.skipUnless(torch is not None, "PyTorch is not installed")
    def test_append_then_suffix_crop_is_exact(self) -> None:
        waveforms = [torch.tensor([1.0, 2.0, 3.0]), torch.tensor([4.0])]
        padded, lengths = analysis._append_silence_and_pad(
            waveforms,
            silence_ms=200,
            sample_rate=10,
            max_seconds=0.4,
            pad_side="left",
            torch=torch,
        )

        self.assertEqual(lengths.tolist(), [4, 3])
        self.assertEqual(
            padded.tolist(),
            [[2.0, 3.0, 0.0, 0.0], [0.0, 4.0, 0.0, 0.0]],
        )


@unittest.skipUnless(torch is not None, "PyTorch is not installed")
class SilenceSensitivityCliTest(unittest.TestCase):
    @staticmethod
    def _write_checkpoint(path: Path) -> None:
        from turn_detection.models.features import LogMelConfig
        from turn_detection.models.tiny_tcn import TinyTCNConfig, TinyTurnDetector

        torch.manual_seed(31)
        model_config = TinyTCNConfig(
            n_mels=8,
            channels=4,
            num_blocks=1,
            kernel_size=2,
            dilation_cycle=(1,),
            attention_channels=3,
            head_hidden=3,
            dropout=0.0,
            auxiliary_fillers=False,
        )
        feature_config = LogMelConfig(
            sample_rate=8_000,
            n_fft=64,
            hop_length=32,
            win_length=64,
            n_mels=8,
            f_max=4_000.0,
            normalize=False,
            center=False,
            pad_side="left",
        )
        model = TinyTurnDetector(model_config)
        torch.save(
            {
                "model_state": model.state_dict(),
                "model_config": model.model_config(),
                "threshold": 0.5,
                "metadata": {
                    "feature_config": feature_config.__dict__,
                    "max_seconds": 0.25,
                    "data_scope": "unit-test aggregate",
                    "data_revision": "fixed-test-revision",
                    "run_metadata": {"status": "test"},
                },
            },
            path,
        )

    @staticmethod
    def _write_manifest(path: Path) -> None:
        rows = []
        for index in range(3):
            samples = [
                0.15 * math.sin(2.0 * math.pi * (180 + 20 * index) * sample / 8_000)
                for sample in range(800 + index * 80)
            ]
            rows.append(
                {
                    "record_id": f"private-record-{index}",
                    "split": "validation",
                    "audio": samples,
                    "sample_rate": 8_000,
                    "endpoint": index % 2,
                }
            )
        path.write_text(
            "".join(json.dumps(row) + "\n" for row in rows),
            encoding="utf-8",
        )

    def test_cli_is_deterministic_bounded_and_privacy_safe(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            checkpoint = root / "checkpoint.pt"
            source = root / "manifest.jsonl"
            first_output = root / "first.json"
            second_output = root / "second.json"
            self._write_checkpoint(checkpoint)
            self._write_manifest(source)

            base_command = [
                sys.executable,
                str(ROOT / "scripts/analyze_silence_sensitivity.py"),
                "--checkpoint",
                str(checkpoint),
                "--source",
                str(source),
                "--source-root",
                str(root),
                "--split",
                "validation",
                "--max-examples",
                "2",
                "--batch-size",
                "2",
                "--device",
                "cpu",
            ]
            for output in (first_output, second_output):
                completed = subprocess.run(
                    [*base_command, "--output", str(output)],
                    cwd=ROOT,
                    check=False,
                    capture_output=True,
                    text=True,
                )
                self.assertEqual(completed.returncode, 0, completed.stderr)

            first_text = first_output.read_text(encoding="utf-8")
            second_text = second_output.read_text(encoding="utf-8")
            report = json.loads(first_text)

        self.assertEqual(first_text, second_text)
        self.assertEqual(report["example_count"], 2)
        self.assertEqual(report["positive_count"], 1)
        self.assertEqual(report["negative_count"], 1)
        self.assertEqual(report["selection"]["max_examples"], 2)
        self.assertEqual(
            [item["trailing_silence_ms"] for item in report["conditions"]],
            [0, 200, 400, 800],
        )
        for condition in report["conditions"]:
            self.assertEqual(condition["classification_metrics"]["count"], 2)
            self.assertEqual(condition["probability_summary"]["count"], 2)
        self.assertIsNone(report["conditions"][0]["relative_to_0ms"])
        for condition in report["conditions"][1:]:
            relative = condition["relative_to_0ms"]
            self.assertEqual(relative["probability_shift"]["count"], 2)
            self.assertEqual(
                relative["threshold_decision_flips"]["count"]
                + relative["threshold_decision_flips"]["unchanged_count"],
                2,
            )
        self.assertTrue(report["privacy"]["aggregate_only"])
        self.assertNotIn("private-record", first_text)
        self.assertNotIn("record_id", first_text)


if __name__ == "__main__":
    unittest.main()