fix: normalize ambience loudness for v0.2.5

2026-10-14 12:12:13 +00:00
parent 812540c743
commit 91aff66445
390 changed files with 97709 additions and 257 deletions
@@ -71,7 +71,7 @@ class AmbienceConversion(unittest.TestCase):
            self.assertEqual(actual["conversion"]["seekMethod"], "full decode then sample trim")
            expected["conversion"]["seekMethod"] = actual["conversion"]["seekMethod"]
            self.assertEqual(actual, expected)
-            self.assertEqual(len(calls), 2)
+            self.assertEqual(len(calls), 5)
            self.assertLess(calls[0].index("-ss"), calls[0].index("-i"))
            self.assertNotIn("-ss", calls[1])
            self.assertIn("aresample=24000,atrim=start_sample=24000:end_sample=312000,asetpts=N/SR/TB", calls[1])
@@ -286,6 +286,31 @@ class AtomicPublication(unittest.TestCase):
                    with self.assertRaises(ValueError): self.run_generate()
                self.assert_prior()

+    def test_real_loudness_measurement_failure_preserves_prior_directory(self):
+        """최종 실측 FFmpeg만 실패시켜 원자 교환 전 실패 전파를 확인한다."""
+        import subprocess
+        from unittest.mock import patch
+        run = subprocess.run
+        def fail_measurement(command, **kwargs):
+            if any("ebur128=" in str(arg) for arg in command):
+                raise subprocess.CalledProcessError(1, command, stderr=b"injected measurement failure")
+            return run(command, **kwargs)
+        with patch.object(audio.subprocess, "run", side_effect=fail_measurement):
+            with self.assertRaises(subprocess.CalledProcessError): self.run_generate()
+        self.assert_prior()
+
+    def test_actual_target_or_true_peak_violation_preserves_prior_directory(self):
+        """manifest가 정상이라고 주장해도 별도 실측 위반은 공개하지 않는다."""
+        from unittest.mock import patch
+        for metric, value in (("integratedLUFS", -34.0), ("truePeakDBTP", -5.0), ("integratedLUFS", float("nan"))):
+            with self.subTest(metric=metric):
+                stats = {"integratedLUFS": -32.5, "truePeakDBTP": -7.0,
+                         "loudnessRangeLU": 1.0, "thresholdLUFS": -42.5}
+                stats[metric] = value
+                with patch.object(audio, "measure_loudness", return_value=stats, create=True):
+                    with self.assertRaises(ValueError): self.run_generate()
+                self.assert_prior()
+
    def test_exchange_failure_preserves_prior_directory(self):
        from unittest.mock import patch
        import ctypes
