🤖💬在个人电脑上运行的基于Faster Whisper与GPT4All的简单英文聊天机器人 | A simple offline chatbot running on your computer using Faster Whisper and GPT4All.
import os
import sys
import time
import wave
import readline  # keep for arrow keys in input()
import tempfile
import argparse
import threading
from statistics import mean
from typing import Optional
from multiprocessing import Process

import pyaudio
import keyboard
from termcolor import colored
from textual import on, work
from textual.validation import Length
from textual.app import App, ComposeResult
from textual.widgets import Header, Footer
from textual.widgets import Input, RichLog, LoadingIndicator, Button, Sparkline

from service.base import ServerBase
from service.action import ActionServer, ActionClient
from service.language_model import LanguageModelServer, LanguageModelClient
from service.speech2text import Speech2TextServer, Speech2TextClient
from service.text2speech import EdgettsServer, Pyttsx3Server


def config():
    parser = argparse.ArgumentParser(description="Talky Talky Chatbot App")

    parser.add_argument(
        "--tui",
        action="store_true",
        dest="tui",
        help="Start the App in TUI mode",
    )

    parser.add_argument(
        "--user-input",
        default="voice",
        choices=["voice", "keyboard"],
        dest="user_input",
        help="Select the way to interact with the chatbot \
                (valid values: voice, keyboard)",
    )

    parser.add_argument(
        "--noecho-server-log",
        action="store_true",
        dest="noecho_server_log",
        help="Do not print the stdout of servers to terminal",
    )

    parser.add_argument(
        "--tts.engine",
        default="pyttsx3",
        choices=["pyttsx3", "edge-tts"],
        dest="tts_engine",
        help="Select the text-to-speech engine \
                (valid values: pyttsx3, edge-tts)",
    )

    parser.add_argument(
        "--stt.model-dir",
        default="model/speech_to_text/",
        dest="stt_model_dir",
        help="The directory to store the speech-to-text models",
    )

    parser.add_argument(
        "--lm.model-dir",
        default="model/language_model/",
        dest="lm_model_dir",
        help="The directory to store the language models",
    )

    args = parser.parse_args()

    print(colored("======== Application Configuration ========", 'green'))
    for argname, argvalue in vars(args).items():
        print(colored(f"{argname}:\t\t{argvalue}", 'green'))
    return args


def run_server(ServerClass: ServerBase,
               log_redirect_to: Optional[str],
               **kwargs):
    if log_redirect_to == "stdout":
        sys.stdout = sys.__stdout__
    elif log_redirect_to is None:
        sys.stdout = None
    else:
        sys.stdout = sys.__stdout__

    server = ServerClass(**kwargs)
    server.connect()
    server.run()


def start_server_process(ServerClass: ServerBase,
                         log_redirect_to,
                         **kwargs):
    proc_server = Process(
        target=run_server,
        args=[ServerClass, log_redirect_to],
        kwargs=kwargs
    )
    proc_server.daemon = True
    proc_server.start()
    return proc_server


def start_services(app_cfg):
    log_redirect_to = \
        None if app_cfg.noecho_server_log else "stdout"

    start_server_process(
        ActionServer, log_redirect_to,
    )

    start_server_process(
        LanguageModelServer, log_redirect_to,
        model_dir=app_cfg.lm_model_dir
    )

    start_server_process(
        Speech2TextServer, log_redirect_to,
        model_dir=app_cfg.stt_model_dir
    )

    if app_cfg.tts_engine == "pyttsx3":
        TtsServerClass = Pyttsx3Server
    elif app_cfg.tts_engine == "edge-tts":
        TtsServerClass = EdgettsServer
    else:
        TtsServerClass = EdgettsServer
    start_server_process(
        TtsServerClass, log_redirect_to,
    )


