Fix bug causing TTS to redownload models every time

* Playing a clip now returns a stream (what used to be called streams is 
now called Output) that can be used to track playback state. 
* Multiple streams can play at the same time on one Output (no more 
audio driver limitations)
* Clips can now be mixed and concatinated using + and / respectively 
(you can now compose the audio how you want it rather than relying on 
async scheduling)

pyproject.toml

@@ -8,21 +8,20 @@ authors = [
 dependencies = [
     "click",
     "kdl-py",
-    "quart==0.18.*",
-    "werkzeug==2.3.7",
+    "quart",
+    "werkzeug",
     "hypercorn",
-    "requests",
-    "websockets==11.0.3",
-    "aioprocessing",
+    "websockets",
     "aioscheduler",
-    "pyaudio==0.2.*",
-    "librosa==0.8.*",
+    "pyaudio",
+    "librosa",
     "pytsmod",
     "numpy",
     "multipledispatch",
     "blessed",
     "appdirs",
     "maya",
+    "httpx",
 ]
 requires-python = ">=3.10,<3.11"
 readme = "README.md"
@@ -30,18 +29,17 @@ license = {text = "GPLv2"}
 
 [project.optional-dependencies]
 tts = [
-  "TTS==0.9.*",
-  "torch==1.13.*",
+  "coqui-tts",
 ]
 phrasecounter = ["num2words"]
-jail = ["owoify-py==2.*"]
+jail = ["owoify-py"]
 twitch = ["miniirc"]
 midi = [
   "mido",
   "python-rtmidi",
 ]
 obs = ["simpleobsws"]
-osc = ["python-osc>=1.9.0"]
+osc = ["python-osc"]
 yt-dlp = ["yt-dlp"]
 
 [build-system]

src/ovtk_audiencekit/chats/ChatProcess.py

@@ -1,5 +1,5 @@
 from abc import ABC, abstractmethod
-from multiprocessing import Process, Pipe, Manager
+import asyncio
 import sys
 import os
 import json
@@ -16,22 +16,30 @@ class GracefulShutdownException(Exception):
 class ShutdownRequest(Event):
     _hidden = True
 
+class ManagerMock(dict):
+    __getattr__ = dict.get
+    __setattr__ = dict.__setitem__
+    __delattr__ = dict.__delitem__
 
-class ChatProcess(Process, ABC):
-    def __init__(self, stdin_lock, name, readonly=False):
+
+class ChatProcess(ABC):
+    def __init__(self, name, event_queue, readonly=False):
         super().__init__()
-        self._stdin_lock = stdin_lock
         self._name = name
+        self._event_queue = event_queue
+        self._control_queue = asyncio.Queue()
         self.readonly = readonly
 
-        self.event_pipe, self._event_pipe = Pipe(duplex=False)
-        self._control_pipe, self.control_pipe = Pipe(duplex=False)
         self.logger = logging.getLogger(f'chat.{self._name}')
 
         self._state = None
         self._next_state = None
 
-        self.shared = Manager().Namespace()
+        self.shared = ManagerMock({})
+
+    @abstractmethod
+    def setup(self, *args, **kwargs):
+        pass
 
     @abstractmethod
     def loop(self, next_state):
@@ -41,7 +49,7 @@ class ChatProcess(Process, ABC):
         pass
 
     def send(self, event):
-        self.control_pipe.send(event)
+        self._control_queue.put_nowait(event)
 
     @property
     def state(self):
@@ -82,24 +90,20 @@ class ChatProcess(Process, ABC):
         print(f'WARNING: Unprocessed control message in {self._name} - {event}')
 
     def safe_input(self, prompt):
-        self._stdin_lock.aquire()
-        with os.fdopen(0) as real_stdin:
-            old_stdin = sys.stdin
-            sys.stdin = real_stdin
-            response = input(prompt)
-            sys.stdin = old_stdin
-        self._stdin_lock.release()
+        response = input(prompt)
         return response
 
     def publish(self, event):
-        self._event_pipe.send(event)
+        self._event_queue.put_nowait(event)
 
-    def run(self):
+    async def run(self):
         while True:
             try:
                 timeout = 0
                 # Run the code loop
                 response = self.loop(self._next_state)
+                if asyncio.iscoroutine(response):
+                    response = await response
                 # Check response type and use the appropriate interface
                 # # integer / float: timeout value to use for the next loop
                 # # two states (tuple): current state and next state
@@ -113,22 +117,26 @@ class ChatProcess(Process, ABC):
                     self.state = response
 
                 # Wait for inconming events for at most timeout seconds - handle if any come in
-                if self._control_pipe.poll(timeout):
-                    incoming_event = self._control_pipe.recv()
-                    # Handle messages implimented by this base class
-                    if isinstance(incoming_event, ShutdownRequest):
-                        raise GracefulShutdownException()
-                    # Pass the rest on
-                    elif not self.readonly:
-                        response = self.on_control_event(incoming_event, self._next_state)
-                        # Similar to above, handle response states
-                        if response is None:
-                            continue
-                        elif type(response) is tuple:
-                            self.state, self._next_state = response
-                        else:
-                            self.state = response
+                try:
+                    incoming_event = await asyncio.wait_for(self._control_queue.get(), timeout)
+                except asyncio.exceptions.TimeoutError:
+                    continue
 
+                # Handle messages implimented by this base class
+                if isinstance(incoming_event, ShutdownRequest):
+                    raise GracefulShutdownException()
+                # Pass the rest on
+                elif not self.readonly:
+                    response = self.on_control_event(incoming_event, self._next_state)
+                    if asyncio.iscoroutine(response):
+                        response = await response
+                    # Similar to above, handle response states
+                    if response is None:
+                        continue
+                    elif type(response) is tuple:
+                        self.state, self._next_state = response
+                    else:
+                        self.state = response
             except (GracefulShutdownException, KeyboardInterrupt):
                 return 0
             except Exception as e:

src/ovtk_audiencekit/chats/FakeChat/FakeChat.py

@@ -1,4 +1,5 @@
 import random
+import asyncio
 from enum import Enum, auto
 
 from ovtk_audiencekit.chats import ChatProcess
@@ -16,8 +17,7 @@ class StartStop(Event):
 
 
 class FakeChat(ChatProcess):
-    def __init__(self, *args, max_delay=10, max_messages_per_chunk=1, start_paused=True, max_monitization=None, **kwargs):
-        super().__init__(*args, **kwargs)
+    def setup(self, max_delay=10, max_messages_per_chunk=1, start_paused=True, max_monitization=None):
         self._max_delay = max_delay
         self._max_messages_per_chunk = max_messages_per_chunk
         self._max_monitization = max_monitization

src/ovtk_audiencekit/chats/Misskey/misskey.py

@@ -3,9 +3,11 @@ import random
 import logging
 from enum import Enum, auto
 from itertools import chain
+import asyncio
+
+import websockets
 
 from ovtk_audiencekit.chats import ChatProcess
-from ovtk_audiencekit.utils import NonBlockingWebsocket
 from ovtk_audiencekit.events.Message import Message, USER_TYPE
 
 logger = logging.getLogger(__name__)
@@ -17,8 +19,7 @@ class STATES(Enum):
 
 
 class MisskeyProcess(ChatProcess):
-    def __init__(self, *args, instance=None, channel=None, token=None, **kwargs):
-        super().__init__(*args, **kwargs)
+    def setup(self, instance=None, channel=None, token=None):
         self._url = f'wss://{instance}/streaming'
         if token:
             self._url += f'?token={token}'
@@ -42,9 +43,8 @@ class MisskeyProcess(ChatProcess):
             return msg
         return None
 
-    def on_connecting(self, next_state):
-        self._ws = NonBlockingWebsocket(self._url)
-        self._ws.start()
+    async def on_connecting(self, next_state):
+        self._ws = await websockets.connect(self._url)
         payload = {
             'type': 'connect',
             'body': {
@@ -55,25 +55,28 @@ class MisskeyProcess(ChatProcess):
                 }
             }
         }
-        self._ws.send(json.dumps(payload))
+        logger.debug("Subscribed to channel bus")
+        await self._ws.send(json.dumps(payload))
         return STATES.READING
 
-    def on_reading(self, next_state, timeout=0.1):
-        if self._ws.poll(timeout):
-            note_event = self._ws.recv()
-            try:
-                misskey_event = json.loads(note_event)
-                if misskey_event['body']['id'] == self._subid and misskey_event['body']['type'] == 'note':
-                    note = misskey_event['body']['body'] # LOL
-                    norm = None
-                    try:
-                        norm = self.normalize_event(note)
-                    except Exception as e:
-                        logger.error(f'Failed to process note data: {note}')
-                    if norm:
-                        self.publish(norm)
-            except (KeyError, json.JSONDecodeError):
-                logger.error(f'Unknown data in websocket: {note_event}')
+    async def on_reading(self, next_state, timeout=0.1):
+        try:
+            note_event = await asyncio.wait_for(self._ws.recv(), timeout)
+        except asyncio.exceptions.TimeoutError:
+            return 0
+        try:
+            misskey_event = json.loads(note_event)
+            if misskey_event['body']['id'] == self._subid and misskey_event['body']['type'] == 'note':
+                note = misskey_event['body']['body'] # LOL
+                norm = None
+                try:
+                    norm = self.normalize_event(note)
+                except Exception as e:
+                    logger.error(f'Failed to process note data: {note}')
+                if norm:
+                    self.publish(norm)
+        except (KeyError, json.JSONDecodeError):
+            logger.error(f'Unknown data in websocket: {note_event}')
         return 0
 
     def loop(self, next_state):

