homes/common/.local/bin/spd-read

#!/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)