class TalkyTalkyCLI:
    def __init__(self, app_cfg):
        self.app_cfg = app_cfg
        start_services(self.app_cfg)

    def run(self):
        if self.app_cfg.user_input == "voice":
            self.voice_interact()
        elif self.app_cfg.user_input == "keyboard":
            self.keyboard_interact()

    def keyboard_interact(self):
        language_model_client = LanguageModelClient()
        action_client = ActionClient()

        language_model_client.connect()
        action_client.connect()

        print(colored("======== Type to Chat! ========", 'green'))
        while True:
            me = 'ME  >> '
            prompt = input(colored(me, 'red'))
            if len(prompt) > 0:
                # get answer from the prompt
                answer = language_model_client.get_answer(prompt)

                her = 'BOT >> '
                print(colored(her, 'green'), answer)
                print('')

                # make reaction to the answer
                action_client.react_to(answer)

    def voice_interact(self):
        speech2text_client = Speech2TextClient()
        language_model_client = LanguageModelClient()
        action_client = ActionClient()

        speech2text_client.connect()
        language_model_client.connect()
        action_client.connect()

        # Constants for audio settings
        FORMAT = pyaudio.paInt16
        CHANNELS = 2
        RATE = 44100
        CHUNK = 1024

        audio = pyaudio.PyAudio()
        stream = audio.open(format=FORMAT, channels=CHANNELS,
                            rate=RATE, input=True,
                            frames_per_buffer=CHUNK)

        print(colored("======== Holding SPACE KEY to Record ========", 'green'))
        frames = []
        while True:
            stream.start_stream()
            frames.clear()

            # start recording if key is pressed
            while keyboard.is_pressed('space'):
                data = stream.read(CHUNK)
                frames.append(data)
                # recording progress bar
                print(colored('>', 'blue'), end='', flush=True)

            stream.stop_stream()

            # Save the recorded audio to a WAV file
            if len(frames) > 0:
                wav_fp, wav_path = tempfile.mkstemp()
                wf = wave.open(wav_path, 'wb')
                wf.setnchannels(CHANNELS)
                wf.setsampwidth(audio.get_sample_size(FORMAT))
                wf.setframerate(RATE)
                wf.writeframes(b''.join(frames))
                wf.close()

                # speech to text
                print('')
                prompt = speech2text_client.get_text(wav_path)
                os.close(wav_fp)
                os.remove(wav_path)

                # print my prompt words
                me = 'ME  >> '
                print(colored(me, 'red'), prompt)
                if prompt == '[EMPTY SPEECH]':
                    answer = "Sorry, I can't hear you clearly. Please try again."
                else:
                    # get answer from the prompt
                    answer = language_model_client.get_answer(prompt)

                her = 'BOT >> '
                print(colored(her, 'green'), answer)
                print('')

                # make reaction to the answer
                action_client.react_to(answer)

            time.sleep(0.1)

        stream.close()
        audio.terminate()