src/ovtk_audiencekit/chats/PeerTube/__init__.py

@@ -1 +0,0 @@
-from .peertube import PtChatProcess as Process

src/ovtk_audiencekit/chats/PeerTube/peertube.py

@@ -1,70 +0,0 @@
-import json
-import random
-import logging
-from enum import Enum, auto
-from itertools import chain
-
-import socketio
-import requests
-
-from ovtk_audiencekit.chats import ChatProcess
-from ovtk_audiencekit.events.Message import Message, USER_TYPE
-
-logger = logging.getLogger(__name__)
-
-
-class STATES(Enum):
-    CONNECTING = auto()
-    READING = auto()
-
-
-class PtChatProcess(ChatProcess):
-    def __init__(self, *args, instance_url=None, channel=None, token=None, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.instance_url = instance_url
-        self.channel = channel
-
-        self._state_machine = self.bind_to_states(STATES)
-        self.state = STATES.CONNECTING
-
-    def normalize_event(self, event):
-        message = event['message']
-        user_name = message['account']['displayName'] or message['account']['name']
-        user_id = message['account']['id']
-        text = message.get('text', '')
-        if text:
-            msg = Message(self._name, text,
-                          user_name, user_id, USER_TYPE.USER,
-                          id=message['id'])
-            return msg
-        return None
-
-    def on_connecting(self, next_state):
-        self._session = requests.Session()
-        self._sio = socketio.Client()
-
-        api_channel = self._session.get(f'{self.instance_url}/api/v1/video-channels/{self.channel}')
-        api_channel.raise_for_status()
-        room_id = api_channel.json().get('roomId')
-        if room_id is None:
-            raise ValueError('No chatroom returned from channel api!')
-
-        self._sio.connect(self.instance_url, namespaces=['/live-chat'], wait=True)
-        self._sio.emit('subscribe', data={'roomId': room_id}, namespace='/live-chat')
-        self._sio.on('new-message', self.handle_message, namespace='/live-chat')
-
-        return STATES.READING
-
-    def handle_message(self, data):
-        try:
-            norm = self.normalize_event(data)
-            self.publish(norm)
-        except Exception:
-            logger.error(f'Failed to process message: {data}')
-
-    def on_reading(self, next_state, timeout=0.5):
-        self._sio.sleep(timeout)
-        return 0
-
-    def loop(self, next_state):
-        return self._state_machine(self.state, next_state)

src/ovtk_audiencekit/chats/Twitch/TwitchProcess.py

@@ -1,4 +1,4 @@
-import time
+import asyncio
 from enum import Enum, auto
 from itertools import chain
 
@@ -20,7 +20,7 @@ valid_emote_resolutions = [1.0, 2.0, 3.0, 4.0]
 
 
 class TwitchProcess(ChatProcess):
-    def __init__(self, *args,
+    async def setup(self,
                  # Shared
                  channel_name=None, client_id=None, access_token=None,
                  # IRC options
@@ -28,11 +28,7 @@ class TwitchProcess(ChatProcess):
                  # EventSub options
                  eventsub=True, eventsub_host='wss://ovtk.skeh.site/twitch',
                  # BTTV integration
-                 bttv=False,
-                 # Inheritance boilerplate
-                 **kwargs):
-        super().__init__(*args, **kwargs)
-
+                 bttv=False):
         if channel_name is None or client_id is None or access_token is None:
             raise ValueError('Twitch chat is missing config requirements')
         if emote_res not in valid_emote_resolutions:
@@ -50,8 +46,8 @@ class TwitchProcess(ChatProcess):
 
         self.api = TwitchAPI(self._client_id, self._token)
         self.shared.api = self.api
-        cheermotes = self.api.get_cheermotes(self._channel_name)
-        target_data = self.api.get_user_details(self._channel_name)
+        cheermotes = await self.api.get_cheermotes(self._channel_name)
+        target_data = await self.api.get_user_details(self._channel_name)
 
         self.shared.target_data = target_data
         self.shared.users = []
@@ -64,12 +60,12 @@ class TwitchProcess(ChatProcess):
             self._sources.append(self.eventsub)
 
         self.bttv = BTTV(target_data['user']['id']) if bttv else None
-
+        await self.bttv.setup()
 
     def loop(self, next_state):
         return self._state_machine(self.state, next_state)
 
-    def on_control_event(self, event, next_state):
+    async def on_control_event(self, event, next_state):
         if isinstance(event, Message):
             # Twitch.... why... why no newlines....
             for line in event.text.split('\n'):
@@ -77,7 +73,7 @@ class TwitchProcess(ChatProcess):
                     continue
                 if event.replies_to:
                     line = f"@{event.replies_to.user_name} {line}"
-                self.irc.send(line)
+                await self.irc.send(line)
 
     def on_state_enter(self, new_state):
         status_messages = {
@@ -89,15 +85,16 @@ class TwitchProcess(ChatProcess):
             sys_msg = SysMessage(self._name, message)
             self.publish(sys_msg)
 
-    def on_connecting(self, next_state):
-        self.irc.connect()
+    async def on_connecting(self, next_state):
+        await self.irc.connect()
         if self.__dict__.get('eventsub'):
-            self.eventsub.subscribe(self._channel_name)
+            await self.eventsub.subscribe(self._channel_name)
         return STATES.READING
 
-    def on_reading(self, next_state):
+    async def on_reading(self, next_state):
         try:
-            for event in chain(*(source.read(0.1) for source in self._sources)):
+            events = [event for source in self._sources async for event in source.read(0.1)]
+            for event in events:
                 # Retarget event
                 event.via = self._name
                 if self.bttv and isinstance(event, Message):
@@ -107,8 +104,8 @@ class TwitchProcess(ChatProcess):
         except TimeoutError:
             return STATES.TIMEOUT
 
-    def on_timeout(self, next_state):
-        time.sleep(3)
+    async def on_timeout(self, next_state):
+        await asyncio.sleep(3)
         return STATES.CONNECTING
 
     def on_failure(self, next_state):

src/ovtk_audiencekit/chats/Twitch/sources/BTTV.py

@@ -1,7 +1,7 @@
 from itertools import chain
 import re
 
-import requests
+import httpx
 
 API_URL = 'https://api.betterttv.net/3/cached'
 CDN_URL = 'https://cdn.betterttv.net/emote'
@@ -11,9 +11,10 @@ word_regex = re.compile('\b([A-z0-9]+)\b')
 class BTTV:
     def __init__(self, user_id):
         self.user_id = user_id
-        self._session = requests.Session()
+        self._session = httpx.AsyncClient()
 
-        self.emotes = {emote['code']: emote for emote in chain(self._get_global(), self._get_channel())}
+    async def setup(self):
+        self.emotes = {emote['code']: emote for emote in chain(await self._get_global(), await self._get_channel())}
 
     def hydrate(self, event):
         text, emotes = self.parse_emotes(event.text)
@@ -37,13 +38,13 @@ class BTTV:
 
         return output, used_emotes
 
-    def _get_global(self):
-        response = self._session.get(f'{API_URL}/emotes/global')
+    async def _get_global(self):
+        response = await self._session.get(f'{API_URL}/emotes/global')
         response.raise_for_status()
         return response.json()
 
-    def _get_channel(self):
-        response = self._session.get(f'{API_URL}/users/twitch/{self.user_id}')
+    async def _get_channel(self):
+        response = await self._session.get(f'{API_URL}/users/twitch/{self.user_id}')
         response.raise_for_status()
         data = response.json()
         return chain(data['channelEmotes'], data['sharedEmotes'])

src/ovtk_audiencekit/chats/Twitch/sources/TwitchAPI.py

@@ -1,4 +1,4 @@
-import requests
+import httpx
 
 API_URL = 'https://api.twitch.tv/helix'
 
@@ -6,33 +6,33 @@ class TwitchAPI:
     def __init__(self, client_id, token):
         self._client_id = client_id
         self._token = token
-        self._session = requests.Session()
+        self._session = httpx.AsyncClient()
         self._session.headers.update({'Authorization': f'Bearer {self._token}', 'Client-ID': client_id, 'Accept': 'application/vnd.twitchtv.v5+json'})
 
-    def get_cheermotes(self, channel):
-        response = self._session.get(f'{API_URL}/bits/cheermotes', params={'channel_id': channel})
+    async def get_cheermotes(self, channel):
+        response = await self._session.get(f'{API_URL}/bits/cheermotes', params={'channel_id': channel})
         response.raise_for_status()
         return response.json()['data']
 
-    def get_user_details(self, username):
+    async def get_user_details(self, username):
         data = dict(link=f'https://twitch.tv/{username}')
         try:
-            user_response = self._session.get(f'{API_URL}/users', params={'login': username})
+            user_response = await self._session.get(f'{API_URL}/users', params={'login': username})
             user_response.raise_for_status()
             data['user'] = user_response.json()['data'][0]
 
-            channel_response = self._session.get(f'{API_URL}/channels', params={'broadcaster_id': data['user']['id']})
+            channel_response = await self._session.get(f'{API_URL}/channels', params={'broadcaster_id': data['user']['id']})
             channel_response.raise_for_status()
             data['channel'] = channel_response.json()['data'][0]
         except (KeyError, IndexError):
             return None
         return data
 
-    def get_clips(self, username):
-        user_response = self._session.get(f'{API_URL}/users', params={'login': username})
+    async def get_clips(self, username):
+        user_response = await self._session.get(f'{API_URL}/users', params={'login': username})
         user_response.raise_for_status()
         channel_id = user_response.json()['data'][0]['id']
 
-        clips_response = self._session.get(f'{API_URL}/clips', params={'broadcaster_id': broadcaster_id})
+        clips_response = await self._session.get(f'{API_URL}/clips', params={'broadcaster_id': broadcaster_id})
         clips_response.raise_for_status()
         return clips_response.json()['data']

src/ovtk_audiencekit/chats/Twitch/sources/TwitchEventSub.py

@@ -1,7 +1,8 @@
 import json
 import logging
+import asyncio
 
-from ovtk_audiencekit.utils import NonBlockingWebsocket
+import websockets
 from ovtk_audiencekit.core.Data import ovtk_user_id
 from ovtk_audiencekit.events import Follow
 
@@ -15,14 +16,13 @@ class TwitchEventSub:
         self.api = api
         self.mirror_socket = f'{eventsub_host}/ws/{ovtk_user_id}'
 
-    def subscribe(self, username):
-        self._ws = NonBlockingWebsocket(self.mirror_socket)
-        self._ws.start()
+    async def subscribe(self, username):
+        self._ws = await websockets.connect(self.mirror_socket)
 
-        greet = self._ws.recv()
+        greet = await self._ws.recv()
         greet = json.loads(greet)
         if len(greet['subs']) == 0:
-            target_user = self.api.get_user_details(username)
+            target_user = await self.api.get_user_details(username)
             target_user_id = target_user['user']['id']
 
             supported_eventsub_types = [
@@ -37,8 +37,8 @@ class TwitchEventSub:
                     'condition': condition,
                     'version': '1',
                 }
-                self._ws.send(json.dumps(payload))
-                status = self._ws.recv()
+                await self._ws.send(json.dumps(payload))
+                status = await self._ws.recv()
                 status = json.loads(status)
                 if status['type'] != 'subsuccess':
                     logger.warning(f'Failed to subscribe to {sub_type} EventSub topics - these events will not come through!!')
@@ -63,18 +63,21 @@ class TwitchEventSub:
                                 user_name, user_id)
             return norm_event
 
-    def read(self, timeout):
-        if self._ws.poll(timeout):
-            messages = self._ws.recv()
-            for message in messages.splitlines():
-                if message == 'PING':
-                    self._ws.send('PONG')
-                    continue
-                try:
-                    data = json.loads(message)
-                    if data.get('type') == 'event':
-                        yield self.normalize(data['data'])
-                    else:
-                        logger.warning(f'Unknown eventsub mirror data: {data}')
-                except json.JSONDecodeError as e:
-                    logger.error(f'Bad data from eventsub mirror: {message}')
+    async def read(self, timeout):
+        try:
+            messages = await asyncio.wait_for(self._ws.recv(), timeout)
+        except asyncio.exceptions.TimeoutError:
+            return
+
+        for message in messages.splitlines():
+            if message == 'PING':
+                await self._ws.send('PONG')
+                continue
+            try:
+                data = json.loads(message)
+                if data.get('type') == 'event':
+                    yield self.normalize(data['data'])
+                else:
+                    logger.warning(f'Unknown eventsub mirror data: {data}')
+            except json.JSONDecodeError as e:
+                logger.error(f'Bad data from eventsub mirror: {message}')

src/ovtk_audiencekit/chats/Twitch/sources/TwitchIRC.py

@@ -4,13 +4,13 @@ import re
 from itertools import chain, islice
 import logging
 from collections import OrderedDict
+import asyncio
 
-import websockets.exceptions
+import websockets
 from miniirc import ircv3_message_parser
 
 from ovtk_audiencekit.events.Message import Message, USER_TYPE
 from ovtk_audiencekit.events import Subscription
-from ovtk_audiencekit.utils import NonBlockingWebsocket
 
 from ..Events import Raid
 
@@ -43,56 +43,57 @@ class TwitchIRC:
         self._reply_buffer = OrderedDict()
         self._group_gifts = {}
 
-    def connect(self):
-        self._ws = NonBlockingWebsocket(WEBSOCKET_ADDRESS)
-        self._ws.start()
-        self._ws.send(f'PASS oauth:{self._token}')
-        self._ws.send(f'NICK {self._username}')
-        response = self._ws.recv()
+    async def connect(self):
+        self._ws = await websockets.connect(WEBSOCKET_ADDRESS)
+        await self._ws.send(f'PASS oauth:{self._token}')
+        await self._ws.send(f'NICK {self._username}')
+        response = await self._ws.recv()
         if any('Welcome, GLHF!' in msg for msg in response.splitlines()):
-            self._ws.send(f'JOIN #{self._channel_name}')
-            self._ws.send('CAP REQ :twitch.tv/tags')
-            self._ws.send('CAP REQ :twitch.tv/commands')
-            self._ws.send('CAP REQ :twitch.tv/membership')
+            await self._ws.send(f'JOIN #{self._channel_name}')
+            await self._ws.send('CAP REQ :twitch.tv/tags')
+            await self._ws.send('CAP REQ :twitch.tv/commands')
+            await self._ws.send('CAP REQ :twitch.tv/membership')
         else:
             raise TwitchIRCException(f'Got bad response during auth: {response}')
 
-    def read(self, timeout):
+    async def read(self, timeout):
         try:
-            if self._ws.poll(timeout):
-                messages = self._ws.recv()
-                for message in messages.splitlines():
-                    normalized = None
-                    cmd, hostmask, tags, args = ircv3_message_parser(message)
-                    logger.debug(f'cmd: {cmd}, hostmask: {hostmask}, args: {args}, tags: {tags}')
-                    if cmd == 'PRIVMSG':
-                        normalized = self.normalize_message(hostmask, tags, args)
-                    elif cmd == 'USERNOTICE':
-                        normalized = self.normalize_event(hostmask, tags, args)
-                    elif cmd == 'PING':
-                        self._ws.send(f"PONG {' '.join(args)}")
-                    elif cmd == 'RECONNECT':
-                        raise TimeoutError('Twitch API requested timeout')
-                    elif cmd == 'JOIN':
-                        self.users.append(hostmask[0])
-                    elif cmd == 'PART':
-                        try:
-                            self.users.remove(hostmask[0])
-                        except ValueError:
-                            pass
+            messages = await asyncio.wait_for(self._ws.recv(), timeout)
+        except asyncio.exceptions.TimeoutError:
+            return
+        try:
+            for message in messages.splitlines():
+                normalized = None
+                cmd, hostmask, tags, args = ircv3_message_parser(message)
+                logger.debug(f'cmd: {cmd}, hostmask: {hostmask}, args: {args}, tags: {tags}')
+                if cmd == 'PRIVMSG':
+                    normalized = self.normalize_message(hostmask, tags, args)
+                elif cmd == 'USERNOTICE':
+                    normalized = self.normalize_event(hostmask, tags, args)
+                elif cmd == 'PING':
+                    await self._ws.send(f"PONG {' '.join(args)}")
+                elif cmd == 'RECONNECT':
+                    raise TimeoutError('Twitch API requested timeout')
+                elif cmd == 'JOIN':
+                    self.users.append(hostmask[0])
+                elif cmd == 'PART':
+                    try:
+                        self.users.remove(hostmask[0])
+                    except ValueError:
+                        pass
 
-                    if normalized:
-                        self._reply_buffer[normalized.id] = normalized
-                        if len(self._reply_buffer.keys()) > self._reply_buffer_maxlen:
-                            self._reply_buffer.popitem()
-                        yield normalized
+                if normalized:
+                    self._reply_buffer[normalized.id] = normalized
+                    if len(self._reply_buffer.keys()) > self._reply_buffer_maxlen:
+                        self._reply_buffer.popitem()
+                    yield normalized
         except websockets.exceptions.ConnectionClosedError:
-            self.logger.info('Twitch websocket disconnected - trying reconnet')
-            self.connect()
+            self.logger.warning('Twitch websocket disconnected - trying reconnet')
+            await self.connect()
 
-    def send(self, message):
+    async def send(self, message):
         irc_msg = f'PRIVMSG #{self._username} :{message}'
-        self._ws.send(irc_msg)
+        await self._ws.send(irc_msg)
 
     def parse_badges(self, tags):
         if isinstance(tags.get('badges'), str):

src/ovtk_audiencekit/chats/YoutubeLive/YoutubeLive.py

@@ -1,9 +1,8 @@
-import os
 import json
 import webbrowser
 from enum import Enum, auto
 
-import requests
+import httpx
 
 from ovtk_audiencekit.chats import ChatProcess
 from ovtk_audiencekit.events.Message import Message, SysMessage, USER_TYPE
@@ -24,7 +23,7 @@ class YoutubeLivePollProcess(ChatProcess):
     GOOGLE_OAUTH_TOKEN_URL = 'https://accounts.google.com/o/oauth2/token'
     YOUTUBE_API_URL = 'https://www.googleapis.com/youtube/v3'
 
-    def __init__(self, *args, client_secrets_path=None):
+    async def setup(self, client_secrets_path=None):
         if client_secrets_path is None or not os.path.exists(client_secrets_path):
             raise ValueError('Missing client secrets')
 
@@ -33,7 +32,6 @@ class YoutubeLivePollProcess(ChatProcess):
             if client_secrets is None:
                 raise ValueError('Malformed client secrets file - missing installed section')
 
-        super().__init__(*args)
         self._client_secrets = client_secrets
         self._state_machine = self.bind_to_states(STATES)
 
@@ -44,14 +42,14 @@ class YoutubeLivePollProcess(ChatProcess):
         self._live_chat_id = None
         self._page_token = None
 
-    def loop(self, next_state):
+    async def loop(self, next_state):
         if self.state is None:
             if os.path.exists('refresh_token.secret'):
                 with open('refresh_token.secret', 'r') as f:
                     self._refresh_token = f.read()
                 return STATES.REFRESH, STATES.WAITING_FOR_BROADCAST
             return STATES.UNAUTHORIZED, STATES.WAITING_FOR_BROADCAST
-        return self._state_machine(self.state, next_state)
+        return await self._state_machine(self.state, next_state)
 
     def on_unauthorized(self, next_state):
         self.request_oauth_consent()
@@ -66,10 +64,10 @@ class YoutubeLivePollProcess(ChatProcess):
         self.get_fresh_access_token()
         return next_state
 
-    def on_waiting_for_broadcast(self, next_state):
-        response = requests.get(f'{self.__class__.YOUTUBE_API_URL}/liveBroadcasts',
-                                params={'part': 'snippet', 'broadcastStatus': 'active'},
-                                headers={'Authorization': f'Bearer {self._access_token}'})
+    async def on_waiting_for_broadcast(self, next_state):
+        response = await httpx.get(f'{self.__class__.YOUTUBE_API_URL}/liveBroadcasts',
+                                   params={'part': 'snippet', 'broadcastStatus': 'active'},
+                                   headers={'Authorization': f'Bearer {self._access_token}'})
 
         if (response.status_code == 401):
             return STATES.REFRESH, self.state
@@ -83,19 +81,19 @@ class YoutubeLivePollProcess(ChatProcess):
 
         return 30
 
-    def on_polling(self, next_state):
-        response = requests.get(f'{self.__class__.YOUTUBE_API_URL}/liveChat/messages',
-                                params={'liveChatId': self._live_chat_id,
-                                        'part': 'snippet,authorDetails',
-                                        'hl': 'en_US',
-                                        'pageToken': self._page_token},
-                                headers={'Authorization': f'Bearer {self._access_token}'}
-                                )
+    async def on_polling(self, next_state):
+        response = await httpx.get(f'{self.__class__.YOUTUBE_API_URL}/liveChat/messages',
+                                   params={'liveChatId': self._live_chat_id,
+                                           'part': 'snippet,authorDetails',
+                                           'hl': 'en_US',
+                                           'pageToken': self._page_token},
+                                   headers={'Authorization': f'Bearer {self._access_token}'}
+                                  )
 
         if (response.status_code == 401):
             return STATES.REFRESH, self.state
 
-        if response.status_code == requests.codes.ok:
+        if response.status_code == httpx.codes.ok:
             data = response.json()
             self._page_token = data['nextPageToken']
             if len(data['items']):
@@ -156,8 +154,8 @@ class YoutubeLivePollProcess(ChatProcess):
         url = f"{self._client_secrets['auth_uri']}?{param_str}"
         webbrowser.open(url)
 
-    def setup_oauth_consent(self, consent_code):
-        response = requests.post(self.__class__.GOOGLE_OAUTH_TOKEN_URL, data={
+    async def setup_oauth_consent(self, consent_code):
+        response = await httpx.post(self.__class__.GOOGLE_OAUTH_TOKEN_URL, data={
             'code': consent_code,
             'client_id': self._client_secrets['client_id'],
             'client_secret': self._client_secrets['client_secret'],
@@ -171,8 +169,8 @@ class YoutubeLivePollProcess(ChatProcess):
         self._access_token = auth['access_token']
         self._refresh_token = auth['refresh_token']
 
-    def get_fresh_access_token(self):
-        response = requests.post(self.__class__.GOOGLE_OAUTH_TOKEN_URL, data={
+    async def get_fresh_access_token(self):
+        response = await httpx.post(self.__class__.GOOGLE_OAUTH_TOKEN_URL, data={
             'client_id': self._client_secrets['client_id'],
             'client_secret': self._client_secrets['client_secret'],
             'refresh_token': self._refresh_token,

src/ovtk_audiencekit/cli/group.py

@@ -59,6 +59,12 @@ def cli(loglevel, show_time=False):
     logging.getLogger('numba').setLevel(logging.WARN)
     logging.getLogger('hypercorn.error').setLevel(logging.WARN)
     logging.getLogger('hypercorn.access').setLevel(logging.WARN)
+    logging.getLogger('httpx').setLevel(logging.WARN)
+    logging.getLogger('httpcore').setLevel(logging.INFO)
+    logging.getLogger('torio._extension.utils').setLevel(logging.WARN)
+    logging.getLogger('matplotlib').setLevel(logging.INFO)
+    logging.getLogger('fsspec').setLevel(logging.INFO)
+    logging.getLogger('TTS').setLevel(logging.INFO if loglevel == logging.DEBUG else logging.WARN)
     # Quiet warnings
     if loglevel > logging.DEBUG:
         warnings.filterwarnings("ignore")

src/ovtk_audiencekit/cli/start.py

@@ -27,5 +27,7 @@ def start(config_file, bus_bind=None, bus_port=None, web_bind=None, web_port=Non
         asyncio.run(main.run())
     except KeyboardInterrupt:
         pass
+    except asyncio.CancelledError:
+        pass
     finally:
         logger.info('Suya~')

src/ovtk_audiencekit/core/Audio.py

@@ -8,7 +8,6 @@ import pyaudio as pya
 import librosa
 import pytsmod as tsm
 import soundfile
-from aioprocessing import AioEvent
 
 # HACK: Redirect stderr to /dev/null to silence portaudio boot
 devnull = os.open(os.devnull, os.O_WRONLY)
@@ -26,100 +25,287 @@ os.close(old_stderr)
 logger = logging.getLogger(__name__)
 
 class Clip:
-    def __init__(self, path, samplerate=None, speed=1, keep_pitch=True, force_stereo=True):
-        self.path = path
-        raw, native_rate = librosa.load(self.path, sr=None, dtype='float32', mono=False)
+    def __init__(self, samples, samplerate):
+        self.samplerate = samplerate
+        self.samples = np.ascontiguousarray(samples, dtype='float32')
 
-        self.channels = raw.shape[0] if len(raw.shape) == 2 else 1
-        if force_stereo and self.channels == 1:
-            raw = np.resize(raw, (2,*raw.shape))
-            self.channels = 2
+    @classmethod
+    def from_file(cls, path):
+        samples, samplerate = librosa.load(path, sr=None, dtype='float32', mono=False)
+        return cls(samples, samplerate)
 
-        self.samplerate = samplerate or native_rate
-        if native_rate != self.samplerate:
-            raw = librosa.resample(raw, native_rate, self.samplerate, fix=True, scale=True)
+    @classmethod
+    def empty(cls, channels, samplerate, length=0):
+        samples = int(samplerate * length)
+        if channels > 1:
+            blank = np.zeros((channels, samples), dtype='float32')
+        else:
+            blank = np.zeros((samples,), dtype='float32')
+        return cls(blank, samplerate)
 
-        self.raw = np.ascontiguousarray(self._stereo_transpose(raw), dtype='float32')
+    def copy(self):
+        return Clip(self.samples, self.samplerate)
 
-        if speed != 1:
-            self.stretch(speed, keep_pitch=keep_pitch)
+    def save(self, filename):
+        soundfile.write(filename, self.samples, self.samplerate)
+
+    @property
+    def channels(self):
+        return self.samples.shape[0] if len(self.samples.shape) == 2 else 1
 
     @property
     def length(self):
-        return self.raw.shape[0] / self.samplerate
+        return self.samples.shape[-1]
 
-    def _stereo_transpose(self, ndata):
-        return ndata if self.channels == 1 else ndata.T
+    @property
+    def duration(self):
+        return self.length / self.samplerate
 
-    def stretch(self, speed, keep_pitch=True):
-        if keep_pitch:
-            stretched = tsm.wsola(self._stereo_transpose(self.raw), 1 / speed)
+    def __repr__(self):
+        return f"<Clip {self.duration}s @ {self.samplerate}hz ({self.channels} channels)>"
+
+    def resample(self, target_samplerate):
+        self.samples = librosa.resample(
+            self.samples, fix=True, scale=False,
+            orig_sr=self.samplerate, target_sr=target_samplerate,
+        )
+        self.samplerate = target_samplerate
+        return self
+
+    def resize(self, target_channels):
+        if target_channels == 1:
+            self.samples = librosa.to_mono(self.samples)
         else:
-            stretched = librosa.resample(self._stereo_transpose(self.raw), self.samplerate * speed, self.samplerate, fix=False, scale=True)
-        self.raw = np.ascontiguousarray(self._stereo_transpose(stretched), dtype='float32')
+            self.samples = np.resize(self.samples, (target_channels, self.length))
+        return self
 
-    def save(self, filename):
-        soundfile.write(filename, self._stereo_transpose(self.raw), self.samplerate)
+    def format_to(self, other):
+        """Convert format to match device or clip"""
+        if not isinstance(other, Clip) and not isinstance(other, Output):
+            raise TypeError("Must be Clip or Output")
+        if other.samplerate != self.samplerate:
+            self.resample(other.samplerate)
+        if other.channels != self.channels:
+            self.resize(other.channels)
+        return self
+
+    def trim(self, aggressive=False):
+        """Remove leading and trailing silence"""
+        self.samples, _ = librosa.effects.trim(self.samples, top_db=10 if aggressive else 30)
+        return self
+
+    def stretch(self, speed):
+        self.samples = tsm.wsola(self.samples, 1 / speed)
+        return self
+
+    def pitch(self, cents, resample=False):
+        if resample:
+            speed = 2 ** (cents / 1200)
+            self *= speed
+        else:
+            steps = cents / 100
+            self.samples = librosa.effects.pitch_shift(self.samples, sr=self.samplerate, n_steps=steps)
+        return self
+
+    def __mul__(self, speed):
+        """Change playback speed"""
+        if not isinstance(speed, int) and not isinstance(speed, float):
+            raise TypeError("Must be a number")
+        stretched = librosa.resample(
+            self.samples, fix=False, scale=False,
+            orig_sr=self.samplerate * speed, target_sr=self.samplerate,
+        )
+        return Clip(stretched, self.samplerate)
+
+    def __imul__(self, speed):
+        """Change playback speed"""
+        if not isinstance(speed, int) and not isinstance(speed, float):
+            raise TypeError("Must be a number")
+        self.samples = librosa.resample(
+            self.samples, fix=False, scale=False,
+            orig_sr=self.samplerate * speed, target_sr=self.samplerate,
+        )
+        return self
+
+    def _opcheck(self, other):
+        if not isinstance(other, Clip):
+            raise TypeError("Clips can only operate with other clips!")
+        if self.channels != other.channels or self.samplerate != other.samplerate:
+            raise TypeError("Clips must have the same samplerate and number of channels!")
+
+    def __and__(self, other):
+        """Mix two clips together"""
+        self._opcheck(other)
+        len_difference = self.length - other.length
+        if len_difference == 0:
+            mix = self.samples + other.samples
+        else:
+            shortest, longest = (other.samples, self.samples) if len_difference > 0 else (self.samples, other.samples)
+            # Expand to longest
+            shortest = np.pad(shortest, ((0, 0), (0, abs(len_difference))))
+            mix = shortest + longest
+        return Clip(mix, self.samplerate)
+
+    def __iand__(self, other):
+        """
+        Mix two clips together.
+        If incoming clip is longer, it will be croped to the length of the existing clip.
+        """
+        self._opcheck(other)
+        len_difference = self.length - other.length
+        if len_difference == 0:
+            self.samples += other.samples
+        else:
+            # Crop other to self
+            length = min(self.length, other.length)
+            self.samples[:, :length] += other[:, :length]
+        return self
+
+    def __add__(self, other):
+        """Concatenate two clips"""
+        self._opcheck(other)
+        extended = np.concatenate((self.samples, other.samples), axis=1)
+        return Clip(extended, self.samplerate)
+
+    def __iadd__(self, other):
+        """Concatenate two clips"""
+        self._opcheck(other)
+        self.samples = np.concatenate((self.samples, other.samples), axis=1)
+        return self
 
 
 class Stream:
-    def __init__(self, clip, output_index, buffer_length=4096):
-        self.clip = clip
+    def __init__(self, samples, length):
+        self.samples = samples
+
         self.pos = 0
-        self.playing = False
-        self._end_event = AioEvent()
-        self._stream = pyaudio.open(
-            output_device_index=output_index,
+        self.active = False
+        self.loop = asyncio.get_event_loop()
+        self.end_event = asyncio.Event()
+        self._dead = False
+        self.autoclose = False
+
+    def start(self, retrigger=True):
+        if self._dead:
+            raise RuntimeError("Cannot start closed Stream!")
+        if retrigger:
+            self.pos = 0
+        self.active = True
+        self.end_event.clear()
+
+    async def done(self):
+        await self.end_event.wait()
+        return self
+
+    def next(self, length):
+        newpos = self.pos + length
+        # Entire chunk is before origin
+        if newpos < 0:
+            chunk = np.zeros((length, self.samples.shape[1]), dtype='float32')
+        # Origin is within chunk
+        elif self.pos < 0:
+            chunk = self.samples[0:newpos]
+            chunk = np.pad(chunk, ((length - chunk.shape[0], 0), (0, 0)))
+        # Playhead is within bounds
+        elif self.pos >= 0 and newpos < self.samples.shape[0]:
+            chunk = self.samples[self.pos:newpos]
+        # Last chunk (or out of bounds)
+        elif newpos >= self.samples.shape[0]:
+            chunk = self.samples[self.pos:]
+            chunk = np.pad(chunk, ((0, length - chunk.shape[0]), (0, 0)))
+            if self.autoclose:
+                self.close()
+            else:
+                self.stop()
+
+        self.pos = newpos
+
+        return chunk
+
+    def stop(self):
+        self.active = False
+        self.loop.call_soon_threadsafe(self.end_event.set)
+
+    def close(self):
+        if self.active:
+            self.stop()
+        self._dead = True
+
+
+class Output:
+    def __init__(self, target, channels=2, samplerate=None, buffer_length=4096):
+        if isinstance(target, str):
+            index = Output.find(target)
+        elif isinstance(target, int):
+            index = target
+        else:
+            raise ValueError("Invalid target type")
+
+        if samplerate is None:
+            for rate in [44100, 48000, 96000]:
+                if Output.check_rate(index, channels, rate):
+                    samplerate = rate
+                    break
+            if samplerate is None:
+                logger.warn('Target audio device does not claim to support common sample rates! Assuming 44.1khz')
+                samplerate = 44100
+        else:
+            if not Output.check_rate(index, channels, rate):
+                logger.warn('Target audio device does not claim to support requested rate!')
+
+        self.channels = channels
+        self.samplerate = samplerate
+        self.buffer_length = buffer_length
+
+        self.streams = set()
+        self._buffer = np.zeros((buffer_length, self.channels), dtype='float32')
+        self._pa_stream = pyaudio.open(
+            output_device_index=index,
             format=pya.paFloat32,
-            channels=self.clip.channels,
-            rate=self.clip.samplerate,
+            channels=channels,
+            rate=samplerate,
             frames_per_buffer=buffer_length,
             output=True,
             stream_callback=self._read_callback,
             start=False)
 
-    def _play(self):
-        self.playing = True
-        self.pos = 0
+    def check_clip_format(self, clip):
+        return clip.samplerate == self.samplerate and clip.channels == self.channels
 
-        if not self._stream.is_active():
-            self._stream.start_stream()
+    def start(self):
+        self._pa_stream.start_stream()
 
-    def play(self):
-        self._end_event.clear()
-        self._play()
-        self._end_event.wait(timeout=self.clip.length)
+    def play(self, clip, start=True, autoclose=False):
+        self.check_clip_format(clip)
+        samples = clip.samples.T if self.channels > 1 else clip.samples
 
-    async def aplay(self):
-        self._end_event.clear()
-        self._play()
-        try:
-            await self._end_event.coro_wait(timeout=self.clip.length)
-        except asyncio.CancelledError:
-            self.playing = False
-            self._stream.stop_stream()
+        stream = Stream(samples, self.buffer_length)
+        stream.autoclose = autoclose
+        self.streams.add(stream)
+        if start:
+            stream.start()
+
+        if not self._pa_stream.is_active():
+            self._pa_stream.start_stream()
+
+        return stream
+
+    def stop(self):
+        self._pa_stream.stop_stream()
 
     def close(self):
-        self._stream.stop_stream()
-        self._stream.close()
+        self._pa_stream.close()
+        for stream in self.streams:
+            stream.close()
 
     def _read_callback(self, in_data, frame_count, time_info, status):
-        if self.clip.channels > 1:
-            buffer = np.zeros((frame_count, self.clip.channels), dtype='float32')
-        else:
-            buffer = np.zeros((frame_count,), dtype='float32')
-
-        if self.playing:
-            newpos = self.pos + frame_count
-            clip_chunk = self.clip.raw[self.pos:newpos]
-            self.pos = newpos
-            buffer[0:clip_chunk.shape[0]] = clip_chunk
-
-            if self.pos >= self.clip.raw.shape[0]:
-                self.playing = False
-                self._end_event.set()
-
-        return buffer, pya.paContinue
+        self._buffer[:] = 0
+        for stream in list(self.streams):
+            if stream.active:
+                self._buffer += stream.next(frame_count)
+            if stream._dead:
+                self.streams.remove(stream)
+        return self._buffer, pya.paContinue
 
     @staticmethod
     def check_rate(index, channels, rate):
@@ -132,7 +318,7 @@ class Stream:
             return False
 
     @staticmethod
-    def find_output_index(output):
+    def find(output):
         if output is None:
             return None
 

src/ovtk_audiencekit/core/Config.py

@@ -139,7 +139,7 @@ def parse_kdl_deep(path, relativeto=None):
     if relativeto:
         path = os.path.normpath(os.path.join(relativeto, path))
 
-    with open(path, 'r') as f:
+    with open(path, 'r', encoding='utf-8') as f:
         try:
             config = kdl.parse(f.read(), kdl_parse_config)
             for node in config.nodes:

src/ovtk_audiencekit/core/MainProcess.py

@@ -1,10 +1,8 @@
 import importlib
-from multiprocessing import Lock
 import asyncio
 from datetime import datetime, timedelta
 import logging
 import os
-import os.path
 import pathlib
 import sys
 import signal
@@ -61,27 +59,22 @@ class MainProcess:
         self.max_concurrent = max_concurrent
 
         self.chat_processes = {}
+        self.chat_tasks = set()
         self.plugins = {}
         self.event_queue = asyncio.Queue()
 
-        # Init websocket server (event bus)
-        # HACK: Must be done here to avoid shadowing its asyncio loop
-        self.server_process = WebsocketServerProcess(*self.bus_conf)
-        self.server_process.start()
-
         # Save sys.path since some config will clobber it
         self._initial_syspath = sys.path
 
+        if os.name == 'nt':
+            asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
+
     def _unload_plugin(self, plugin_name):
         plugin = self.plugins[plugin_name]
         plugin.close()
         del self.plugins[plugin_name]
         del plugin
 
-    def _get_event_from_pipe(self, pipe):
-        event = pipe.recv()
-        self.event_queue.put_nowait(event)
-
     def _setup_webserver(self):
         self.webserver = Quart(__name__, static_folder=None, template_folder=None)
         listen = ':'.join(self.web_conf)
@@ -137,10 +130,10 @@ class MainProcess:
                     if event is None:
                         break
                 else:
-                    self.server_process.message_pipe.send(event)
+                    await self.bus_server.send(event)
                 logger.debug(f'Event after plugin chain - {event}')
             elif isinstance(event, Delete):
-                self.server_process.message_pipe.send(event)
+                await self.bus_server.send(event)
             else:
                 logger.error(f'Unknown data in event loop - {event}')
 
@@ -157,7 +150,6 @@ class MainProcess:
 
     async def user_setup(self):
         config = kdl.Document(list(parse_kdl_deep(self.config_path)))
-        stdin_lock = Lock()
         # Load secrets
         secrets = {}
         if node := config.get('secrets'):
@@ -181,8 +173,11 @@ class MainProcess:
                     chat_module = import_or_reload_mod(module_name,
                                                        default_package='ovtk_audiencekit.chats',
                                                        external=False)
-                    chat_process = chat_module.Process(stdin_lock, chat_name, **node.props, **secrets_for_mod)
+                    chat_process = chat_module.Process(chat_name, self.event_queue)
                     self.chat_processes[chat_name] = chat_process
+                    res = chat_process.setup(**node.props, **secrets_for_mod)
+                    if asyncio.iscoroutine(res):
+                        await res
                 except Exception as e:
                     raise ValueError(f'Failed to initalize {module_name} module "{chat_name}" - {e}')
 
@@ -190,13 +185,9 @@ class MainProcess:
             logger.warning('No chats configured!')
 
         # Start chat processes
+        loop = asyncio.get_event_loop()
         for process in self.chat_processes.values():
-            process.start()
-            # Bridge pipe to asyncio event loop
-            pipe = process.event_pipe
-            # REVIEW: This does not work on windows!!!! add_reader is not implemented
-            #         in a way that supports pipes on either windows loop runners
-            asyncio.get_event_loop().add_reader(pipe.fileno(), lambda pipe=pipe: self._get_event_from_pipe(pipe))
+            self.chat_tasks.add(loop.create_task(process.run()))
 
         # Load plugins
         global_ctx = {}
@@ -228,6 +219,8 @@ class MainProcess:
                 except Exception as e:
                     raise ValueError(f'Failed to initalize {module_name} plugin "{plugin_name}" - {e}')
 
+        self.bus_server.update_eventclasses()
+
         # Run plugin definitions
         with progressbar(list(config.nodes), label=f"Executing {self.config_path}") as bar:
             for node in bar:
@@ -241,13 +234,10 @@ class MainProcess:
                     await plugin_module._kdl_call(node, global_ctx)
 
     async def user_shutdown(self):
+        for chat_task in self.chat_tasks:
+            chat_task.cancel()
+        self.chat_tasks.clear()
         for process_name, process in list(reversed(self.chat_processes.items())):
-            pipe = process.event_pipe
-            process.control_pipe.send(ShutdownRequest('root'))
-            process.join(5)
-            if process.exitcode is None:
-                process.terminate()
-            asyncio.get_event_loop().remove_reader(pipe.fileno())
             del self.chat_processes[process_name]
         for plugin_name in list(reversed(self.plugins.keys())):
             # NOTE: The plugin will likely stick around in memory for a bit after this,
@@ -255,6 +245,19 @@ class MainProcess:
             self._unload_plugin(plugin_name)
         sys.path = self._initial_syspath
 
+    async def _discount_repl(self):
+        loop = asyncio.get_event_loop()
+        # REVIEW: Not a good UX at the moment (as new logs clobber the terminal entry)
+        while True:
+            line = await loop.run_in_executor(None, sys.stdin.readline)
+            line = line.strip()
+            logger.debug(f'Got terminal input: {line}')
+            if line == 'reload':
+                self.reload_ev.set()
+            elif line == 'quit':
+                self.shutdown_ev.set()
+                break
+
     async def run(self):
         self.shutdown_ev = asyncio.Event()
         self.reload_ev = asyncio.Event()
@@ -263,26 +266,11 @@ class MainProcess:
 
         try:
             # System setup
-            ## Bridge websocket server pipe to asyncio loop
-            ## REVIEW: This does not work on windows!!!! add_reader is not implemented
-            ##         in a way that supports pipes on either windows loop runners
-            ws_pipe = self.server_process.message_pipe
-            loop.add_reader(ws_pipe.fileno(), lambda: self._get_event_from_pipe(ws_pipe))
-            ## Register stdin handler
-            reader = asyncio.StreamReader()
-            await loop.connect_read_pipe(lambda: asyncio.StreamReaderProtocol(reader), sys.stdin)
-            async def discount_repl():
-                # REVIEW: Not a good UX at the moment (as new logs clobber the terminal entry)
-                async for line in reader:
-                    line = line.strip()
-                    if line == b'reload':
-                        self.reload_ev.set()
-                    elif line == b'quit':
-                        self.shutdown_ev.set()
-            self.cli_task = loop.create_task(discount_repl())
-            ## Scheduler for timed tasks
-            self._skehdule = TimedScheduler(max_tasks=1)
-            self._skehdule.start()
+            ## Make stdin handler
+            self.cli_task = loop.create_task(self._discount_repl())
+            ## Init websocket server (external end of the event bus)
+            self.bus_server = WebsocketServerProcess(self.event_queue, *self.bus_conf)
+            self.bus_task = loop.create_task(self.bus_server.run())
             ## UI server
             serve_coro = self._setup_webserver()
             self.webserver_task = loop.create_task(serve_coro)
@@ -299,7 +287,7 @@ class MainProcess:
 
                     logger.info(f'Ready to rumble! Press Ctrl+C to shut down')
                     reload_task = loop.create_task(self.reload_ev.wait())
-                    done, pending = await asyncio.wait([*user_tasks, self.webserver_task, reload_task], return_when=asyncio.FIRST_COMPLETED)
+                    done, pending = await asyncio.wait([*user_tasks, self.webserver_task, self.bus_task, reload_task], return_when=asyncio.FIRST_COMPLETED)
 
                     if reload_task in done:
                         logger.warn('Reloading (some events may be missed!)')
@@ -324,6 +312,7 @@ class MainProcess:
                         serve_coro = self._setup_webserver()
                         self.webserver_task = loop.create_task(serve_coro)
                     else:
+                        logger.debug(f'Task {done} completed - assuming something went wrong!')
                         break
         except KeyboardInterrupt:
             pass
@@ -341,4 +330,6 @@ class MainProcess:
                 task.cancel()
             await self.user_shutdown()
             self.webserver_task.cancel()
-            self.server_process.terminate()
+            self.bus_task.cancel()
+            for task in asyncio.all_tasks():
+                task.cancel()

src/ovtk_audiencekit/core/Plugins.py

@@ -2,6 +2,7 @@ from abc import ABC, abstractmethod
 import logging
 import asyncio
 import os.path
+from pathlib import Path
 import sys
 import copy
 
@@ -31,7 +32,7 @@ class OvtkBlueprint(quart.Blueprint):
 
     def render(self, name, **kwargs):
         """render_template that prefers the plugin-specific templates"""
-        full = self.template_folder / name
+        full = Path(self.template_folder) / name
         if os.path.exists(full):
             template_string = None
             with open(full, 'r') as template_file:

src/ovtk_audiencekit/core/WebsocketServerProcess.py

@@ -1,34 +1,31 @@
 import asyncio
 import json
-from multiprocessing import Process, Pipe
 import logging
 
 import websockets
 
 from ovtk_audiencekit.events import Event
-from ovtk_audiencekit.utils import get_subclasses
+from ovtk_audiencekit.utils import get_subclasses, format_exception
 
 logger = logging.getLogger(__name__)
 
 
-class WebsocketServerProcess(Process):
-    def __init__(self, bind, port):
-        super().__init__()
-
+class WebsocketServerProcess:
+    def __init__(self, event_queue, bind, port):
         self._bind = bind
         self._port = port
 
-        self._pipe, self._caller_pipe = Pipe()
+        self.event_queue = event_queue
+        self._send_queue = asyncio.Queue()
         self.clients = set()
 
+        self.update_eventclasses()
+
+    def update_eventclasses(self):
         self._event_classes = get_subclasses(Event)
 
-    @property
-    def message_pipe(self):
-        return self._caller_pipe
-
     # Data input (external application socket -> plugin/chat pipe)
-    async def handle_websocket(self, ws, path):
+    async def _handle_client(self, ws):
         self.clients.add(ws)
         try:
             async for message in ws:
@@ -39,7 +36,7 @@ class WebsocketServerProcess(Process):
                             type = type[0]
                         event_class = next(cls for cls in self._event_classes if cls.__name__ == type)
                         event = event_class.hydrate(**data.get('data', {}))
-                        self._pipe.send(event)
+                        self.event_queue.put_nowait(event)
                     else:
                         logger.warn('Unknown data recieved on websocket', message)
                 except json.decoder.JSONDecodeError as e:
@@ -51,47 +48,44 @@ class WebsocketServerProcess(Process):
         except websockets.exceptions.ConnectionClosedError:
             pass
         except asyncio.CancelledError:
-            ws.close()
+            await ws.close()
         finally:
             self.clients.discard(ws)
 
     # Data output (plugin/chat pipe -> external application socket)
-    async def handle_pipe(self, pipe_ready):
+    async def send(self, event):
+        await self._send_queue.put(event)
+
+    async def _send_loop(self):
         while True:
-            # Let other co-routines process until file descriptor is readable
-            await pipe_ready.wait()
-            pipe_ready.clear()
-            # Check if messages exist on the pipe before attempting to recv
-            # to avoid accidentally blocking the event loop when file
-            # descriptor does stuff we don't expect
-            if not self._pipe.poll():
-                continue
-            event = self._pipe.recv()
+            event = await self._send_queue.get()
             # Serialize and send to registered clients
             serialized = event.serialize()
             if self.clients:
-                await asyncio.wait([self.safe_send(client, serialized) for client in self.clients])
+                await asyncio.wait([self._safe_send(client, serialized) for client in self.clients])
 
-    async def safe_send(self, client, serialized):
+    async def _safe_send(self, client, serialized):
         try:
             await client.send(serialized)
         except (websockets.exceptions.ConnectionClosedError, websockets.exceptions.ConnectionClosedOK):
             self.clients.discard(client)
 
-    def run(self):
-        # Setup asyncio websocket server
-        start_server = websockets.serve(self.handle_websocket, self._bind, self._port)
-        asyncio.get_event_loop().run_until_complete(start_server)
+    async def run(self):
+        loop = asyncio.get_event_loop()
+        tasks = set()
 
-        # Make an awaitable object that flips when the pipe's underlying file descriptor is readable
-        pipe_ready = asyncio.Event()
-        # REVIEW: This does not work on windows!!!!
-        asyncio.get_event_loop().add_reader(self._pipe.fileno(), pipe_ready.set)
-        # Make and start our infinite pipe listener task
-        asyncio.get_event_loop().create_task(self.handle_pipe(pipe_ready))
+        # Setup websocket server (input)
+        self.ws_server = await websockets.serve(self._handle_client, self._bind, self._port)
+        tasks.add(loop.create_task(self.ws_server.serve_forever()))
+        # Setup sending loop (output)
+        tasks.add(loop.create_task(self._send_loop()))
 
-        # Keep the asyncio code running in this thread until explicitly stopped
         try:
-            asyncio.get_event_loop().run_forever()
-        except KeyboardInterrupt:
-            return 0
+            await asyncio.gather(*tasks)
+        except Exception as e:
+            logger.critical(f'Failure in bus process - {e}')
+            logger.info(format_exception(e))
+            raise e
+        finally:
+            for task in tasks:
+                task.cancel()

src/ovtk_audiencekit/core/__init__.py

@@ -1,4 +1,4 @@
 from .WebsocketServerProcess import WebsocketServerProcess
 from .Plugins import PluginBase, PluginError
 from .MainProcess import MainProcess
-from .Audio import Clip, Stream
+from .Audio import Clip, Output

src/ovtk_audiencekit/plugins/AudioAlert/AudioAlert.py

@@ -4,25 +4,19 @@ from collections import deque
 import maya
 
 from ovtk_audiencekit.core import PluginBase
-from ovtk_audiencekit.core import Clip, Stream
+from ovtk_audiencekit.core.Audio import Clip, Output
 
 class AudioAlert(PluginBase):
-    def setup(self, output=None, timeout_min=1, sample_rate=None, buffer_length=4096, force_stereo=True):
-        self._cleanup_task = asyncio.create_task(self._cleanup())
-
-        self.force_stereo = force_stereo
+    def setup(self, output=None, stereo=True, timeout_min=1):
+        self.channels = 2 if stereo else 1
         self.timeout_min = timeout_min
+
+        self.output = Output(output, self.channels)
+
+        self._cleanup_task = asyncio.create_task(self._cleanup())
+        self.tasks = set()
         self.clips = {}
         self.streams = {}
-        self.tasks = set()
-        self.buffer_length = int(buffer_length)
-        self.output_index = Stream.find_output_index(output)
-        if sample_rate is None:
-            try:
-                sample_rate = next((rate for rate in [44100, 48000] if Stream.check_rate(self.output_index, 1, rate)))
-            except StopIteration:
-                self.logger.warn('Target audio device does not claim to support common sample rates! Attempting playback at native rate of audio')
-        self.sample_rate = sample_rate
 
     async def run(self, path, speed=1, keep_pitch=False, wait=False, poly=1, **kwargs):
         poly = int(poly)
@@ -30,8 +24,12 @@ class AudioAlert(PluginBase):
         clip = self.clips.get(key, [None, None])[0]
 
         if clip is None:
-            clip = Clip(path, speed=speed, keep_pitch=keep_pitch,
-                        samplerate=self.sample_rate, force_stereo=self.force_stereo)
+            clip = Clip.from_file(path).format_to(self.output)
+            if speed != 1:
+                if keep_pitch:
+                    clip.stretch(speed)
+                else:
+                    clip *= speed
             self.clips[key] = [clip, maya.now()]
         else:
             self.clips[key][1] = maya.now()
@@ -47,26 +45,24 @@ class AudioAlert(PluginBase):
         if len(stream_dq) == stream_dq.maxlen:
             stream_dq.rotate(1)
             stream = stream_dq[0]
+            stream.stop()
+            stream.start()
         else:
-            stream = Stream(clip, self.output_index,
-                            buffer_length=self.buffer_length)
+            stream = self.output.play(clip)
             stream_dq.append(stream)
 
+        task = asyncio.create_task(stream.done())
+        task.add_done_callback(self.tasks.discard)
+        self.tasks.add(task)
 
         if wait:
-            await stream.aplay()
-        else:
-            task = asyncio.create_task(stream.aplay())
-            task.add_done_callback(self.tasks.discard)
-            self.tasks.add(task)
+            await task
 
     def close(self):
         self._cleanup_task.cancel()
         for task in self.tasks:
             task.cancel()
-        for stream_dq in self.streams.values():
-            for stream in stream_dq:
-                stream.close()
+        self.output.close()
 
     async def _cleanup(self):
         while True:

src/ovtk_audiencekit/plugins/Jail/JailPlugin.py

@@ -4,7 +4,7 @@ import pickle
 import os
 
 import maya
-from requests.exceptions import HTTPError
+from httpx import HTTPError
 from owoify.owoify import owoify, Owoness
 
 from ovtk_audiencekit.core import PluginBase

src/ovtk_audiencekit/plugins/Shoutout/ShoutoutPlugin.py

@@ -1,6 +1,6 @@
 from argparse import ArgumentError
 
-from requests.exceptions import HTTPError
+from httpx import HTTPError
 
 from ovtk_audiencekit.core import PluginBase
 from ovtk_audiencekit.plugins.builtins.Command import Command, CommandTypes
@@ -17,10 +17,10 @@ class ShoutoutPlugin(PluginBase):
             self.command.add_arg('username', type=CommandTypes.Username)
             self.plugins['command'].register_help(self.command)
 
-    def make_shoutout_msg(self, username, via):
+    async def make_shoutout_msg(self, username, via):
         if isinstance(self.chats[via], Twitch):
             try:
-                detail = self.chats[via].shared.api.get_user_details(username)
+                detail = await self.chats[via].shared.api.get_user_details(username)
             except HTTPError:
                 detail = None
 
@@ -43,7 +43,7 @@ class ShoutoutPlugin(PluginBase):
             self.logger.warning('Shoutout attempted with no event context!')
 
 
-    def on_bus_event(self, event):
+    async def on_bus_event(self, event):
         if isinstance(event, Message):
             if self.command and self.command.invoked(event):
                 try:
@@ -53,7 +53,7 @@ class ShoutoutPlugin(PluginBase):
                     self.chats[event.via].send(msg)
                     return None
 
-                text = self.make_shoutout_msg(args.get('username'), event.via)
+                text = await self.make_shoutout_msg(args.get('username'), event.via)
                 if text is not None:
                     msg = SysMessage(self._name, text)
                     self.broadcast(msg)

src/ovtk_audiencekit/plugins/TTS/TTS.py

@@ -1,6 +1,7 @@
 import uuid
 import os
 import asyncio
+import shutil
 
 from TTS.utils.synthesizer import Synthesizer
 from TTS.utils.manage import ModelManager
@@ -8,7 +9,7 @@ from TTS.config import load_config
 
 from ovtk_audiencekit.core import PluginBase
 from ovtk_audiencekit.events import Message, SysMessage
-from ovtk_audiencekit.core import Clip, Stream
+from ovtk_audiencekit.core.Audio import Clip, Output
 from ovtk_audiencekit.core.Data import CACHE_DIR
 
 
@@ -17,20 +18,17 @@ class TextToSpeechPlugin(PluginBase):
                  engine="tts_models/en/ljspeech/tacotron2-DDC", speaker_wav=None, **kwargs):
 
         self.speaker_wav = speaker_wav
-
-        self.output_index = Stream.find_output_index(output)
-        sample_rate = None
-        try:
-            sample_rate = next((rate for rate in [44100, 48000] if Stream.check_rate(self.output_index, 1, rate)))
-        except StopIteration:
-            self.logger.warn('Target audio device does not claim to support common sample rates! Attempting playback at native rate of audio')
-        self.sample_rate = sample_rate
+        self.output = Output(output)
 
         conf_overrides = {k[2:]: v for k, v in kwargs.items() if k.startswith('o_')}
 
         self.cache_dir = os.path.join(CACHE_DIR, 'tts')
         os.makedirs(os.path.dirname(self.cache_dir), exist_ok=True)
 
+        self.output_dir = os.path.join(self.cache_dir, 'outputs')
+        os.makedirs(self.output_dir, exist_ok=True)
+        self.cache = {}
+
         self.cuda = cuda
 
         manager = ModelManager(output_prefix=CACHE_DIR) # HACK: coqui automatically adds 'tts' subdir
@@ -52,8 +50,8 @@ class TextToSpeechPlugin(PluginBase):
             config_path = override_conf_path
 
         self.synthesizer = Synthesizer(
-            model_path,
-            config_path,
+            tts_checkpoint=model_path,
+            tts_config_path=config_path,
             vocoder_checkpoint=vocoder_path,
             vocoder_config=vocoder_config_path,
             use_cuda=self.cuda,
@@ -64,37 +62,46 @@ class TextToSpeechPlugin(PluginBase):
     def close(self):
         for task in self.tasks:
             task.cancel()
+        self.output.close()
+        shutil.rmtree(self.output_dir)
 
-    def make_tts_wav(self, text, filename=None):
+    async def text_to_clip(self, text):
         # Force punctuation (keeps the models from acting unpredictably)
         text = text.strip()
         if not any([text.endswith(punc) for punc in '.!?:']):
             text += '.'
 
-        if filename is None:
-            filename = os.path.join(self.cache_dir, f'{uuid.uuid1()}.wav')
-
-        self.logger.info(f'Generating TTS "{text}"...')
-        if self.speaker_wav:
-            wav = self.synthesizer.tts(text, None, 'en', self.speaker_wav)
+        if cached := self.cache.get(hash(text)):
+            self.logger.info(f'Cache hit - {cached}')
+            return Clip.from_file(cached)
         else:
-            wav = self.synthesizer.tts(text)
+            self.logger.info(f'Generating TTS "{text}"...')
+            filename = os.path.join(self.output_dir, f'{uuid.uuid1()}.wav')
 
-        self.synthesizer.save_wav(wav, filename)
-        self.logger.info(f'Done - saved as {filename}')
-        return filename
+            if self.speaker_wav:
+                fn = lambda _text: self.synthesizer.tts(_text, None, 'en', self.speaker_wav)
+            else:
+                fn = lambda _text: self.synthesizer.tts(_text)
+
+            # Do TTS processing in a thread to avoid blocking main loop
+            samples = await asyncio.get_running_loop().run_in_executor(None, fn, text)
+            clip = Clip(samples, self.synthesizer.output_sample_rate)
+
+            clip.save(filename)
+            self.cache[hash(text)] = filename
+            self.logger.info(f'Done - cached as {filename}')
+
+        return clip
 
     async def run(self, text, *args, _ctx={}, wait=False, **kwargs):
         try:
-            # Do TTS processing in a thread to avoid blocking main loop
-            filename = await asyncio.get_running_loop().run_in_executor(None, self.make_tts_wav, text)
+            clip = await self.text_to_clip(text)
+            clip.format_to(self.output)
 
-            # TODO: Play direct from memory
-            clip = Clip(filename, force_stereo=True, samplerate=self.sample_rate)
-            stream = Stream(clip, self.output_index)
             async def play():
                 try:
-                    await stream.aplay()
+                    stream = self.output.play(clip)
+                    await stream.done()
                 finally:
                     stream.close()
                     os.remove(os.path.join(self.cache_dir, filename))

src/ovtk_audiencekit/plugins/builtins/Command.py

@@ -103,14 +103,14 @@ class CommandPlugin(PluginBase):
             raise TypeError('Should be instance of Command')
         self.commands[cmd.name] = (cmd, None, True)
 
-    def run(self, name, help=None, display=False, _children=None, **kwargs):
+    def run(self, name, help=None, display=False, required_level=None, _children=None, **kwargs):
         actionnode = next((node for node in _children if node.name == 'do'), None)
         if actionnode is None:
             raise ValueError('Command defined without an action (`do` tag)')
 
         aliases = [node.args for node in _children if node.name == 'alias']
         aliases = list(itertools.chain(*aliases))
-        cmd = Command(name, help, aliases=aliases)
+        cmd = Command(name, help, aliases=aliases, required_level=required_level)
 
         for argnode in [node for node in _children if node.name == 'arg']:
             if argnode.props.get('type'):

src/ovtk_audiencekit/plugins/builtins/Scene/Plugin.py

@@ -34,7 +34,7 @@ class ScenePlugin(PluginBase):
 
         self.blueprint.add_url_rule('/', 'ctrlpanel', self.ui_ctrlpanel)
         self.blueprint.add_url_rule('/<name>/<cmd>', 'api-sceneset', self.ui_setscene)
-        self.blueprint.add_url_rule('/monitor', 'monitor', self.ui_monitor_ws, is_websocket=True)
+        self.blueprint.add_url_rule('/monitor', 'monitor', self.ui_monitor_ws, websocket=True)
 
     async def run(self, name, _children=None, _ctx={}, active=None, group=None, oneshot=False, **kwargs):
         if _children is None:

src/ovtk_audiencekit/utils/NonBlockingWebsocket.py

@@ -1,60 +0,0 @@
-from multiprocessing import Process, Pipe
-import logging
-import asyncio
-
-import websockets
-
-logger = logging.getLogger(__name__)
-
-
-class NonBlockingWebsocket(Process):
-    def __init__(self, ws_uri):
-        super().__init__()
-        self.uri = ws_uri
-        self._pipe, self._caller_pipe = Pipe()
-        self.daemon = True
-        self._send_queue = asyncio.Queue()
-
-    async def _setup(self):
-        self._ws = await websockets.connect(self.uri)
-
-    def send(self, data):
-        self._caller_pipe.send(data)
-
-    async def _send(self, pipe_ready):
-        while True:
-            await pipe_ready.wait()
-            pipe_ready.clear()
-            if not self._pipe.poll():
-                continue
-
-            data = self._pipe.recv()
-            await self._ws.send(data)
-
-    def poll(self, *args):
-        return self._caller_pipe.poll(*args)
-
-    def recv(self, *args):
-        return self._caller_pipe.recv(*args)
-
-    async def _read(self):
-        while True:
-            data = await self._ws.recv()
-            self._pipe.send(data)
-
-    def run(self):
-        loop = asyncio.new_event_loop()
-        # Setup ws client
-        loop.run_until_complete(self._setup())
-        # Make an awaitable object that flips when the pipe's underlying file descriptor is readable
-        pipe_ready = asyncio.Event()
-        loop.add_reader(self._pipe.fileno(), pipe_ready.set)
-        # Make and start our infinite tasks
-        loop.create_task(self._send(pipe_ready))
-        loop.create_task(self._read())
-
-        # Keep the asyncio code running in this thread until explicitly stopped
-        try:
-            loop.run_forever()
-        except KeyboardInterrupt:
-            return 0

src/ovtk_audiencekit/utils/__init__.py

@@ -1,4 +1,3 @@
-from .NonBlockingWebsocket import NonBlockingWebsocket
 from .make_sync import make_sync
 from .get_subclasses import get_subclasses
 from .format_exception import format_exception