|
import time |
|
import win32com.client |
|
|
|
class VoiceText: |
|
""" |
|
Minimal, safe wrapper for SAPI 4 'Speech.VoiceText' as seen on your system. |
|
|
|
Observed behavior: |
|
- Register(site, app): must be called once per COM instance. |
|
- Speak(text, flags): queues if already speaking. Use flags=0. |
|
- Enabled=0 causes Speak() to raise; use 'mute' by simply not calling Speak. |
|
- Speed is clamped by the engine: effective range ~50..300 on your box. |
|
- Transport: Pause/Resume/Stop work; FastForward/Rewind hang (do not use). |
|
""" |
|
def __init__(self, site="pyvoice", app="pyapp", default_speed=100, auto_enable=True): |
|
self._create(site, app, default_speed, auto_enable) |
|
|
|
# --- lifecycle --- |
|
def _create(self, site, app, default_speed, auto_enable): |
|
self.vt = win32com.client.Dispatch("Speech.VoiceText") |
|
self.vt.Register(site, app) # one-time; re-calling on same instance may throw |
|
if auto_enable: |
|
self.vt.Enabled = 1 |
|
self.vt.Speed = int(default_speed) |
|
|
|
def recreate(self, site="pyvoice", app="pyapp", default_speed=100, auto_enable=True): |
|
"""Dispose current COM object and recreate (use to change site/app).""" |
|
try: |
|
del self.vt |
|
except Exception: |
|
pass |
|
self._create(site, app, default_speed, auto_enable) |
|
|
|
# --- properties --- |
|
@property |
|
def enabled(self) -> bool: |
|
try: |
|
return bool(self.vt.Enabled) |
|
except Exception: |
|
return False |
|
|
|
@enabled.setter |
|
def enabled(self, on: bool): |
|
self.vt.Enabled = 1 if on else 0 |
|
|
|
@property |
|
def speed(self) -> int: |
|
return int(self.vt.Speed) |
|
|
|
@speed.setter |
|
def speed(self, val: int): |
|
# clamp to observed effective range |
|
v = max(50, min(int(val), 300)) |
|
self.vt.Speed = v |
|
|
|
def is_speaking(self) -> bool: |
|
try: |
|
return bool(self.vt.IsSpeaking) |
|
except Exception: |
|
return False |
|
|
|
# --- core ops --- |
|
def speak_async(self, text: str, flags: int = 0): |
|
"""Queue text and return immediately.""" |
|
if not self.enabled: |
|
# mirror engine behavior with clearer message |
|
raise RuntimeError("VoiceText.Enabled==0: engine will reject Speak(). Enable first.") |
|
self.vt.Speak(text, flags) |
|
|
|
def speak_and_wait(self, text: str, flags: int = 0, poll: float = 0.02): |
|
"""Queue text and block until all speech finishes.""" |
|
self.speak_async(text, flags) |
|
self.wait_finish(poll=poll) |
|
|
|
def stop(self): |
|
"""Stop current and flush any queued utterances.""" |
|
try: |
|
self.vt.StopSpeaking() |
|
finally: |
|
time.sleep(0.12) # allow buffer to settle |
|
|
|
def pause(self): self.vt.AudioPause() |
|
def resume(self): self.vt.AudioResume() |
|
|
|
# --- helpers --- |
|
def purge(self): |
|
"""Hard flush of current+queued audio; use before 'exclusive' speaks.""" |
|
self.stop() |
|
|
|
def speak_exclusive(self, text: str, flags: int = 0, poll: float = 0.02): |
|
"""Flush queue, speak text, wait for completion (single-flight).""" |
|
if not self.enabled: |
|
self.enabled = 1 |
|
self.purge() |
|
self.speak_async(text, flags) |
|
self.wait_finish(poll=poll) |
|
|
|
def speak_batch(self, lines, inter_gap: float = 0.0, flags: int = 0): |
|
"""Queue multiple lines in order; optional small gap between enqueues.""" |
|
if not self.enabled: |
|
self.enabled = 1 |
|
for line in lines: |
|
self.vt.Speak(line, flags) |
|
if inter_gap: |
|
time.sleep(inter_gap) |
|
|
|
def wait_start(self, timeout: float = 3.0, poll: float = 0.01) -> bool: |
|
t0 = time.monotonic() |
|
while time.monotonic() - t0 < timeout: |
|
if self.is_speaking(): |
|
return True |
|
time.sleep(poll) |
|
return False |
|
|
|
def wait_finish(self, poll: float = 0.02): |
|
while self.is_speaking(): |
|
time.sleep(poll) |
|
|
|
# quick demo |
|
if __name__ == "__main__": |
|
tts = VoiceText() |
|
tts.speak_and_wait("Hello from SAPI four via Python.") |
|
tts.speak_async("This will queue.") |
|
tts.speak_async("And play after the previous sentence.") |
|
tts.wait_finish() |