class TalkyTalkyTUI(App):
    CSS = '''
Screen {
    layout: vertical;
    layers: below above;
    align: center middle;
}

RichLog {
    layer: below;
    height: 0.7fr;
    border: round darkgreen;
    background: green 10%;
    margin: 1;
}

RichLog:focus {
    border: round green;
    background: green 15%;
}

#prompt_box {
    layer: below;
    height: 0.2fr;
    border: round darkblue;
    background: blue 10%;
    margin: 1;
}

#prompt_box:focus {
    border: round blue;
    background: blue 15%;
}

Button {
    layer: below;
    height: 0.1fr;
    width: 1fr;
    margin: 2;
    min-height: 1;
}

#start_record {
    display: block;
}

#stop_record {
    display: none;
}

LoadingIndicator {
    layer: above;
    height: 10%;
    width: 30%;
    margin: 2;
    padding: 1;
    min-height: 3;
    display: none;
    content-align: center middle;
}

#processing_chat {
    color: $accent;
    border: $accent round;
}

#processing_record {
    color: $warning;
    border: $warning round;
}

Sparkline {
    layer: above;
    height: 10%;
    width: 30%;
    margin: 2;
    padding: 1;
    min-height: 3;
    display: none;
    border: $warning round;
    content-align: center middle;
}

Sparkline > .sparkline--max-color {
    color: $error;
}

Sparkline > .sparkline--min-color {
    color: $warning;
}
'''

    BINDINGS = [("d", "toggle_dark", "Toggle dark mode"),
                ("q", "quit", "Quit")]

    def __init__(self, app_cfg):
        self.app_cfg = app_cfg
        start_services(self.app_cfg)

        self.speech2text_client = Speech2TextClient()
        self.language_model_client = LanguageModelClient()
        self.action_client = ActionClient()

        self.speech2text_client.connect()
        self.language_model_client.connect()
        self.action_client.connect()

        self.mutex_on_submit = threading.Lock()

        self.is_recording = False
        self.chunk = 1024
        self.sample_format = pyaudio.paInt16
        self.channels = 2
        self.fs = 44100
        self.audio = pyaudio.PyAudio()
        self.stream = self.audio.open(format=self.sample_format,
                                      channels=self.channels,
                                      rate=self.fs,
                                      frames_per_buffer=self.chunk,
                                      input=True)
        super().__init__()

    def compose(self) -> ComposeResult:
        yield Header()
        yield Footer()
        yield RichLog(wrap=True, highlight=True, markup=True)
        yield Input(id="prompt_box",
                    placeholder="Write your prompt here. Press ENTER to send.",
                    validate_on=["changed", "submitted"],
                    validators=[Length(minimum=1, failure_description="Your prompt must not be empty.")]).focus()
        yield Button(label="SPEECH TO TEXT", id="start_record")
        yield Button(label="FINISH SPEAKING", id="stop_record", variant='warning')
        yield LoadingIndicator(id="processing_chat")
        yield LoadingIndicator(id="processing_record")
        yield Sparkline(data=[], summary_function=mean)

    def action_toggle_dark(self) -> None:
        """An action to toggle dark mode."""
        self.dark = not self.dark

    # TODO: After I add id="#prompt_box", input_box.action_submit() does not trigger submit_prompt(), why?
    # @on(Input.Submitted, "#prompt_box")
    @on(Input.Submitted)
    @work(exclusive=True, thread=True)
    def submit_prompt(self, event: Input.Submitted) -> None:
        if self.mutex_on_submit.acquire(blocking=False):
            if event.validation_result.is_valid:
                loading_box = self.query_one("#processing_chat")
                loading_box.display = True
                chat_box = self.query_one(RichLog)

                me = '[bold red]ME  >> [/bold red]'
                her = '[bold green]BOT >> [/bold green]'
                prompt = event.value
                chat_box.write(me + prompt)
                self.query_one("#prompt_box").value = ''

                answer = self.language_model_client.get_answer(prompt)
                chat_box.write(her + answer)

                loading_box.display = False
                self.action_client.react_to(answer)

            self.mutex_on_submit.release()

    @on(Button.Pressed, "#start_record")
    @work(exclusive=True, thread=True)
    async def start_record(self):
        if not self.is_recording:
            self.is_recording = True
            self.query_one("#start_record").display = False
            self.query_one("#stop_record").display = True

            frames = []
            sparkline = self.query_one(Sparkline)
            self.stream.start_stream()
            sparkline.display = True
            while self.is_recording:
                data = self.stream.read(self.chunk)
                frames.append(data)
                sparkline.data = data
            self.stream.stop_stream()
            sparkline.display = False

            input_box = self.query_one("#prompt_box")
            loading_box = self.query_one("#processing_record")
            loading_box.display = True
            if len(frames) > 0:
                wav_fp, wav_path = tempfile.mkstemp()
                wf = wave.open(wav_path, 'wb')
                wf.setnchannels(self.channels)
                wf.setsampwidth(self.audio.get_sample_size(self.sample_format))
                wf.setframerate(self.fs)
                wf.writeframes(b''.join(frames))
                wf.close()

                prompt = self.speech2text_client.get_text(wav_path)
                input_box.value = prompt
                os.close(wav_fp)
                os.remove(wav_path)

            loading_box.display = False
            self.query_one("#stop_record").display = False
            self.query_one("#start_record").display = True
            input_box.focus()

            await input_box.action_submit()

    @on(Button.Pressed, "#stop_record")
    def stop_record(self):
        if self.is_recording:
            self.is_recording = False


if __name__ == "__main__":
    app_cfg = config()
    if app_cfg.tui:
        app = TalkyTalkyTUI(app_cfg)
    else:
        app = TalkyTalkyCLI(app_cfg)
    app.run()

