Integrate approved voice records and complete reuse environment history

tools/audio/generate_voice.py

@@ -6,6 +6,7 @@ import importlib.metadata
 import json
 from pathlib import Path
 import re
+import subprocess
 import sys
 import time
 import uuid
@@ -147,6 +148,27 @@ def reusable_entries(root, catalog, profile):
     return result
 
 
+def reuse_environment(profile, entries, root):
+    environment = {'python': sys.version, 'packages': {name: importlib.metadata.version(name) for name in
+                   ('qwen-tts', 'torch', 'torchaudio', 'transformers', 'soundfile', 'huggingface-hub')},
+                   'modelLoad': 'not_needed', 'requestedDevice': profile['device'], 'sourceGenerations': []}
+    try:
+        probe = subprocess.run(['nvidia-smi', '-i', '0', '--query-gpu=name', '--format=csv,noheader'],
+                               check=True, capture_output=True, text=True, timeout=5)
+        environment['device'] = probe.stdout.strip()
+        environment['deviceProbe'] = 'nvidia-smi; no model or CUDA inference executed'
+    except (OSError, subprocess.SubprocessError) as error:
+        environment['deviceProbeError'] = str(error)
+    for run_id in sorted({entry['originalRunId'] for entry in entries.values()}):
+        manifest_path = Path(root) / 'generated' / run_id / 'manifest.json'
+        source = load_json(manifest_path)
+        environment['sourceGenerations'].append({'runId': run_id,
+                                                'manifest': manifest_path.relative_to(root).as_posix(),
+                                                'manifestSha256': file_hash(manifest_path),
+                                                'environment': source['environment']})
+    return environment
+
+
 def generate(catalog, profile, root, ids=None, synthesize=None, environment=None, regenerate=False):
     validate_inputs(catalog, profile)
     selected = select_clips(catalog, ids)
@@ -252,7 +274,11 @@ def main():
         else:
             selected = select_clips(catalog, ids)
             reusable = reusable_entries(root, catalog, profile)
-            synthesize, environment = (None, {'modelLoad': 'not_needed'}) if not args.regenerate and all(c['id'] in reusable for c in selected) else load_synthesizer(profile)
+            if not args.regenerate and all(c['id'] in reusable for c in selected):
+                synthesize = None
+                environment = reuse_environment(profile, {c['id']: reusable[c['id']] for c in selected}, root)
+            else:
+                synthesize, environment = load_synthesizer(profile)
             manifest = generate(catalog, profile, root, ids, synthesize, environment, args.regenerate)
             print('Manifest: ' + str(root / 'generated' / manifest['runId'] / 'manifest.json'))
             errors = validate_manifest(manifest, catalog, profile, root, ids)

tools/audio/test_generate_voice.py

@@ -3,6 +3,7 @@ import importlib.util
 from pathlib import Path
 import tempfile
 import unittest
+from unittest import mock
 
 import numpy as np
 import soundfile as sf
@@ -134,6 +135,23 @@ class VoiceHistoryTests(unittest.TestCase):
         self.assertEqual(original, (self.root / first['clips'][0]['file']).read_bytes())
         self.assertEqual(['count_01', 'guide_collect', 'count_01'], self.calls)
 
+    def test_reuse_environment_records_packages_device_and_original_run_without_model_load(self):
+        first = self.run_generation()
+        entries = voice.reusable_entries(self.root, self.catalog, self.profile)
+        with mock.patch.object(voice.importlib.metadata, 'version', return_value='test-version'), \
+                mock.patch.object(voice.subprocess, 'run', return_value=mock.Mock(stdout='Test GPU\n')):
+            environment = voice.reuse_environment(self.profile, entries, self.root)
+        second = voice.generate(self.catalog, self.profile, self.root, environment=environment)
+        self.assertEqual(['reused', 'reused'], [c['result'] for c in second['clips']])
+        self.assertEqual(['count_01', 'guide_collect'], self.calls)
+        self.assertEqual('not_needed', second['environment']['modelLoad'])
+        self.assertEqual('Test GPU', second['environment']['device'])
+        self.assertEqual('test-version', second['environment']['packages']['qwen-tts'])
+        source = second['environment']['sourceGenerations'][0]
+        self.assertEqual(first['runId'], source['runId'])
+        self.assertEqual(voice.file_hash(self.root / source['manifest']), source['manifestSha256'])
+        self.assertEqual({'test': True}, source['environment'])
+
 
 if __name__ == '__main__':
     unittest.main()