Add configurable TTS and STT residency policies

test_audio_policy.py

@@ -0,0 +1,36 @@
+import tempfile
+import unittest
+from pathlib import Path
+from audio_policy import AudioPolicy
+
+class Worker:
+ def __init__(self):
+  import threading
+  self.lock=threading.RLock();self.policy='auto';self.job=None;self.process=None;self.warming=False;self.unloaded=0
+ def unload_idle(self):self.unloaded+=1
+
+class Profiles:
+ def status(self):return {'profiles':[{'id':'tts-1','kind':'audio','runnable':True},{'id':'stt-1','kind':'stt','runnable':True}]}
+
+class Endpoint:
+ state='stopped'
+
+class Scheduler:
+ def __init__(self):
+  import threading
+  self.cv=threading.Condition();self.gpu_mode='llm';self.active=0;self.transition=False
+
+class PolicyTests(unittest.TestCase):
+ def test_persist_validate_and_unload(self):
+  with tempfile.TemporaryDirectory() as directory:
+   path=Path(directory)/'audio-policy.json';tts=Worker();stt=Worker()
+   manager=AudioPolicy(path,Profiles(),tts,stt,Scheduler(),Endpoint())
+   with self.assertRaises(ValueError):manager.configure({'kind':'tts','mode':'warm','profile_id':'missing'})
+   self.assertFalse(path.exists())
+   manager.configure({'kind':'tts','mode':'warm','profile_id':'tts-1'})
+   self.assertEqual(tts.policy,'warm')
+   manager.configure({'kind':'stt','mode':'per_request','profile_id':None})
+   self.assertEqual(stt.unloaded,1)
+   restored=AudioPolicy(path,Profiles(),Worker(),Worker(),Scheduler(),Endpoint())
+   self.assertEqual(restored.status()['settings']['tts']['mode'],'warm')
+   self.assertEqual(restored.status()['settings']['stt']['mode'],'per_request')