app.py
NewComer00/TalkyTalky
import os
import time
import enum
import socket
import threading
from service.base import ServerBase, ClientBase
from service.text2speech import Text2SpeechClient


class Action(enum.Enum):
    IDLE = 'idle'
    SPEAKING = 'speaking'


class ActionClient(ClientBase):
    def __init__(self,
                 server_ip='127.0.0.1', server_port=12346, recv_buflen=4096):

        super().__init__(server_ip=server_ip, server_port=server_port,
                         recv_buflen=recv_buflen)

    def react_to(self, sentence):
        self.socket.sendall(sentence.encode('utf-8'))
        self.socket.recv(self.recv_buflen).decode('utf-8')


class ActionServer(ServerBase):

    OPENSEEFACE_FRAME_LEN = 1785
    OPENSEEFACE_ACTION_DIR = r'action/openseeface/'

    def __init__(self,
                 action_fps: int = 24,
                 tts_server_ip: str = '127.0.0.1',
                 tts_server_port: int = 12347,
                 openseeface_client_ip: str = '127.0.0.1',
                 openseeface_client_port: int = 11573,
                 ip: str = '127.0.0.1',
                 port: str = 12346,
                 recv_buflen: int = 4096):

        super().__init__(ip=ip, port=port, recv_buflen=recv_buflen)

        self.tts_client = Text2SpeechClient(tts_server_ip, tts_server_port)

        # the role is in the idle state by default
        self.actor_state = Action.IDLE
        self.action_fps = action_fps
        self.action_daemon_thread = None

        self.openseeface_client_ip = openseeface_client_ip
        self.openseeface_client_port = openseeface_client_port
        # Openseeface uses UDP, so we init the socket here
        self.openseeface_client_socket = socket.socket(
            socket.AF_INET, socket.SOCK_DGRAM)
        self.openseeface_actions = self._init_actions()

    def __del__(self):
        super().__del__()

        if self.openseeface_client_socket:
            self.openseeface_client_socket.close()
        if self.action_daemon_thread:
            self.action_daemon_thread.join()

    def connect(self, listen_timeout=1):
        super().connect(listen_timeout)
        self.tts_client.connect()

    def run(self):
        self.action_daemon_thread = threading.Thread(
            target=self._action_daemon)
        self.action_daemon_thread.daemon = True
        self.action_daemon_thread.start()

        # Communication loop
        while True:
            bot_answer = self.client_socket.recv(
                self.recv_buflen).decode('utf-8')

            if bot_answer:
                self._server_log(
                    f"Received from {self.client_ip} << {bot_answer}")
                iter_reader = self.tts_client.read_aloud(bot_answer)
                next(iter_reader)
                self.actor_state = Action.SPEAKING
                next(iter_reader)
                self.actor_state = Action.IDLE

                response = "[ACTION DONE]"
                self.client_socket.sendall(response.strip().encode('utf-8'))
                self._server_log(f"Send >> {response}")

    def _init_actions(self):
        # List all action files in the directory
        action_files = [f for f in os.listdir(
            ActionServer.OPENSEEFACE_ACTION_DIR) if os.path.isfile(
            os.path.join(ActionServer.OPENSEEFACE_ACTION_DIR, f))]

        # Read actions into memory
        action_frames = {}
        for file in action_files:
            # Filename should be the valid value of Enum Action
            if any((file == a.value) for a in Action):
                action = Action(file)
            else:
                continue

            frames = []
            with open(os.path.join(
                    ActionServer.OPENSEEFACE_ACTION_DIR, file),
                    'rb') as fp:
                while True:
                    frame = fp.read(ActionServer.OPENSEEFACE_FRAME_LEN)
                    if not frame:
                        break  # End of file
                    frames.append(frame)
                action_frames[action] = frames
        return action_frames

    # Perform different actions based on self.actor_state
    def _action_daemon(self):
        delay = 1 / self.action_fps
        while True:
            for action in Action:
                if self.actor_state is action:
                    for frame in self.openseeface_actions[action]:
                        if self.actor_state is not action:
                            break
                        self.openseeface_client_socket.sendto(
                            frame, (self.openseeface_client_ip,
                                    self.openseeface_client_port))
                        time.sleep(delay)
                    break


