255 lines
6.9 KiB
Text
255 lines
6.9 KiB
Text
|
#!/usr/bin/env python3
|
||
|
|
# -*- coding: utf-8 -*-
|
||
|
|
|
||
|
|
"""
|
||
|
|
Read from a file or from stdin.
|
||
|
|
|
||
|
|
File is under EUPL1.2
|
||
|
|
Author: kujiu
|
||
|
|
"""
|
||
|
|
|
||
|
|
import argparse
|
||
|
|
|
||
|
|
import os.path
|
||
|
|
import secrets
|
||
|
|
import speechd
|
||
|
|
import chardet
|
||
|
|
import queue
|
||
|
|
import threading
|
||
|
|
import sys
|
||
|
|
|
||
|
|
|
||
|
|
PUNCTUATION = {
|
||
|
|
'none': speechd.PunctuationMode.NONE,
|
||
|
|
'some': speechd.PunctuationMode.SOME,
|
||
|
|
'all': speechd.PunctuationMode.ALL,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
class SpdRead:
|
||
|
|
def __init__(self, module, voice, lang, rate, punctuation, spelling):
|
||
|
|
|
||
|
|
"""
|
||
|
|
Spd read software
|
||
|
|
"""
|
||
|
|
self.module = module or 'default'
|
||
|
|
self.rate = rate
|
||
|
|
self.lang = lang
|
||
|
|
self.voice = voice
|
||
|
|
self.spelling = spelling
|
||
|
|
self.punctuation = punctuation
|
||
|
|
self._queue = queue.Queue()
|
||
|
|
self._semaphore = threading.Semaphore(1)
|
||
|
|
self._state = 'init'
|
||
|
|
self.spdc = speechd.client.Client()
|
||
|
|
|
||
|
|
def __del__(self):
|
||
|
|
if getattr(self, 'spdc', None):
|
||
|
|
self.spdc.close()
|
||
|
|
|
||
|
|
def run(self, source):
|
||
|
|
"""
|
||
|
|
Run application, blocking method
|
||
|
|
"""
|
||
|
|
self._state = "running"
|
||
|
|
try:
|
||
|
|
self.set_speechd_parameters()
|
||
|
|
if source:
|
||
|
|
self.queue_file(source)
|
||
|
|
else:
|
||
|
|
self.queue_stdin()
|
||
|
|
|
||
|
|
self.start_worker()
|
||
|
|
self.join()
|
||
|
|
except KeyboardInterrupt:
|
||
|
|
self.stop()
|
||
|
|
|
||
|
|
def stop(self):
|
||
|
|
self._state = "stopping"
|
||
|
|
self._semaphore.release()
|
||
|
|
self.spdc.close()
|
||
|
|
|
||
|
|
def start_worker(self):
|
||
|
|
"""
|
||
|
|
Start play thread.
|
||
|
|
"""
|
||
|
|
threading.Thread(target=self.say_loop, daemon=True).start()
|
||
|
|
|
||
|
|
def join(self):
|
||
|
|
"""
|
||
|
|
Join thread
|
||
|
|
"""
|
||
|
|
return self._queue.join()
|
||
|
|
|
||
|
|
def set_speechd_parameters(self):
|
||
|
|
"""
|
||
|
|
Configure speechd
|
||
|
|
"""
|
||
|
|
self.spdc.set_priority('important')
|
||
|
|
if self.module != 'default' and self.module:
|
||
|
|
self.spdc.set_output_module(self.module)
|
||
|
|
|
||
|
|
voices = [
|
||
|
|
voice[0]
|
||
|
|
for voice
|
||
|
|
in self.spdc.list_synthesis_voices()
|
||
|
|
if self._voice_match(voice)]
|
||
|
|
self.spdc.set_language(self.lang)
|
||
|
|
if len(voices) == 1:
|
||
|
|
self.spdc.set_synthesis_voice(voices[0])
|
||
|
|
elif len(voices) >= 1:
|
||
|
|
self.spdc.set_synthesis_voice(secrets.choice(voices))
|
||
|
|
else:
|
||
|
|
print("No voices installed for lang %s." % self.lang)
|
||
|
|
self.stop()
|
||
|
|
|
||
|
|
self.spdc.set_punctuation(PUNCTUATION[self.punctuation])
|
||
|
|
self.spdc.set_rate(self.rate)
|
||
|
|
self.spdc.set_spelling(self.spelling)
|
||
|
|
|
||
|
|
def queue_file(self, source):
|
||
|
|
"""
|
||
|
|
Queue speech from a file
|
||
|
|
"""
|
||
|
|
with open(source, "rb") as fin:
|
||
|
|
buffer = []
|
||
|
|
for line in fin.readlines():
|
||
|
|
encoding = chardet.detect(line)["encoding"]
|
||
|
|
line = line[:-1].strip()
|
||
|
|
line = line.decode(encoding, "replace").strip()
|
||
|
|
if line:
|
||
|
|
buffer.append(line)
|
||
|
|
elif len(buffer):
|
||
|
|
to_say = '\n'.join(buffer)
|
||
|
|
buffer = []
|
||
|
|
self.queue_speech(to_say)
|
||
|
|
|
||
|
|
if self._state != 'running':
|
||
|
|
break
|
||
|
|
|
||
|
|
if len(buffer) and self._state == 'running':
|
||
|
|
self.queue_speech('\n'.join(buffer))
|
||
|
|
|
||
|
|
def queue_stdin(self):
|
||
|
|
"""
|
||
|
|
Queue speech from stdin
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
buffer = []
|
||
|
|
for line in sys.stdin.buffer.readlines():
|
||
|
|
if not len(line):
|
||
|
|
break
|
||
|
|
|
||
|
|
encoding = chardet.detect(line)["encoding"]
|
||
|
|
line = line[:-1].strip()
|
||
|
|
line = line.decode(encoding, "replace").strip()
|
||
|
|
if line:
|
||
|
|
buffer.append(line)
|
||
|
|
elif len(buffer):
|
||
|
|
to_say = '\n'.join(buffer)
|
||
|
|
buffer = []
|
||
|
|
self.queue_speech(to_say)
|
||
|
|
|
||
|
|
if self._state != 'running':
|
||
|
|
break
|
||
|
|
|
||
|
|
if len(buffer) and self._state == 'running':
|
||
|
|
self.queue_speech('\n'.join(buffer))
|
||
|
|
|
||
|
|
except KeyboardInterrupt:
|
||
|
|
self.stop()
|
||
|
|
|
||
|
|
def next(self, *args, **kwargs):
|
||
|
|
"""
|
||
|
|
Wait for speechd
|
||
|
|
"""
|
||
|
|
pass
|
||
|
|
|
||
|
|
def _voice_match(self, voice):
|
||
|
|
"""
|
||
|
|
Check if a voice is in correct language
|
||
|
|
"""
|
||
|
|
if self.voice:
|
||
|
|
return voice[0].lower() == self.voice.lower()
|
||
|
|
if voice[1].lower() != self.lang.lower()[0:2]:
|
||
|
|
return False
|
||
|
|
if len(self.lang) > 4 and voice[2].lower() != self.lang[3:5]:
|
||
|
|
return False
|
||
|
|
return True
|
||
|
|
|
||
|
|
def queue_speech(self, text):
|
||
|
|
"""
|
||
|
|
Queue text to speech
|
||
|
|
"""
|
||
|
|
self._queue.put_nowait(text)
|
||
|
|
|
||
|
|
def spd_callback(self, callback_type):
|
||
|
|
"""
|
||
|
|
Process and of speechd
|
||
|
|
"""
|
||
|
|
if callback_type == speechd.CallbackType.CANCEL:
|
||
|
|
print("Cancelled speaking")
|
||
|
|
self.stop()
|
||
|
|
self._semaphore.release()
|
||
|
|
|
||
|
|
def say_loop(self):
|
||
|
|
"""
|
||
|
|
Say blocks
|
||
|
|
"""
|
||
|
|
while self._semaphore.acquire() and self._state == 'running':
|
||
|
|
try:
|
||
|
|
text = self._queue.get_nowait()
|
||
|
|
self.spdc.speak(
|
||
|
|
text, self.spd_callback,
|
||
|
|
event_types=(
|
||
|
|
speechd.CallbackType.CANCEL,
|
||
|
|
speechd.CallbackType.END))
|
||
|
|
self._queue.task_done()
|
||
|
|
except queue.Empty:
|
||
|
|
print("End of speech queue")
|
||
|
|
self.stop()
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
parser = argparse.ArgumentParser(description="Read with TTS.")
|
||
|
|
parser.add_argument(
|
||
|
|
'-l', '--lang', default='en',
|
||
|
|
help='Language')
|
||
|
|
parser.add_argument(
|
||
|
|
'-V', '--voice', default='',
|
||
|
|
help='Voice')
|
||
|
|
parser.add_argument(
|
||
|
|
'-m', '--module', default='default',
|
||
|
|
help='Speech Dispatcher output module')
|
||
|
|
parser.add_argument(
|
||
|
|
'-r', '--rate', default=15,
|
||
|
|
help='Speech Dispatcher rate', type=int)
|
||
|
|
parser.add_argument(
|
||
|
|
'-p', '--punctuation', default='none',
|
||
|
|
help='Punctuation mode (none, all or some)')
|
||
|
|
parser.add_argument(
|
||
|
|
'-S', '--spelling', default=False, action='store_true',
|
||
|
|
help='Spelling mode.')
|
||
|
|
parser.add_argument(
|
||
|
|
'source', default='', nargs='?',
|
||
|
|
help='Source file (if not stdin)')
|
||
|
|
|
||
|
|
options = parser.parse_args()
|
||
|
|
|
||
|
|
source = os.path.expanduser(
|
||
|
|
os.path.expandvars(options.source)
|
||
|
|
)
|
||
|
|
|
||
|
|
module = getattr(options, 'module', None)
|
||
|
|
rate = getattr(options, 'rate', 15)
|
||
|
|
if options.punctuation not in PUNCTUATION.keys():
|
||
|
|
print('Punctuation must be one of all, some or none')
|
||
|
|
exit(2)
|
||
|
|
|
||
|
|
reader = SpdRead(
|
||
|
|
module, options.voice,
|
||
|
|
options.lang, rate,
|
||
|
|
options.punctuation, options.spelling)
|
||
|
|
|
||
|
|
reader.run(options.source)
|