@@ -0,0 +1,78 @@
+"""선형 gain만 허용하고 과도 peak·비수렴·측정 실패는 공개 전에 거부한다."""
+from array import array
+import math
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+import wave
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / 'scripts'))
+import ambience_audio as audio
+
+
+def stats(i=-32.5, tp=-15):
+    return dict(integratedLUFS=i, truePeakDBTP=tp, loudnessRangeLU=1.0, thresholdLUFS=i-10)
+
+
+class LinearNormalization(unittest.TestCase):
+    def setUp(self):
+        self.temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temporary.cleanup)
+        self.target = Path(self.temporary.name) / 'staged.wav'
+        values = [0.03 * math.sin(i / 20) for i in range(24000 * 12)]
+        values[0] = values[-1] = 0
+        self.samples = array('f', values)
+        self.pcm = self.samples.tobytes()
+
+    def test_real_output_is_only_scalar_gain_with_pcm16_rounding(self):
+        result = audio.normalize_pcm(self.pcm, self.target)
+        self.assertEqual(len(result['iterations']), 1)
+        gain = 10 ** (result['iterations'][0]['gainDB'] / 20)
+        with wave.open(str(self.target)) as stream:
+            actual = array('h', stream.readframes(stream.getnframes()))
+        # 리미터/컴프레서 없이 전 sample이 동일 gain이며 PCM16 반올림 오차만 허용한다.
+        self.assertEqual(len(actual), len(self.samples))
+        self.assertLessEqual(max(abs(x / 32768 - y * gain) for x, y in zip(actual, self.samples)), 0.5001 / 32768)
+        self.assertLessEqual(abs(result['final']['integratedLUFS'] + 32.5), 0.2 + 1e-9)
+        self.assertLessEqual(result['final']['truePeakDBTP'], -6.2)
+
+    def test_unsafe_predicted_peak_is_rejected_before_volume_conversion(self):
+        with patch.object(audio, 'measure_loudness', return_value=stats(-40, -10)), patch.object(audio.subprocess, 'run') as run:
+            with self.assertRaisesRegex(ValueError, 'peak'):
+                audio.normalize_pcm(self.pcm, self.target)
+            run.assert_not_called()
+        self.assertFalse(self.target.exists())
+
+    def test_correction_restarts_from_identical_source_and_records_both_passes(self):
+        real_run = subprocess.run
+        with patch.object(audio, 'measure_loudness', side_effect=[stats(-35), stats(-32.9), stats()]), patch.object(audio.subprocess, 'run', wraps=real_run) as run:
+            result = audio.normalize_pcm(self.pcm, self.target)
+        self.assertEqual(len(result['iterations']), 2)
+        self.assertEqual([r['gainDB'] for r in result['iterations']], [2.5, 2.9])
+        self.assertEqual([c.kwargs['input'] for c in run.call_args_list], [self.pcm, self.pcm])
+        for call, expected in zip(run.call_args_list, ['volume=2.500000000000dB:precision=double', 'volume=2.900000000000dB:precision=double']):
+            self.assertIn(expected, call.args[0])
+
+    def test_nonconvergence_is_bounded_and_fails(self):
+        with patch.object(audio, 'measure_loudness', side_effect=[stats(-35), stats(-33), stats(-33), stats(-33)]) as measure:
+            with self.assertRaisesRegex(ValueError, '수렴'):
+                audio.normalize_pcm(self.pcm, self.target)
+            self.assertEqual(measure.call_count, 4)
+
+    def test_each_corrective_gain_is_peak_checked_before_conversion(self):
+        with patch.object(audio, 'measure_loudness', side_effect=[stats(-35, -9), stats(-33)]), patch.object(audio.subprocess, 'run', wraps=subprocess.run) as run:
+            with self.assertRaisesRegex(ValueError, 'peak'):
+                audio.normalize_pcm(self.pcm, self.target)
+            self.assertEqual(run.call_count, 1)
+
+    def test_final_peak_and_nonfinite_measurement_are_rejected(self):
+        for invalid in [stats(tp=-6.1), stats(i=float('nan')), stats(tp=float('inf'))]:
+            with self.subTest(invalid=invalid), patch.object(audio, 'measure_loudness', side_effect=[stats(-35), invalid]):
+                with self.assertRaises(ValueError): audio.normalize_pcm(self.pcm, self.target)
+
+    def test_parser_rejects_missing_nonfinite_or_partial_summary(self):
+        for text in ['', 'Summary:\nI: -32.5 LUFS', 'Summary:\nI: nan LUFS\nThreshold: -42.5 LUFS\nLRA: 0.0 LU\nPeak: -10.0 dBFS']:
+            with self.subTest(text=text), self.assertRaises(ValueError): audio.parse_loudness(text)
@@ -0,0 +1,76 @@
+"""생성기와 독립된 새 ebur128 프로세스가 정확한 최종 WAV의 릴리스 계기다."""
+import hashlib
+import json
+import math
+from pathlib import Path
+import re
+import subprocess
+import unittest
+
+# manifest가 목표를 바꾸어 실패를 숨길 수 없도록 수용값을 독립적으로 고정한다.
+TARGET = -32.5
+CEILING = -6.0
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def independent_measure(path):
+    """정규화 필터의 자기 보고 없이 저장된 전체 PCM을 전용 EBU 계기로 측정한다."""
+    command = ["ffmpeg", "-nostdin", "-hide_banner", "-i", str(path), "-af",
+               "ebur128=peak=true", "-f", "null", "-"]
+    result = subprocess.run(command, check=True, capture_output=True, text=True, timeout=30)
+    summary = result.stderr.rsplit("Summary:", 1)[1]
+    values = {
+        "integratedLUFS": float(re.search(r"I:\s+([-\d.]+) LUFS", summary)[1]),
+        "truePeakDBTP": float(re.search(r"Peak:\s+([-\d.]+) dBFS", summary)[1]),
+        "loudnessRangeLU": float(re.search(r"LRA:\s+([-\d.]+) LU", summary)[1]),
+        "thresholdLUFS": float(re.search(r"Threshold:\s+([-\d.]+) LUFS", summary)[1]),
+    }
+    if not all(math.isfinite(v) for v in values.values()):
+        raise ValueError("non-finite ebur128 measurement")
+    return values
+
+
+class FinalLoudness(unittest.TestCase):
+    """8 WAV의 동등성 assertion은 계기 불일치를 이유로 면제하지 않는다."""
+    @classmethod
+    def setUpClass(cls):
+        cls.rows = {p.stem: independent_measure(p) for p in sorted((ROOT / "public/ambience").glob("*.wav"))}
+
+    def test_final_wavs_meet_fixed_loudness_and_true_peak(self):
+        self.assertEqual(len(self.rows), 8)
+        spread = max(r["integratedLUFS"] for r in self.rows.values()) - min(r["integratedLUFS"] for r in self.rows.values())
+        print(json.dumps({"measurementFilter": "ebur128=peak=true", "targetLUFS": TARGET,
+                          "truePeakCeilingDBTP": CEILING, "spreadLU": round(spread, 2),
+                          "files": self.rows}, indent=2), flush=True)
+        for name, row in self.rows.items():
+            with self.subTest(environment=name):
+                self.assertLessEqual(abs(row["integratedLUFS"] - TARGET), 0.2 + 1e-9)
+                self.assertLessEqual(row["truePeakDBTP"], CEILING)
+        # 삭제되었던 ebur128 integrated 동등성 검사를 릴리스 필수 조건으로 복원한다.
+        self.assertLessEqual(spread, 0.4 + 1e-9)
+
+    def test_manifest_records_linear_gain_and_exact_ebur128_measurements(self):
+        manifest = json.loads((ROOT / "public/ambience/manifest.json").read_text())
+        settings = manifest["normalization"]
+        for key, value in {"targetIntegratedLUFS": TARGET, "truePeakCeilingDBTP": CEILING,
+                           "internalTruePeakCeilingDBTP": -6.2, "toleranceLU": 0.2,
+                           "maximumSpreadLU": 0.4, "measurementFilter": "ebur128=peak=true",
+                           "maximumIterations": 3}.items():
+            self.assertEqual(settings[key], value)
+        self.assertIn("ffmpeg version", manifest["toolVersions"]["ffmpeg"])
+        for entry in manifest["files"]:
+            norm = entry["conversion"]["loudness"]
+            self.assertEqual(norm["measurementFilter"], "ebur128=peak=true")
+            self.assertRegex(norm["fadedPcmSha256"], r"^[a-f0-9]{64}$")
+            self.assertGreaterEqual(len(norm["iterations"]), 1)
+            self.assertLessEqual(len(norm["iterations"]), 3)
+            gain = TARGET - norm["input"]["integratedLUFS"]
+            for iteration in norm["iterations"]:
+                self.assertAlmostEqual(iteration["gainDB"], gain, places=9)
+                self.assertEqual(iteration["filter"], f"volume={gain:.12f}dB:precision=double")
+                self.assertLessEqual(norm["input"]["truePeakDBTP"] + gain, -6.2 + 1e-9)
+                gain += TARGET - iteration["output"]["integratedLUFS"]
+            self.assertEqual(norm["final"], norm["iterations"][-1]["output"])
+            # manifest 수치는 새 프로세스 실측과 정확히 같아야 한다.
+            self.assertEqual(norm["final"], self.rows[entry["environmentId"]])
+            self.assertEqual(entry["sha256"], hashlib.sha256((ROOT / "public/ambience" / entry["file"]).read_bytes()).hexdigest())
@@ -0,0 +1,180 @@
+"""실제 승인/ZIP을 임시 사본에서 변조해 fail-closed 릴리스 계약을 검사한다."""
+import copy
+import json
+from pathlib import Path
+import shutil
+import tempfile
+import unittest
+
+from scripts import release_approval as gate
+
+
+class ReleaseApprovalTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+        # 실제 증거에 쓰지 않는다. 자동 성공 fixture도 이 임시 사본에만 존재한다.
+        for name in (gate.APPROVAL, gate.ASR, gate.PRIOR_ASR, gate.BUNDLE,
+                     *(f"public/ambience/{name}" for name in gate.SCOPE["files"])):
+            target = self.root / name
+            target.parent.mkdir(parents=True, exist_ok=True)
+            shutil.copyfile(gate.ROOT / name, target)
+        self.approval = gate.read_json(self.root / gate.APPROVAL)
+
+    def write(self, path, value):
+        (self.root / path).write_text(json.dumps(value, ensure_ascii=False) + "\n")
+
+    def rejected(self):
+        with self.assertRaises((ValueError, KeyError, TypeError, OSError)):
+            gate.validate_release(self.root)
+
+    def synthetic_empty_screening(self):
+        """OR의 자동 분기 검사용 fixture. 실제 보고서는 절대 수정하지 않는다."""
+        screening = gate.read_json(self.root / gate.ASR)
+        for row in screening["files"]:
+            row["segments"] = []
+            row["text"] = ""
+            row["info"]["duration_after_vad"] = 0
+        screening["requiredEmptyPassed"] = True
+        self.write(gate.ASR, screening)
+
+    def test_current_asr_failure_passes_only_through_exact_owner_approval(self):
+        result = gate.validate_release(self.root)
+        self.assertEqual(result["acceptancePath"], "exact-byte-owner-listening-approval")
+        self.assertFalse(result["automatedEmptyAsrPassed"])
+        self.assertTrue(result["humanListeningAcceptance"])
+        self.assertFalse(result["legalPrivacyClearance"])
+        (self.root / gate.APPROVAL).unlink()
+        self.rejected()
+
+    def test_automated_empty_criteria_can_pass_without_owner_approval(self):
+        self.synthetic_empty_screening()
+        (self.root / gate.APPROVAL).unlink()
+        result = gate.validate_release(self.root)
+        self.assertEqual(result["acceptancePath"], "automated-empty-asr")
+        self.assertFalse(result["humanListeningAcceptance"])
+        self.assertFalse(result["legalPrivacyClearance"])
+
+    def test_every_missing_approval_field_fails(self):
+        for key in self.approval:
+            with self.subTest(key=key):
+                changed = copy.deepcopy(self.approval)
+                del changed[key]
+                self.write(gate.APPROVAL, changed)
+                self.rejected()
+
+    def test_statement_owner_scope_timestamp_and_clearance_cannot_be_reinterpreted(self):
+        mutations = {
+            "owner": ["Neo", "oz", ""],
+            "statement": ["승인합니다", "승인합니다. 이대로 진행하세요 ", ""],
+            "timestampUtc": ["", "not-a-date", "2026-10-14T11:29:50Z"],
+            "timestampKst": ["", "2026-10-14T11:29:51+09:00"],
+            "scope": [{}, {**gate.SCOPE, "version": "0.2.3"}, {**gate.SCOPE, "files": ["market.wav"]}],
+            "productOwnerReleaseApproval": [False, "true", 1],
+            "humanListeningAcceptance": [False, "true", 1],
+            "legalPrivacyClearance": [True, "false", 0, None],
+        }
+        for key, values in mutations.items():
+            for value in values:
+                with self.subTest(key=key, value=value):
+                    self.write(gate.APPROVAL, {**self.approval, key: value})
+                    self.rejected()
+
+    def test_approval_hash_mismatches_and_incomplete_asset_scope_fail(self):
+        for field in ("reviewBundle", "manifest", "notice"):
+            with self.subTest(field=field):
+                changed = copy.deepcopy(self.approval)
+                changed[field]["sha256"] = "0" * 64
+                self.write(gate.APPROVAL, changed)
+                self.rejected()
+        for index in range(8):
+            with self.subTest(wav=index):
+                changed = copy.deepcopy(self.approval)
+                changed["wavs"][index]["sha256"] = "0" * 64
+                self.write(gate.APPROVAL, changed)
+                self.rejected()
+        for wavs in (self.approval["wavs"][:-1], self.approval["wavs"] + self.approval["wavs"][:1]):
+            self.write(gate.APPROVAL, {**self.approval, "wavs": wavs})
+            self.rejected()
+
+    def test_current_wavs_manifest_notice_and_bundle_are_bound_to_reviewed_bytes(self):
+        for name in (gate.BUNDLE, *(f"public/ambience/{name}" for name in gate.SCOPE["files"])):
+            with self.subTest(file=name):
+                path = self.root / name
+                original = path.read_bytes()
+                path.write_bytes(original + b" ")
+                self.rejected()
+                path.write_bytes(original)
+
+    def test_manifest_reformat_cannot_be_self_approved_by_updating_all_recorded_hashes(self):
+        path = self.root / gate.MANIFEST
+        path.write_bytes(path.read_bytes() + b"\n")
+        digest = gate.sha256(path.read_bytes())
+        self.approval["manifest"]["sha256"] = digest
+        self.write(gate.APPROVAL, self.approval)
+        screening = gate.read_json(self.root / gate.ASR)
+        screening["method"]["manifestSha256"] = digest
+        self.write(gate.ASR, screening)
+        self.rejected()  # 전달 ZIP의 manifest 바이트는 여전히 다르다.
+
+    def test_wav_change_cannot_be_self_approved_by_updating_manifest_asr_and_approval(self):
+        path = self.root / "public/ambience/market.wav"
+        raw = bytearray(path.read_bytes())
+        raw[100] ^= 1
+        path.write_bytes(raw)
+        digest = gate.sha256(raw)
+        manifest = gate.read_json(self.root / gate.MANIFEST)
+        screening = gate.read_json(self.root / gate.ASR)
+        for rows in (manifest["files"], screening["files"], self.approval["wavs"]):
+            next(row for row in rows if row["environmentId"] == "market")["sha256"] = digest
+        self.write(gate.MANIFEST, manifest)
+        manifest_hash = gate.sha256((self.root / gate.MANIFEST).read_bytes())
+        screening["method"]["manifestSha256"] = manifest_hash
+        self.approval["manifest"]["sha256"] = manifest_hash
+        self.write(gate.ASR, screening)
+        self.write(gate.APPROVAL, self.approval)
+        self.rejected()
+
+    def test_different_bundle_hash_cannot_be_self_approved(self):
+        path = self.root / gate.BUNDLE
+        path.write_bytes(path.read_bytes() + b" ")
+        self.approval["reviewBundle"]["sha256"] = gate.sha256(path.read_bytes())
+        self.write(gate.APPROVAL, self.approval)
+        self.rejected()
+
+    def test_invalid_approval_is_not_ignored_even_when_automated_criteria_pass(self):
+        self.synthetic_empty_screening()
+        for key, value in (("legalPrivacyClearance", True), ("scope", {}), ("timestampUtc", "")):
+            self.write(gate.APPROVAL, {**self.approval, key: value})
+            self.rejected()
+
+    def test_nonempty_asr_cannot_be_relabeled_pass(self):
+        screening = gate.read_json(self.root / gate.ASR)
+        screening["requiredEmptyPassed"] = True
+        self.write(gate.ASR, screening)
+        self.rejected()
+
+    def test_stale_asr_settings_and_clearance_overclaims_fail_with_valid_owner_approval(self):
+        original = gate.read_json(self.root / gate.ASR)
+        for field, key, value in (("method", "manifestSha256", "0" * 64),
+                                  ("method", "parameters", {}),
+                                  ("limitations", "legalPrivacyClearance", True),
+                                  ("limitations", "textIsVerifiedTranscript", True)):
+            changed = copy.deepcopy(original)
+            changed[field][key] = value
+            self.write(gate.ASR, changed)
+            self.rejected()
+
+    def test_unknown_clearance_claim_and_duplicate_json_keys_fail(self):
+        self.write(gate.APPROVAL, {**self.approval, "legalClearance": True})
+        self.rejected()
+        raw = json.dumps(self.approval)
+        (self.root / gate.APPROVAL).write_text(raw[:-1] + ', "legalPrivacyClearance": false}')
+        self.rejected()
+        (self.root / gate.APPROVAL).write_text('{"statement":')
+        self.rejected()
+
+
+if __name__ == "__main__":
+    unittest.main()