if __name__ == '__main__':
    acts = ActionServer()
    acts.connect()
    acts.run()

service/action.py
NewComer00/TalkyTalky
# TODO: Rewrite me using socketserver
# https://docs.python.org/3/library/socketserver.html
# https://stackoverflow.com/questions/12233940/passing-extra-metadata-to-a-requesthandler-using-pythons-socketserver-and-child

import time
import socket
import datetime
from select import select


def service_log(obj, msg):
    class_name = obj.__class__.__name__
    obj_id = id(obj)
    time_stamp = datetime.datetime.now(datetime.timezone.utc)
    print(f"[{time_stamp}][{class_name}@{obj_id}] {msg}")


class ClientBase:
    def __init__(self, server_ip, server_port, recv_buflen):
        self._client_log("Initializing the client...")
        self.server_ip = server_ip
        self.server_port = server_port
        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.recv_buflen = recv_buflen

    def __del__(self):
        if self.socket:
            self.socket.close()

    def connect(self, max_retry=None, wait_time=1):
        count = 0
        while max_retry is None or count < max_retry:
            try:
                self.socket.connect((self.server_ip, self.server_port))
            except Exception as e:
                if max_retry is None:
                    self._client_log(f'{e} (Retry={count+1}/∞)')
                else:
                    self._client_log(f'{e} (Retry={count+1}/{max_retry})')
                time.sleep(wait_time)
            else:
                self._client_log("Connected to server")
                return
            count += 1
        self._client_log("Failed to connect to server")

    def _client_log(self, msg):
        service_log(self, msg)


