python - 使用线程时未触发 Pyttsx3 回调

Question

我使用线程作为绕过 pyttsx3 的“运行和等待”默认功能的一种方式，这样我就可以在讲话中打断正在说的话。但是，这会阻止库的回调，因此我无法检测到一段文本何时完成 - 我必须使用长度（以秒为单位）来猜测它。

我希望检测的回调是finished-utterance，它与self.engine.connect('finished-utterance', self.onEnd). 这遵循文档中给出的示例。当我在实际程序中使用它时，一个文本块可能有数百个单词长，而下一个文本块可能只有几个单词长。我想知道一个人什么时候说完，这样程序就可以自动前进到下一个“说”功能。

我读过有人通过使用多处理成功解决了这个问题，但我不知道该怎么做。有没有办法让回调与线程一起工作？我正在使用 Windows 10

from threading import Event, Thread
import pyttsx3

class Voice(object):
    def __init__(self, skip, play_beep):
        self.t = None
        self._running = False
        self.engine = pyttsx3.init()
        self.skip = skip
        self.engine.connect('finished-utterance', self.onEnd)

    def onEnd(self, name, completed):
        print('finishing : ', name, completed)
        self.stop()

    def on_finished_utterance(self, name, completed):
        print('END')
        t = Thread(target=self.killme, args=(self.engine), daemon=True)
        t.start()

    def process_speech(self, text):
        self.engine.say(str(text))
        self.engine.startLoop(False)
        while self._running:
            self.engine.iterate()

    def say(self, text, length=2):
        # check if thread is running
        if self.t and self._running:
            # stop it if it is
            self.stop()
        # iterate speech in a thread

        self.t = Thread(target=self.process_speech, args=(text,), daemon=True)
        self._running = True
        self.t.start()

        elapsed_seconds = 0
        poll_interval = .1
        while not self.skip.is_set() and elapsed_seconds < length:
            self.skip.wait(poll_interval)
            elapsed_seconds += poll_interval

    def stop(self):
        self._running = False
        try:
            self.engine.endLoop()
        except:
            pass
        try:
            self.t.join()
        except Exception as e:
            pass

skip = Event()
myVoice = Voice(skip, 0)
myVoice.say("test", 2)
myVoice.say("test two", 2)

标签： pythonpython-3.xtext-to-speech

Answer 1

问题：多个say()被解雇/结束基于event 'finished-utterance'

此实现基于pyttsx3.readthedocs.io中的 running-a-driver-event-loop。

注意： 从TTSThread实例化开始并且永远运行！.
你必须打电话.terminate()来EXIT __main__防止永远等待！

# TTS.py
import threading, time, pyttsx3


class TTSThread(threading.Thread):
    def __init__(self, rate=115, event=None):
        super().__init__()

        if event:
            setattr(self, event, threading.Event())

        self._cancel = threading.Event()
        self.rate = rate
        self.engine = None

        self._say = threading.Event()
        self._text_lock = threading.Lock()
        self._text = []

        self._is_alive = threading.Event()
        self._is_alive.set()
        self.start()

    def _init_engine(self, rate):
        engine = pyttsx3.init()
        engine.setProperty('rate', rate)  # setting up new voice rate    
        engine.connect('finished-utterance', self._on_completed)
        engine.connect('started-word', self._on_cancel)
        return engine

    def say(self, text, stop=None):
        if self._is_alive.is_set():
            self._cancel.clear()

            if isinstance(text, str):
                text = [(text, stop)]

            if isinstance(text, (list, tuple)):
                for t in text:
                    if isinstance(t, str):
                        t = t, None

                    with self._text_lock:
                        self._text.append(t)

                    self._say.set()

    def cancel(self):
        self._cancel.set()

    def _on_cancel(self, name, location, length):
        if self._cancel.is_set():
            self.stop()

    def stop(self):        
        self.engine.stop()
        time.sleep(0.5)
        self.engine.endLoop()

    def _on_completed(self, name, completed):
        if completed:
            self.engine.endLoop()
            self.on_finished_utterance(name, completed)

    def on_finished_utterance(self, name, completed):
        pass

    def terminate(self):
        self._is_alive.clear()
        self._cancel.set()
        self.join()

    def run(self):
        self.engine = engine = self._init_engine(self.rate)
        while self._is_alive.is_set():
            while self._say.wait(0.1):
                self._say.clear()

                while not self._cancel.is_set() and len(self._text):
                    with self._text_lock:
                        engine.say(*self._text.pop(0))
                    engine.startLoop()

---

用法1 ：一次传递多个句子并自动运行enging.say(...)所有句子。

from TTS import TTSThread

SAY = ["Use your head to save your feet.", "Time will tell.", "Strike while the iron is hot."]

class Voice(TTSThread):
    def __init__(self):
        super().__init__(rate=115)

if __name__ == "__main__":
    voice = Voice()
    voice.say(SAY)

    # Simulate __main__.is_alive
    count = 0.0
    while True:
        time.sleep(0.1)
        count += 1

        if count >= 100:
            voice.terminate()
            break

print('EXIT __main__'.format())

用法 2：一个接一个地传递，依赖于事件'finished-utterance'。

from TTS import TTSThread
import time

class Voice(TTSThread):
    def __init__(self):
        self.completed = None
        super().__init__(rate=115, event='completed')

    def on_finished_utterance(self, name, completed):
        """
        Overloads `TTSThread.on_finished_utterance`
        which is connected to event 'finished-utterance'
        """
        if len(SAY):
            print('finishing[{}], delay next sentence {} sec.'.format(count, 1.5))
            time.sleep(1.5)
            self.completed.set()
        else:
            print('finishing')


if __name__ == "__main__":
    voice = Voice()

    # Start simulation, `say(...)` while __main__ is running
    voice.completed.set()

    _terminate = 100
    count = 0.0

    while True:
        time.sleep(0.1)
        count += 1

        if voice.completed.is_set():
            voice.completed.clear()

            if len(SAY):
                print('.say add[{}]: "{}..."'.format(count, SAY[0][:10]))
                voice.say(SAY.pop(0))

        if count % 20 == 0:
            print('__main__ {}'.format(count))

        if count >= _terminate:
            voice.terminate()

print('EXIT __main__'.format())

输出：

.say add[1.0]: "Use your h..."
finishing[18.0], delay next sentence 1.5 sec.
__main__ 20.0
.say add[34.0]: "Time will ..."
__main__ 40.0
finishing[51.0], delay next sentence 1.5 sec.
__main__ 60.0
.say add[67.0]: "Strike whi..."
__main__ 80.0
finishing
__main__ 100.0
EXIT __main__

用 Python 测试：3.5 - pyttsx3 版本：2.71