Add pinned local Qwen GPU voice generation and first puzzle drafts

tools/audio/test_generate_voice.py

@@ -0,0 +1,118 @@
+import copy
+import importlib.util
+import json
+from pathlib import Path
+import tempfile
+import unittest
+
+import numpy as np
+import soundfile as sf
+
+SPEC = importlib.util.spec_from_file_location('generate_voice', Path(__file__).with_name('generate_voice.py'))
+voice = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(voice)
+
+
+class VoiceHistoryTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.root = Path(self.temp.name)
+        self.catalog = {'schemaVersion': 1, 'clips': [
+            {'id': 'count_01', 'text': '하나', 'usage': 'count', 'stageIds': [1], 'required': True},
+            {'id': 'guide_collect', 'text': '친구를 찾아 끌어와 볼까?', 'usage': 'guide', 'stageIds': [1], 'required': True}]}
+        self.profile = {'schemaVersion': 1, 'modelId': voice.MODEL_ID, 'modelRevision': 'a' * 40,
+                        'language': 'Korean', 'speaker': 'Sohee', 'instruct': '천천히 읽어 주세요.',
+                        'device': 'cuda:0', 'dtype': 'bfloat16', 'attention': 'sdpa', 'seed': 42,
+                        'generationParameters': {'max_new_tokens': 2048}}
+        self.calls = []
+
+    def tearDown(self):
+        self.temp.cleanup()
+
+    def fake(self, clip):
+        self.calls.append(clip['id'])
+        return np.sin(np.linspace(0, 20, 2400)).astype(np.float32) * .2, 24000
+
+    def run_generation(self, catalog=None, profile=None, ids=None):
+        return voice.generate(catalog or self.catalog, profile or self.profile, self.root,
+                              ids=ids, synthesize=self.fake, environment={'test': True})
+
+    def test_partial_changed_clip_regenerates_and_preserves_original(self):
+        first = self.run_generation()
+        old = (self.root / first['clips'][0]['file']).read_bytes()
+        revised = copy.deepcopy(self.catalog)
+        revised['clips'][1]['text'] = '친구를 끌어와 볼까?'
+        second = self.run_generation(catalog=revised)
+        self.assertEqual(['reused', 'generated'], [c['result'] for c in second['clips']])
+        self.assertEqual(['count_01', 'guide_collect', 'guide_collect'], self.calls)
+        self.assertEqual(old, (self.root / first['clips'][0]['file']).read_bytes())
+        self.assertNotEqual(first['runId'], second['runId'])
+        self.assertEqual([], voice.validate_manifest(second, revised, self.profile, self.root))
+
+    def test_empty_text_duplicate_ids_and_bad_stage_rejected(self):
+        for edit in ('empty', 'duplicate', 'stage'):
+            bad = copy.deepcopy(self.catalog)
+            if edit == 'empty': bad['clips'][0]['text'] = ' '
+            if edit == 'duplicate': bad['clips'][1]['id'] = 'count_01'
+            if edit == 'stage': bad['clips'][0]['stageIds'] = [31]
+            with self.assertRaises(ValueError): self.run_generation(catalog=bad)
+        self.assertEqual([], self.calls)
+
+    def test_missing_corrupt_and_profile_mismatch_cannot_validate(self):
+        manifest = self.run_generation()
+        changed = copy.deepcopy(self.profile)
+        changed['instruct'] = '밝게 읽어 주세요.'
+        self.assertTrue(voice.validate_manifest(manifest, self.catalog, changed, self.root))
+        wav = self.root / manifest['clips'][0]['file']
+        wav.write_bytes(b'not wav')
+        self.assertTrue(voice.validate_manifest(manifest, self.catalog, self.profile, self.root))
+        wav.unlink()
+        self.assertTrue(voice.validate_manifest(manifest, self.catalog, self.profile, self.root))
+
+    def test_partial_scope_does_not_claim_all_required(self):
+        manifest = self.run_generation(ids=['count_01'])
+        self.assertEqual('selected', manifest['scope'])
+        self.assertEqual([], voice.validate_manifest(manifest, self.catalog, self.profile, self.root, ['count_01']))
+        self.assertTrue(voice.validate_manifest(manifest, self.catalog, self.profile, self.root))
+
+    def test_failed_or_invalid_waveform_not_success(self):
+        def broken(clip): return np.array([np.nan]), 24000
+        manifest = voice.generate(self.catalog, self.profile, self.root, synthesize=broken)
+        self.assertEqual(['failed', 'failed'], [c['result'] for c in manifest['clips']])
+        self.assertTrue(voice.validate_manifest(manifest, self.catalog, self.profile, self.root))
+
+    def test_changed_profile_and_tampered_waveform_force_regeneration(self):
+        first = self.run_generation()
+        path = self.root / first['clips'][0]['file']
+        sf.write(path, np.zeros(2400), 24000)
+        second = self.run_generation()
+        self.assertEqual(['generated', 'reused'], [c['result'] for c in second['clips']])
+        revised = copy.deepcopy(self.profile)
+        revised['seed'] = 43
+        third = self.run_generation(profile=revised)
+        self.assertEqual(['generated', 'generated'], [c['result'] for c in third['clips']])
+
+    def test_unknown_requested_id_rejected(self):
+        with self.assertRaises(ValueError): self.run_generation(ids=['missing'])
+
+    def test_pcm_export_resamples_and_preserves_existing_runtime(self):
+        import sys
+        sys.path.insert(0, str(Path(__file__).parent))
+        from prepare_voice import prepare
+        manifest = self.run_generation()
+        target = self.root / 'runtime'
+        selection = prepare(manifest, self.catalog, self.profile, self.root, target)
+        info = sf.info(target / 'count_01.wav')
+        self.assertEqual((44100, 1, 'PCM_16'), (info.samplerate, info.channels, info.subtype))
+        self.assertAlmostEqual(.1, info.duration)
+        self.assertEqual('not_listened', selection['selections'][0]['review'])
+        original = (target / 'count_01.wav').read_bytes()
+        sf.write(target / 'count_01.wav', np.ones(4410) * .1, 44100)
+        changed = (target / 'count_01.wav').read_bytes()
+        with self.assertRaises(ValueError): prepare(manifest, self.catalog, self.profile, self.root, target)
+        self.assertEqual(changed, (target / 'count_01.wav').read_bytes())
+        self.assertNotEqual(original, changed)
+
+
+if __name__ == '__main__':
+    unittest.main()