class ServerBase:
    def __init__(self, ip, port, recv_buflen):
        self._server_log("Initializing the server...")
        self.port = port
        self.server_ip = ip
        self.server_socket = None
        self.client_ip = None
        self.client_socket = None
        self.recv_buflen = recv_buflen

    def __del__(self):
        if self.server_socket:
            self.server_socket.close()
        if self.client_socket:
            self.client_socket.close()

    def connect(self, listen_timeout=1):
        self.server_socket = socket.socket(
            socket.AF_INET, socket.SOCK_STREAM)
        self.server_socket.setsockopt(
            socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

        self.server_socket.bind((self.server_ip, self.port))

        self._server_log(f"Listening on {self.server_ip}:{self.port}")
        while True:
            self.server_socket.listen()
            # Accept incoming connections
            ready, _, _ = select([self.server_socket], [], [], listen_timeout)
            if ready:
                self.client_socket, self.client_ip = \
                    self.server_socket.accept()
                break
        self._server_log(f"Connected to {self.client_ip}")

    def run(self):
        pass

    def _server_log(self, msg):
        service_log(self, msg)

service/base.py
NewComer00/TalkyTalky
import os
from gpt4all import GPT4All
from service.base import ServerBase, ClientBase


class LanguageModelClient(ClientBase):
    def __init__(self,
                 server_ip='127.0.0.1', server_port=12345, recv_buflen=4096):

        super().__init__(server_ip=server_ip, server_port=server_port,
                         recv_buflen=recv_buflen)

    def get_answer(self, prompt):
        self.socket.sendall(prompt.encode('utf-8'))
        answer = self.socket.recv(self.recv_buflen).decode('utf-8')
        return answer


class LanguageModelServer(ServerBase):
    def __init__(self, model_dir,
                 model_name="orca-mini-3b.ggmlv3.q4_0.bin", infer_device='cpu',
                 ip='127.0.0.1', port=12345, recv_buflen=4096):

        super().__init__(ip=ip, port=port, recv_buflen=recv_buflen)

        self._server_log(
            f"Loading Language model {model_name} using {infer_device}")

        os.makedirs(model_dir, exist_ok=True)
        try:
            # try to check if the model has been downloaded
            self.model = GPT4All(model_name, model_path=model_dir,
                                 device=infer_device, allow_download=False)
        except ValueError:
            self.model = GPT4All(model_name, model_path=model_dir,
                                 device=infer_device, allow_download=True)

    def run(self):
        with self.model.chat_session():
            # Communication loop
            while True:
                recv_data = self.client_socket.recv(
                    self.recv_buflen).decode('utf-8').strip()
                self._server_log(
                    f"Received from {self.client_ip} << {recv_data}")

                response = self.model.generate(prompt=recv_data, temp=0)
                self.client_socket.sendall(response.strip().encode('utf-8'))
                self._server_log(f"Send >> {response}")


if __name__ == '__main__':
    lms = LanguageModelServer()
    lms.connect()
    lms.run()

service/language_model.py
NewComer00/TalkyTalky
from faster_whisper import WhisperModel
from service.base import ServerBase, ClientBase


class Speech2TextClient(ClientBase):
    def __init__(self,
                 server_ip='127.0.0.1', server_port=12344, recv_buflen=4096):

        super().__init__(server_ip=server_ip, server_port=server_port,
                         recv_buflen=recv_buflen)

    def get_text(self, wav_file_path):
        self.socket.sendall(wav_file_path.encode('utf-8'))
        answer = self.socket.recv(self.recv_buflen).decode('utf-8')
        return answer


class Speech2TextServer(ServerBase):
    def __init__(self,
                 model_dir, model_name="base.en",
                 infer_device='cpu', compute_type='int8',
                 ip='127.0.0.1', port=12344, recv_buflen=4096):

        super().__init__(ip=ip, port=port, recv_buflen=recv_buflen)

        self._server_log(
            f"Loading speech-to-text model {model_name} using {infer_device}")
        try:
            # we first try to use the local model
            self.model = WhisperModel(model_name, download_root=model_dir,
                                      device=infer_device,
                                      compute_type=compute_type,
                                      local_files_only=True)
        except FileNotFoundError:
            self.model = WhisperModel(model_name, download_root=model_dir,
                                      device=infer_device,
                                      compute_type=compute_type,
                                      local_files_only=False)


    def run(self):
        # Communication loop
        while True:
            recv_data = self.client_socket.recv(
                self.recv_buflen).decode('utf-8')
            self._server_log(
                f"Received from {self.client_ip} << {recv_data}")

            wav_file_path = recv_data
            speech_text = self._speech_to_text(wav_file_path)
            response = speech_text

            self.client_socket.send(response.encode('utf-8'))
            self._server_log(f"Send >> {response}")

    def _speech_to_text(self, wav_file_path):
        segments, _ = self.model.transcribe(wav_file_path)
        segments = list(segments)
        speech_text = ''.join([seg.text for seg in segments]).strip()

        # deal with empty speech
        if len(speech_text) == 0:
            speech_text = "[EMPTY SPEECH]"

        return speech_text


if __name__ == '__main__':
    sts = Speech2TextServer()
    sts.connect()
    sts.run()

service/speech2text.py
NewComer00/TalkyTalky
import os
import asyncio
import platform
import tempfile
from pathlib import Path
import pyttsx3
import edge_tts
from playsound import playsound
from service.base import ServerBase, ClientBase


class Text2SpeechClient(ClientBase):
    def __init__(self,
                 server_ip='127.0.0.1', server_port=12347, recv_buflen=4096):

        super().__init__(server_ip=server_ip, server_port=server_port,
                         recv_buflen=recv_buflen)

    def read_aloud(self, text):
        self.socket.sendall(text.encode('utf-8'))
        start_reading = self.socket.recv(self.recv_buflen).decode('utf-8')
        yield start_reading
        finish_reading = self.socket.recv(self.recv_buflen).decode('utf-8')
        yield finish_reading


class Text2SpeechServerBase(ServerBase):
    def __init__(self,
                 voice="",
                 ip='127.0.0.1', port=12347, recv_buflen=4096):

        super().__init__(ip=ip, port=port, recv_buflen=recv_buflen)
        self.voice = voice

    def run(self):
        # Communication loop
        while True:
            recv_data = self.client_socket.recv(
                self.recv_buflen).decode('utf-8')
            self._server_log(
                f"Received from {self.client_ip} << {recv_data}")

            self._text_to_speech(recv_data)

    def _text_to_speech(self, text):
        response = '[START READING]'
        self.client_socket.send(response.encode('utf-8'))
        self._server_log(f"Send >> {response}")
        pass
        response = '[FINISH READING]'
        self.client_socket.send(response.encode('utf-8'))
        self._server_log(f"Send >> {response}")


class Pyttsx3Server(Text2SpeechServerBase):
    def _text_to_speech(self, text):
        response = '[START READING]'
        self.client_socket.send(response.encode('utf-8'))
        self._server_log(f"Send >> {response}")

        pyttsx3.speak(text)

        response = '[FINISH READING]'
        self.client_socket.send(response.encode('utf-8'))
        self._server_log(f"Send >> {response}")


class EdgettsServer(Text2SpeechServerBase):
    def __init__(self,
                 voice="en-US-AriaNeural",
                 ip='127.0.0.1', port=12347, recv_buflen=4096):

        super().__init__(voice=voice,
                         ip=ip, port=port, recv_buflen=recv_buflen)

    def _text_to_speech(self, text):
        async def amain(text, voice, output_file):
            communicate = edge_tts.Communicate(text, voice)
            await communicate.save(output_file)

        mp3_fp, mp3_path = tempfile.mkstemp(suffix='.mp3')

        # https://stackoverflow.com/a/70758881/15283141
        if platform.system() == 'Windows':
            asyncio.set_event_loop_policy(
                asyncio.WindowsSelectorEventLoopPolicy())
        else:
            # https://stackoverflow.com/a/45600858
            asyncio.set_event_loop(asyncio.new_event_loop())

        loop = asyncio.get_event_loop_policy().get_event_loop()
        try:
            loop.run_until_complete(amain(text, self.voice, mp3_path))
        finally:
            loop.close()

        response = '[START READING]'
        self.client_socket.send(response.encode('utf-8'))
        self._server_log(f"Send >> {response}")

        # To solve playsound issue: a problem occurred in initializing MCI
        os.close(mp3_fp)
        if platform.system() == 'Windows':
            playsound(Path(mp3_path).as_uri())
        else:
            playsound(mp3_path)
        os.remove(mp3_path)

        response = '[STOP READING]'
        self.client_socket.send(response.encode('utf-8'))
        self._server_log(f"Send >> {response}")


if __name__ == '__main__':
    ttss = Pyttsx3Server()
    ttss.connect()
    ttss.run()

service/text2speech.py
NewComer00/TalkyTalky
import socket
import time

UDP_IP = "127.0.0.1"  # Replace with the IP address of the UDP server
UDP_PORT = 11573  # Replace with the port number the UDP server is listening on

# Open the binary file for reading
file_path = "received_data.bin"  # Replace with the path to your binary file
frames = []

with open(file_path, "rb") as file:
    while True:
        frame = file.read(1785)  # Read one frame (1785 bytes) from the binary file
        if not frame:
            break  # End of file

        frames.append(frame)

# Create a UDP socket and send each frame in the list
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

while True:
    for frame in frames:
        sock.sendto(frame, (UDP_IP, UDP_PORT))

        # Optionally, you can print a message to confirm frame transmission
        print(f"Sent {len(frame)} bytes")
        time.sleep(1/25)

sock.close()

print("Transmission complete.")


tool/openseeface_play_packet.py
NewComer00/TalkyTalky
import socket

UDP_IP = "127.0.0.1"
UDP_PORT = 11573

sock = socket.socket(socket.AF_INET, # Internet
                     socket.SOCK_DGRAM) # UDP
sock.bind((UDP_IP, UDP_PORT))

DATA_FRAME_LEN = 1785

file_path = "received_data.bin"

with open(file_path, "wb") as file:
    while True:
        try:
            # Receive data from the UDP socket
            data, addr = sock.recvfrom(2048)  # 1024 is the buffer size, you can adjust it as needed

            # Write the received binary data to the file
            if len(data) == DATA_FRAME_LEN:
                file.write(data)

            # Optionally, you can print a message to confirm data reception
            print(f"Received {len(data)} bytes from {addr}")

        except KeyboardInterrupt:
            # Close the socket and exit if the user interrupts with Ctrl+C
            sock.close()
            break

tool/openseeface_record_packet.py
NewComer00/TalkyTalky