Better youtube chat integration via pytchat (KaitoCross's fork)

.gitignore

@@ -1,7 +1,4 @@
-.venv/
-.pdm-python
-.pdm-build/
-__pycache__/
-/dist/
 *.secret*
 secrets.kdl
+__pycache__/
+/*-log

Pipfile

@@ -0,0 +1,42 @@
+[[source]]
+url = "https://pypi.org/simple"
+verify_ssl = true
+name = "pypi"
+
+[packages]
+requests = "*"
+websockets = "*"
+miniirc = "*"
+num2words = "*"
+pyaudio = "==0.2.*"
+numpy = "*"
+click = "*"
+owoify-py = "==2.*"
+kdl-py = "*"
+maya = "*"
+multipledispatch = "*"
+blessed = "*"
+appdirs = "*"
+watchdog = "*"
+mido = "*"
+python-rtmidi = "*"
+librosa = "==0.8.*"
+pytsmod = "*"
+quart = "==0.17.*"
+aioscheduler = "*"
+TTS = "==0.9.*"
+torch = "==1.13.*"
+simpleobsws = "*"
+python-socketio = {extras = ["client"], version = "*"}
+aioprocessing = "*"
+pytchat = { git="https://github.com/KaitoCross/pytchat.git" }
+
+[requires]
+python_version = "3.10"
+
+[dev-packages]
+pipenv-setup = "*"
+
+[scripts]
+start = "python audiencekit.py start"
+ws = "python audiencekit.py ws"

README.md

@@ -27,7 +27,7 @@ trigger event="Raid" {
 ## For developers
 Extending audiencekit's automation abilities is made as simple as possible. For example, a plugin that automatically changes "simp" to "shrimp" in every message (as seen by those consuming this project's output) is simply:
 ```python
-from ovtk_audiencekit.core import PluginBase
+from ovtk_audiencekit.plugins import PluginBase
 from ovtk_audiencekit.events import Message
 
 class Plugin(PluginBase):
@@ -48,13 +48,13 @@ Beside the plugin system, audiencekit's biggest feature is that it streams the e
 ## Manual / Development Install
 1. Install dependencies
 
-   You'll need Python 3.10, PDM, and some audio libraries. Not to fear though, there is [instructions on how to do that for your specific OS!](https://git.skeh.site/skeh/ovtk_audiencekit/wiki/Dependency-Setup)
+   You'll need Python 3.10 or above, pipenv, JACK and PortAudio libraries. Not to fear though, there is [instructions on how to do that for your specific OS!](https://git.skeh.site/skeh/ovtk_audiencekit/wiki/Dependency-Setup)
 
 2. Download the repository and open a terminal inside it
 
    Extract the [tar](https://git.skeh.site/skeh/ovtk_audiencekit/archive/main.tar.gz) / [zip](https://git.skeh.site/skeh/ovtk_audiencekit/archive/main.zip), or clone via `git`~
 
-3. Run `pdm sync`
+3. Run `pipenv install`
 
    Some dependencies may fail to install initially, but should be retried using a method that works automatically (your terminal should show "Retrying initially failed dependencies..." or similar).
 
@@ -68,15 +68,15 @@ If you want some more tutorializing, head on over to the [Wiki](https://git.skeh
 + Read up a bit on [KDL's syntax][kdl] if you haven't already
 + Crack open your favorite text editor and follow the instructions in the `config.kdl` file
 + Open a terminal in the project location and start it using either:
-   + `pdm run start`, or
-   + [activate the venv](https://pdm.fming.dev/latest/usage/venv/#activate-a-virtualenv) once per terminal session, and then use `./audiencekit.py start` thereafter. This is recommended for development usage.
-+ Test your configuration using the `ws` subcommand (`pdm run ws` or `./audiencekit.py ws`)
-  + The built-in CLI will help for this and more! Just add "--help" to the end of anything, ie `./audiencekit.py --help`, `pdm run ws mkevent --help`, etc
+   + `pipenv run start`, or
+   + `pipenv shell` once per terminal session, and then `./audiencekit.py start` thereafter
++ Test your configuration using the `ws` subcommand (`pipenv run ws` or `./audiencekit.py ws`)
+  + The built-in CLI will help for this and more! Just add "--help" to the end of anything, ie `./audiencekit.py --help`, `pipenv run ws mkevent --help`, etc
 + Make mistakes and [ask questions](https://birb.space/@skeh)!
 
 -----------
 
-Made with :heart: by the [Vtopia Collective][vtopia] and contributors
+Made with :heart: by the [Vtopia Collective][vtopia] and contributers
 
 
 [kdl]: https://kdl.dev

audiencekit.py

@@ -1,6 +1,4 @@
 #!/usr/bin/env python3
-import sys
-sys.path.insert(0, 'src')
 from ovtk_audiencekit.cli import cli
 
 if __name__ == '__main__':

config.kdl

@@ -1,7 +1,7 @@
 /* Comments surrounded by asterisks and slashes (like this one) are instructions,
-   comments staring with two slashes are valid example configuration! */
+   comments staring with two slashes are example configuration! */
 
-/* Step 1: Give it the rights (to party) */
+/* Step 1: Add your credentials */
 secrets {
   /* Generate credentials via https://ovtk.skeh.site/twitch/auth
      and paste them between the curly braces below */
@@ -17,52 +17,46 @@ secrets {
   }
 }
 
-
-/* Step 2: Import what you need
+/* Step 2: Import modules
 
   There are two types of modules, chats and plugins:
   + Chats are self-explanitory: the event providers - livestream services.
   + Plugins are the heart of the system, and can both monitor livestream events
-    and be called on directly to perform actions.
+    and be called by name in your config files to perform actions.
 
-  Some plugins (called "builtins") are always available, but others are not loaded
-  until you ask. Chats are never loaded by default.
+  Some plugins (called "builtins" and always starting with a lowercase character)
+  are always available, but others are not loaded until you ask to conserve
+  resources. Chats are never loaded by default.
 
   You can load either by using the "chat" or "plugin" node respectively,
   and supplying the name of the module. You can also rename them by seperating
-  the desired name and node name with a colon (:), otherwise they take on the
-  module name and throw an error when two are used at the same time.
+  the module name and the new name with a colon (:)
 */
 // chat "Twitch" channel_name="MyTwitchChannel"
-// guest:chat "Twitch" channel_name="CollabChannel" readonly=true
-// aplay:plugin "AudioAlert" output="ALSA:default"
+// chat "Twitch:otherguy" channel_name="SomeOtherChannel"
+// plugin "AudioAlert" output="ALSA:default"
 
 
-/* Step 3: Get silly with it
+/* Step 3: Kick it
 
    Some example automations are provided below to get you started.
-   Config is always valid KDL (https://kdl.dev/), but has some special quirks.
-   See the wiki for more details, but the jist is:
-   + A plugin's main routine is run by making a node with its name
-     + Subroutines can be run by seperating with a dot: `name.routine`
-     + Returned values can be saved to the "context" (see below) by using a colon: `val:name`
-   + Config is evaluated top to bottom, outter-most runs by default
-   + Plugins are free to parse their children (inside braces) however they like, so not all nodes are the same! See `command` and `scene`.
-   + Custom type `t` (for template) can be used to insert data from the context, builtin `set` can be used to write to it.
-     + Plugins often use this to share additional data!
+   See the wiki for more details: https://git.skeh.site/skeh/ovtk_audiencekit/wiki
 */
 
-/* Self-promo every 2 hours */
-// cue hours=2 {
+/* Self-promo every 10 min */
+// cue {
+//   every hours=2 {
 //     reply "Like the setup? Run it yourself! https://git.skeh.site/explore/repos?q=ovtk&topic=1"
 //   }
+// }
+
 
 /* Call an existing shoutout bot on raid */
 // trigger event="Raid" {
-//   reply (t)"!so {event.from_channel}"
+//   reply (arg)"!so {event.from_channel}"
 // }
 
-/* Lurk command (!lurk) */
+/* Lurk command */
 // command "lurk" help="Lurke modeo" display=true {
 //   do {
 //     reply "The Twitch algorithm thanks you for the lurk~" display=true
@@ -71,57 +65,16 @@ secrets {
 
 /* TTS for every donation event */
 // plugin "TTS"
-// trigger monitization="0.01-" {
-//   TTS (t)"{event.user_name} says: {event.text}"
+// trigger monitization="0.01-" source="self" {
+//   TTS (arg)"{event.user_name} says: {event.text}"
 // }
 
-/* Control OBS from your midi controller - see https://github.com/obsproject/obs-websocket/blob/master/docs/generated/protocol.md#requests */
-// plugin "OBS" password="A Very Secret Example Password"
-// plugin "Midi"
-// Midi.listen "control_change" channel=1 control=1 value=127 {
-//   OBS "SetInputMute" inputName="Microphone" inputMuted=true
-// }
-// Midi.listen "control_change" channel=1 control=1 value=0 {
-//   OBS "SetInputMute" inputName="Microphone" inputMuted=false
-// }
-
-/* Talk to your midi rig too! */
+/* Voice effect channel redeem (via midi to physical effect rack / DAW) */
 // trigger event="ChannelPointRedemption" action="Sound Sussy" {
-//   Midi "program_change" channel=1 program=1
-//   cue minutes=5 {
-//     Midi "program_change" channel=1 program=0
+//   midi "sysex" data="FunEffect;1"
+//   cue {
+//     after minutes=5 {
+//       midi "sysex" data="FunEffect;0"
 //     }
 //   }
-
-/* Control just about anything else, from anywhere (http://localhost:8000/scene/) else! */
-// plugin "OSC" port=9000
-// plugin "OBS" password="A Very Secret Example Password"
-// scene "Brb / starting" {
-//   OBS "SetCurrentProgramScene" sceneName="Starting"
-//   OSC "/track/1/mute" (u8)1
-//
-//   exit {
-//     OSC "/track/1/mute" (u8)0
 // }
-// }
-// scene "Live" {
-//   OBS "SetCurrentProgramScene" sceneName="Live"
-// }
-
-
-/* Step 4: Organize
-
-   Split config into multiple files to keep sane, or even define entirely seperate
-   setups for special occasions and run by providing their path to the `start` command.
-*/
-// import "base.kdl"
-// import "secrets_i_promise_not_to_open_on_air.kdl"
-
-
-/* Step 5: Reach for the stars!
-
-   Still can't make that stupid idea a reality? Custom plugins are a single file in "src/ovtk_audiencekit/plugins" away~
-   Learn a little Python, hack on any of the other plugins,
-   toss PluginBase.py at ChatGPT and try your luck,
-   or nicely ask your local birdy <3
-*/

ovtk_audiencekit/chats/ChatProcess.py

@@ -1,12 +1,12 @@
 from abc import ABC, abstractmethod
 from multiprocessing import Process, Pipe, Manager
+from traceback import format_exception
 import sys
 import os
 import json
 import logging
 
 from ovtk_audiencekit.events import Event
-from ovtk_audiencekit.utils import format_exception
 
 
 class GracefulShutdownException(Exception):
@@ -133,7 +133,7 @@ class ChatProcess(Process, ABC):
                 return 0
             except Exception as e:
                 self.logger.error(f'Uncaught exception in {self._name}: {e}')
-                self.logger.debug(format_exception(e))
+                self.logger.debug(''.join(format_exception(None, e, e.__traceback__)))
                 return 1
             finally:
                 self.on_exit()

ovtk_audiencekit/chats/Twitch/TwitchProcess.py

@@ -27,8 +27,6 @@ class TwitchProcess(ChatProcess):
                  botname=None, emote_res=4.0,
                  # EventSub options
                  eventsub=True, eventsub_host='wss://ovtk.skeh.site/twitch',
-                 # BTTV integration
-                 bttv=False,
                  # Inheritance boilerplate
                  **kwargs):
         super().__init__(*args, **kwargs)
@@ -63,7 +61,7 @@ class TwitchProcess(ChatProcess):
             self.eventsub = TwitchEventSub(self.api, eventsub_host)
             self._sources.append(self.eventsub)
 
-        self.bttv = BTTV(target_data['user']['id']) if bttv else None
+        self.bttv = BTTV(target_data['user']['id'])
 
 
     def loop(self, next_state):
@@ -100,7 +98,7 @@ class TwitchProcess(ChatProcess):
             for event in chain(*(source.read(0.1) for source in self._sources)):
                 # Retarget event
                 event.via = self._name
-                if self.bttv and isinstance(event, Message):
+                if isinstance(event, Message):
                     event = self.bttv.hydrate(event)
                 self.publish(event)
             return 0

ovtk_audiencekit/chats/Twitch/sources/TwitchEventSub.py

@@ -2,7 +2,7 @@ import json
 import logging
 
 from ovtk_audiencekit.utils import NonBlockingWebsocket
-from ovtk_audiencekit.core.Data import ovtk_user_id
+from ovtk_audiencekit.core.Config import ovtk_user_id
 from ovtk_audiencekit.events import Follow
 
 from ..Events import ChannelPointRedemption

ovtk_audiencekit/chats/YoutubeLive/YoutubeLive.py

@@ -0,0 +1,72 @@
+import time
+from enum import Enum, auto
+from itertools import chain
+
+from ovtk_audiencekit.chats import ChatProcess
+from ovtk_audiencekit.events import Message, SysMessage, USER_TYPE
+
+import pytchat
+
+
+class STATES(Enum):
+    WAITING = auto()
+    READING = auto()
+
+
+class YoutubeProcess(ChatProcess):
+    def __init__(self, *args, video_id=None, **kwargs):
+        super().__init__(*args, **kwargs)
+
+        self._state_machine = self.bind_to_states(STATES)
+        self._video_id = video_id
+
+        self.state = STATES.CONNECTING
+        self._chat = pytchat.create(video_id=self._video_id)
+
+    def normalize_event(self, event):
+        user_type = USER_TYPE.USER
+        if event.author.isChatOwner:
+            user_type = USER_TYPE.OWNER
+        elif event.author.isChatModerator:
+            user_type = USER_TYPE.MODERATOR
+        elif event.author.isChatSponsor:
+            user_type = USER_TYPE.PATRON
+
+        msg = Message(self._name, event.message,
+                      event.author.name, event.author.channelId, user_type,
+                      id=event.id)
+        return msg
+
+    def loop(self, next_state):
+        return self._state_machine(self.state, next_state)
+
+    def on_state_enter(self, new_state):
+        status_messages = {
+            STATES.READING: 'Connected to channel!',
+        }
+
+        message = status_messages.get(new_state)
+        if message is not None:
+            sys_msg = SysMessage(self._name, message)
+            self.publish(sys_msg)
+
+    def on_waiting(self, next_state):
+        if self._chat.is_alive():
+            return STATES.READING
+        return 5
+
+    def on_reading(self, next_state):
+        try:
+            if not self._chat.is_alive():
+                return STATES.TIMEOUT
+
+            for raw_event in self._chat.get().sync_items():
+                normalized_event = self.normalize_event(raw_event)
+                self.publish(normalized_event)
+
+            return 0
+        except TimeoutError:
+            return STATES.TIMEOUT
+
+    def on_failure(self, next_state):
+        return None

ovtk_audiencekit/cli/group.py

@@ -1,5 +1,4 @@
 import logging
-import warnings
 
 import click
 from blessed import Terminal
@@ -16,11 +15,8 @@ class CustomFormatter(logging.Formatter):
         logging.CRITICAL: term.white_on_red_bold
     }
 
-    def __init__(self, show_time, show_loc):
-        if show_loc:
+    def __init__(self, show_time):
         format = "%(levelname)s:%(name)s (%(filename)s:%(lineno)d): %(message)s"
-        else:
-            format = "%(levelname)s:%(name)s: %(message)s"
         if show_time:
             format = "%(asctime)s:" + format
         super().__init__(format)
@@ -47,7 +43,7 @@ def cli(loglevel, show_time=False):
     logger.setLevel(loglevel)
     log_handler = logging.StreamHandler()
     log_handler.setLevel(loglevel)
-    log_handler.setFormatter(CustomFormatter(show_time, loglevel==logging.DEBUG))
+    log_handler.setFormatter(CustomFormatter(show_time))
     logger.addHandler(log_handler)
     # Quiet the depencency loggers
     logging.getLogger('websockets.server').setLevel(logging.WARN)
@@ -57,8 +53,3 @@ def cli(loglevel, show_time=False):
     logging.getLogger('simpleobsws').setLevel(logging.INFO)
     logging.getLogger('quart.serving').setLevel(logging.WARN)
     logging.getLogger('numba').setLevel(logging.WARN)
-    logging.getLogger('hypercorn.error').setLevel(logging.WARN)
-    logging.getLogger('hypercorn.access').setLevel(logging.WARN)
-    # Quiet warnings
-    if loglevel > logging.DEBUG:
-        warnings.filterwarnings("ignore")

ovtk_audiencekit/cli/start.py

@@ -0,0 +1,20 @@
+import logging
+
+import click
+
+from ovtk_audiencekit.core import MainProcess
+from .group import cli
+
+logger = logging.getLogger(__name__)
+
+
+@cli.command()
+@click.argument('config_file', type=click.Path('r'), default='config.kdl')
+@click.option('--port', default='8080')
+@click.option('--bind', default='127.0.0.1')
+def start(config_file, port=None, bind=None):
+    """Start audiencekit server"""
+    logger.info('Hewwo!!')
+    main = MainProcess(config_file, port, bind)
+    main.run()
+    logger.info('Bye bye~')

ovtk_audiencekit/cli/websocketutils.py

@@ -76,11 +76,10 @@ class EventCommandFactory(click.MultiCommand):
                 param = click.Option(param_decls=[f'--{field.name}'], type=param_type, default=field.default, show_default=True)
             params.append(param)
 
+        if target_event.__dict__.get('__cli__'):
+            params.extend(target_event.__cli__())
+
         command = click.Command(name, callback=mkevent_generic, params=params, help=target_event.__doc__)
-
-        if target_event.__dict__.get('_cli'):
-            command = target_event._cli(command)
-
         return command
 
 

ovtk_audiencekit/core/Clip.py

@@ -25,103 +25,7 @@ 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)
 
-        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
-
-        self.samplerate = samplerate or native_rate
-        if native_rate != self.samplerate:
-            raw = librosa.resample(raw, native_rate, self.samplerate, fix=True, scale=True)
-
-        self.raw = np.ascontiguousarray(self._stereo_transpose(raw), dtype='float32')
-
-        if speed != 1:
-            self.stretch(speed, keep_pitch=keep_pitch)
-
-    @property
-    def length(self):
-        return self.raw.shape[0] / self.samplerate
-
-    def _stereo_transpose(self, ndata):
-        return ndata if self.channels == 1 else ndata.T
-
-    def stretch(self, speed, keep_pitch=True):
-        if keep_pitch:
-            stretched = tsm.wsola(self._stereo_transpose(self.raw), 1 / speed)
-        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')
-
-    def save(self, filename):
-        soundfile.write(filename, self._stereo_transpose(self.raw), self.samplerate)
-
-
-class Stream:
-    def __init__(self, clip, output_index, buffer_length=4096):
-        self.clip = clip
-        self.pos = 0
-        self.playing = False
-        self._end_event = AioEvent()
-        self._stream = pyaudio.open(
-            output_device_index=output_index,
-            format=pya.paFloat32,
-            channels=self.clip.channels,
-            rate=self.clip.samplerate,
-            frames_per_buffer=buffer_length,
-            output=True,
-            stream_callback=self._read_callback,
-            start=False)
-
-    def _play(self):
-        self.playing = True
-        self.pos = 0
-
-        if not self._stream.is_active():
-            self._stream.start_stream()
-
-    def play(self):
-        self._end_event.clear()
-        self._play()
-        self._end_event.wait(timeout=self.clip.length)
-
-    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()
-
-    def close(self):
-        self._stream.stop_stream()
-        self._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
-
-    @staticmethod
 def check_rate(index, channels, rate):
     try:
         return pyaudio.is_format_supported(rate,
@@ -131,6 +35,99 @@ class Stream:
     except ValueError:
         return False
 
+alt_rates = [44100, 48000]
+class Clip:
+    def __init__(self, path, output_index, buffer_length=2048, speed=1, force_stereo=True):
+        _raw, native_rate = librosa.load(path, sr=None, dtype='float32', mono=False)
+        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
+
+        target_samplerate = native_rate
+        if not check_rate(output_index, self._channels , native_rate):
+            try:
+                target_samplerate = next((rate for rate in alt_rates if check_rate(output_index, self._channels , rate)))
+            except StopIteration:
+                logger.warn('Target audio device does not claim to support any sample rates! Attempting playback at native rate')
+        self._samplerate = target_samplerate
+
+        if native_rate != self._samplerate:
+            _raw = librosa.resample(_raw, native_rate, self._samplerate, fix=True, scale=True)
+
+        self._raw = np.ascontiguousarray(self._stereo_transpose(_raw), dtype='float32')
+
+        if speed != 1:
+            self.stretch(speed)
+
+        self._pos = 0
+        self._playing = False
+        self._end_event = AioEvent()
+        self._stream = pyaudio.open(
+            output_device_index=output_index,
+            format=pya.paFloat32,
+            channels=self._channels,
+            rate=self._samplerate,
+            frames_per_buffer=buffer_length,
+            output=True,
+            stream_callback=self._read_callback,
+            start=False)
+
+    @property
+    def length(self):
+        return self._raw.shape[0] / self._samplerate
+
+    def _stereo_transpose(self, ndata):
+        return ndata if self._channels == 1 else ndata.T
+
+    def stretch(self, speed):
+        stretched = tsm.wsola(self._stereo_transpose(self._raw), speed)
+        self._raw = np.ascontiguousarray(self._stereo_transpose(stretched), dtype='float32')
+
+    def save(self, filename):
+        soundfile.write(filename, self._stereo_transpose(self._raw), self._samplerate)
+
+    def _play(self):
+        self._playing = True
+        self._pos = 0
+
+        if not self._stream.is_active():
+            self._stream.start_stream()
+
+    def play(self):
+        self._play()
+        self._end_event.wait(timeout=self.length)
+
+    async def aplay(self):
+        self._end_event.clear()
+        self._play()
+        try:
+            await self._end_event.coro_wait(timeout=self.length)
+        except asyncio.CancelledError:
+            self._playing = False
+            self._stream.stop_stream()
+
+    def close(self):
+        self._stream.close()
+
+    def _read_callback(self, in_data, frame_count, time_info, status):
+        if self._channels > 1:
+            buffer = np.zeros((frame_count, self._channels), dtype='float32')
+        else:
+            buffer = np.zeros((frame_count,), dtype='float32')
+
+        if self._playing:
+            newpos = self._pos + frame_count
+            clip_chunk = self._raw[self._pos:newpos]
+            self._pos = newpos
+            buffer[0:clip_chunk.shape[0]] = clip_chunk
+
+            if self._pos >= self._raw.shape[0]:
+                self._playing = False
+                self._end_event.set()
+
+        return buffer, pya.paContinue
+
     @staticmethod
     def find_output_index(output):
         if output is None:

ovtk_audiencekit/core/Config.py

@@ -2,7 +2,24 @@ import os
 import random
 
 import appdirs
+from kdl import ParseConfig
 
+def csv_parser(text, fragment):
+    text = fragment.fragment[1:-1]
+    return [field.strip() for field in text.split(',')]
+
+def semisv_parser(text, fragment):
+    text = fragment.fragment[1:-1]
+    return [field.strip() for field in text.split(';')]
+
+customParsers = {
+    'list': csv_parser,
+    'csv': csv_parser,
+    'semisv': semisv_parser,
+}
+
+kdl_parse_config = ParseConfig(valueConverters=customParsers)
+kdl_reserved = ['secrets', 'chat', 'plugin', 'import']
 
 CACHE_DIR = appdirs.user_cache_dir('audiencekit', 'ovtk')
 DATA_DIR = appdirs.user_data_dir('audiencekit', 'ovtk')

ovtk_audiencekit/core/MainProcess.py

@@ -0,0 +1,241 @@
+import importlib
+from multiprocessing import Lock
+import asyncio
+from datetime import datetime, timedelta
+from traceback import format_exception
+import logging
+import os
+
+import kdl
+from click import progressbar
+from aioscheduler import TimedScheduler
+from quart import Quart
+
+from ovtk_audiencekit.core import WebsocketServerProcess
+from ovtk_audiencekit.core.Config import kdl_parse_config, kdl_reserved
+from ovtk_audiencekit.events import Event, Delete
+from ovtk_audiencekit.chats.ChatProcess import ShutdownRequest
+from ovtk_audiencekit.plugins import builtins
+from ovtk_audiencekit.plugins.PluginBase import PluginError
+
+logger = logging.getLogger(__name__)
+
+def parse_kdl_deep(path, relativeto=None):
+    if relativeto:
+        path = os.path.normpath(os.path.join(relativeto, path))
+
+    with open(path, 'r') as f:
+        try:
+            config = kdl.parse(f.read(), kdl_parse_config)
+        except kdl.errors.ParseError as e:
+            e.file = path
+            raise e
+
+    for node in config.nodes:
+        if node.name == 'import':
+            yield from parse_kdl_deep(node.args[0], relativeto=os.path.dirname(path))
+        else:
+            yield node
+
+
+class MainProcess:
+    def __init__(self, config_path, port, bind):
+        super().__init__()
+        self.config_path = config_path
+        self.port = port
+        self.bind = bind
+
+        self.chat_processes = {}
+        self.plugins = {}
+
+    @staticmethod
+    def get_external_module_names(node, store):
+        if len(node.args) == 0:
+            raise ValueError(f'Invalid arguments - usage: {node.name} module')
+        nameparts = node.args[0].split(':')
+        module_name = nameparts[0]
+        instance_name = nameparts[1] if len(nameparts) == 2 else module_name
+        if store.get(instance_name):
+            if instance_name != module_name:
+                raise ValueError(f"Multiple nodes named {instance_name}, please specify unique names as the second argument")
+            else:
+                raise ValueError(f"Multiple definitions of {instance_name} exist, please specify unique names as the second argument")
+        return module_name, instance_name
+
+    async def handle_events(self):
+        while True:
+            event = await self.event_queue.get()
+            logger.info(event)
+
+            if isinstance(event, Event):
+                for plugin_name, plugin in list(self.plugins.items()):
+                    try:
+                        event = plugin.on_bus_event(event)
+                        if asyncio.iscoroutinefunction(plugin.on_bus_event):
+                            event = await event
+                    except PluginError as e:
+                        if e.fatal:
+                            logger.critical(f'Failure when processing {e.source} ({e}) - disabling...')
+                        else:
+                            logger.warning(f'Encounterd error when processing {e.source} ({e})')
+                        logger.debug(''.join(format_exception(None, e, e.__traceback__)))
+                        if e.fatal:
+                            self.plugins[e.source].__del__()
+                            del self.plugins[e.source]
+                    except Exception as e:
+                        logger.critical(f'Failure when processing {plugin_name} ({e}) - disabling...')
+                        logger.debug(''.join(format_exception(None, e, e.__traceback__)))
+                        self.plugins[plugin_name].__del__()
+                        del self.plugins[plugin_name]
+                    if event is None:
+                        break
+                else:
+                    self.server_process.message_pipe.send(event)
+                logger.debug(f'Event after plugin chain - {event}')
+            elif isinstance(event, Delete):
+                self.server_process.message_pipe.send(event)
+            else:
+                logger.error(f'Unknown data in event loop - {event}')
+
+    def setup(self):
+        config = kdl.Document(list(parse_kdl_deep(self.config_path)))
+
+        stdin_lock = Lock()
+        # Load secrets
+        secrets = {}
+        if node := config.get('secrets'):
+            for module in node.nodes:
+                fields = secrets.get(module.name, {})
+                for node in module.nodes:
+                    fields[node.name] = node.args[0] if len(node.args) == 1 else node.args
+                secrets[module.name] = fields
+
+        # Dynamically import chats
+        with progressbar(list(config.getAll('chat')), label="Preparing modules (chats)", item_show_func=lambda i: i and i.args[0]) as bar:
+            for node in bar:
+                module_name, chat_name = self.get_external_module_names(node, self.chat_processes)
+                secrets_for_mod = secrets.get(module_name, {})
+
+                try:
+                    chat_module = importlib.import_module(f'.{module_name}', package='ovtk_audiencekit.chats')
+                    chat_process = chat_module.Process(stdin_lock, chat_name, **node.props, **secrets_for_mod)
+                    self.chat_processes[chat_name] = chat_process
+                except Exception as e:
+                    raise ValueError(f'Failed to initalize {module_name} module "{chat_name}" - {e}')
+
+        if len(self.chat_processes.keys()) == 0:
+            logger.warning('No chats configured!')
+
+        # Start chat processes
+        for process in self.chat_processes.values():
+            process.start()
+
+        # Load plugins
+        self.event_queue = asyncio.Queue()
+        ## Builtins
+        for node_name in builtins.__all__:
+            self.plugins[node_name] = builtins.__dict__[node_name](self.chat_processes, self.event_queue, node_name)
+        ## Dynamic
+        with progressbar(list(config.getAll('plugin')), label="Preparing modules (plugins)", item_show_func=lambda i: i and i.args[0]) as bar:
+            for node in bar:
+                module_name, plugin_name = self.get_external_module_names(node, self.plugins)
+                secrets_for_mod = secrets.get(module_name, {})
+
+                try:
+                    plugin_module = importlib.import_module(f'.{module_name}', package='ovtk_audiencekit.plugins')
+                    plugin = plugin_module.Plugin(self.chat_processes, self.event_queue, plugin_name,
+                                                  **node.props, **secrets_for_mod, _children=node.nodes)
+                    self.plugins[plugin_name] = plugin
+
+                    # Register UI with webserver
+                    self.webserver.register_blueprint(plugin.blueprint, url_prefix=f'/{plugin_name}')
+                except Exception as e:
+                    raise ValueError(f'Failed to initalize {module_name} plugin "{plugin_name}" - {e}')
+
+        # Run plugin definitions
+        with progressbar(list(config.nodes), label=f"Executing {self.config_path}") as bar:
+            for node in bar:
+                if node.name in kdl_reserved:
+                    continue
+                plugin_name = node.name
+                plugin_module = self.plugins.get(plugin_name)
+                if plugin_module is None:
+                    logger.error(f'Unknown plugin: {node.name}')
+                else:
+                    asyncio.get_event_loop().run_until_complete(plugin_module._run(*node.args, **node.props, _children=node.nodes))
+
+        # Register watchable handles
+        self.pipes = [process.event_pipe for process in self.chat_processes.values()]
+        self.pipes.append(self.server_process.message_pipe)
+
+    async def tick_plugins(self):
+        for plugin in self.plugins.values():
+            res = plugin.tick(0.5)
+            if asyncio.iscoroutinefunction(plugin.tick):
+                await res
+        self._skehdule.schedule(self.tick_plugins(), datetime.utcnow() + timedelta(seconds=0.5))
+
+    def run(self):
+        # Start websocket server
+        self.server_process = WebsocketServerProcess(self.port, self.bind)
+        self.server_process.start()
+
+        self.webserver = Quart(__name__)
+
+        loop = asyncio.get_event_loop()
+        sys_tasks = []
+
+        try:
+            # Do initial setup
+            self.setup()
+            self._skehdule = TimedScheduler()
+
+            sys_tasks.append(loop.create_task(self.tick_plugins()))
+            sys_tasks.append(loop.create_task(self.handle_events()))
+
+            async def start_scheduler():
+                self._skehdule.start()
+            sys_tasks.append(loop.create_task(start_scheduler()))
+
+            async def start_uiwebserver():
+                try:
+                    # HACK: eats the KeyboardInterrupt - maybe others too
+                    await self.webserver.run_task()
+                except KeyboardInterrupt:
+                    pass
+                except Exception as e:
+                    logger.critical(f'Failure in web process - {e}')
+                    logger.debug(''.join(format_exception(None, e, e.__traceback__)))
+                finally:
+                    raise KeyboardInterrupt()
+            sys_tasks.append(loop.create_task(start_uiwebserver()))
+
+            def get_event(pipe):
+                event = pipe.recv()
+                self.event_queue.put_nowait(event)
+            for pipe in self.pipes:
+                # REVIEW: This does not work on windows!!!!
+                loop.add_reader(pipe.fileno(), lambda pipe=pipe: get_event(pipe))
+
+            logger.info('Ready to rumble!')
+            loop.run_forever()
+        except KeyboardInterrupt:
+            pass
+        except kdl.errors.ParseError as e:
+            if (e.file):
+                logger.critical(f'Invalid configuration in {e.file} at {e.line}:{e.col} - {e.msg}')
+            else:
+                logger.critical(f'Invalid configuration - {e.msg}')
+        except Exception as e:
+            logger.critical(f'Failure in core process - {e}')
+            logger.debug(''.join(format_exception(None, e, e.__traceback__)))
+        finally:
+            logger.warn('Closing up shop...')
+            for task in sys_tasks:
+                task.cancel()
+            for process in self.chat_processes.values():
+                process.control_pipe.send(ShutdownRequest('root'))
+                process.join(5)
+                if process.exitcode is None:
+                    process.terminate()
+            self.server_process.terminate()

ovtk_audiencekit/core/WebsocketServerProcess.py

@@ -12,7 +12,7 @@ logger = logging.getLogger(__name__)
 
 
 class WebsocketServerProcess(Process):
-    def __init__(self, bind, port):
+    def __init__(self, port, bind):
         super().__init__()
 
         self._bind = bind

ovtk_audiencekit/core/__init__.py

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

ovtk_audiencekit/events/Control.py

@@ -0,0 +1,20 @@
+from dataclasses import dataclass, field
+import json
+
+from ovtk_audiencekit.events import Event
+
+
+@dataclass
+class Control(Event):
+    """Generic inter-bus communication"""
+    target: str
+    data: dict = field(default_factory=dict)
+
+    def __repr__(self):
+        return f"Control message : target = {self.target}, data = {self.data}"
+
+    def serialize(self):
+        return json.dumps({
+            'type': [cls.__name__ for cls in self.__class__.__mro__],
+            'data': {**self.data, 'target': self.target},
+        })

ovtk_audiencekit/events/Message.py

@@ -43,13 +43,14 @@ class Message(Event):
         return super().hydrate(user_type=user_type, **kwargs)
 
     @classmethod
-    def _cli(cls, cmd):
+    def __cli__(cls):
         def dono(ctx, param, value):
             if value:
                 return [value, value]
 
-        cmd.params.append(click.Option(['--monitization', '-m'], type=click.FLOAT, callback=dono))
-        return cmd
+        return [
+            click.Option(['--monitization', '-m'], type=click.FLOAT, callback=dono),
+        ]
 
 
 @dataclass

ovtk_audiencekit/events/Subscription.py

@@ -28,12 +28,11 @@ class Subscription(Event):
         return f"Subcription from {self.user_name or 'anonymous'} to {recipent} - tier = {self.tier}"
 
     @classmethod
-    def _cli(cls, cmd):
+    def __cli__(cls):
         def userfactory(ctx, param, value):
             if value:
                 return [User(user_name=name, user_id=name) for name in value]
 
-        cmd.params.append(
-            click.Option(['--gifted_to', '-g'], type=click.STRING, callback=userfactory, multiple=True)
-        )
-        return cmd
+        return [
+            click.Option(['--gifted_to', '-g'], type=click.STRING, callback=userfactory, multiple=True),
+        ]

ovtk_audiencekit/events/__init__.py

@@ -7,4 +7,3 @@ from .Subscription import Subscription
 from .Follow import Follow
 
 __all__ = ['Event', 'Message', 'SysMessage', 'Delete', 'Control', 'Subscription', 'Follow']
-__path__ = __import__('pkgutil').extend_path(__path__, __name__)

ovtk_audiencekit/plugins/AudioAlert/AudioAlert.py

@@ -0,0 +1,27 @@
+import asyncio
+
+from ovtk_audiencekit.plugins import PluginBase
+from ovtk_audiencekit.core import Clip
+
+class AudioAlert(PluginBase):
+    def __init__(self, *args, output=None, buffer_length=2048, cutoff_prevention_buffers=None, **kwargs):
+        super().__init__(*args, **kwargs)
+        if cutoff_prevention_buffers:
+            self.logger.info('`cutoff_prevention_buffers` are depricated')
+
+        self.sounds = {}
+        self._buffer_length = int(buffer_length)
+        self._output_index = Clip.find_output_index(output)
+
+    def run(self, path, speed=1, immediate=True, **kwargs):
+        if self.sounds.get(path) is None:
+            self.sounds[path] = Clip(path,
+                                     self._output_index,
+                                     buffer_length=self._buffer_length,
+                                     speed=speed)
+        sound = self.sounds.get(path)
+
+        if immediate:
+            asyncio.create_task(sound.aplay())
+        else:
+            sound.play()

ovtk_audiencekit/plugins/Jail/JailPlugin.py

@@ -7,8 +7,8 @@ import maya
 from requests.exceptions import HTTPError
 from owoify.owoify import owoify, Owoness
 
-from ovtk_audiencekit.core import PluginBase
-from ovtk_audiencekit.core.Data import CACHE_DIR
+from ovtk_audiencekit.plugins import PluginBase
+from ovtk_audiencekit.core.Config import CACHE_DIR
 from ovtk_audiencekit.plugins.builtins.Command import Command, CommandTypes
 from ovtk_audiencekit.events.Message import Message, SysMessage
 
@@ -35,7 +35,8 @@ owomap = {
 }
 
 class JailPlugin(PluginBase):
-    def setup(self, min_level='vip', persist=True):
+    def __init__(self, *args, min_level='vip', persist=True, **kwargs):
+        super().__init__(*args, **kwargs)
         self.persist = persist
         self._cache = os.path.join(CACHE_DIR, 'Jail', 'sentences')
         os.makedirs(os.path.dirname(self._cache), exist_ok=True)
@@ -87,7 +88,7 @@ class JailPlugin(PluginBase):
         if isinstance(event, Message):
             if self.jail_command.invoked(event):
                 try:
-                    args, _ = self.jail_command.parse(event.text)
+                    args = self.jail_command.parse(event.text)
                     end_date = maya.when(args['length'], prefer_dates_from='future')
                     deets = self.chats[event.via].shared.api.get_user_details(args['username'])
                     if deets is None:
@@ -117,7 +118,7 @@ class JailPlugin(PluginBase):
                 self.send_to_bus(weewoo)
             elif self.unjail_command.invoked(event):
                 try:
-                    args, _ = self.jail_command.parse(event.text)
+                    args = self.jail_command.parse(event.text)
                     deets = self.chats[event.via].shared.api.get_user_details(args['username'])
                     if deets is None:
                         raise ValueError()

ovtk_audiencekit/plugins/OBS/obs.py

@@ -2,14 +2,18 @@ import asyncio
 
 import simpleobsws
 
-from ovtk_audiencekit.core import PluginBase
+from ovtk_audiencekit.plugins import PluginBase
 
 
 class OBSWSPlugin(PluginBase):
-    async def setup(self, password=None, uri='ws://localhost:4455'):
+    def __init__(self, *args, password=None, uri='ws://localhost:4455', **kwargs):
+        super().__init__(*args, **kwargs)
         self.uri = uri
 
         self.obsws = simpleobsws.WebSocketClient(url=uri, password=password)
+        asyncio.get_event_loop().run_until_complete(self.setup())
+
+    async def setup(self):
         await self.obsws.connect()
         success = await self.obsws.wait_until_identified()
         if not success:

ovtk_audiencekit/plugins/PhraseCounter/PhraseCounter.py

@@ -3,8 +3,8 @@ import logging
 import json
 import os
 
-from ovtk_audiencekit.core.Data import CACHE_DIR
-from ovtk_audiencekit.core import PluginBase
+from ovtk_audiencekit.core.Config import CACHE_DIR
+from ovtk_audiencekit.plugins import PluginBase
 from ovtk_audiencekit.events import Message
 
 from .Formatter import PhraseCountFormatter
@@ -64,7 +64,8 @@ class PhraseCounter:
 
 
 class PhraseCounterPlugin(PluginBase):
-    def setup(self, debounce_time=1, persist=False):
+    def __init__(self, *args, debounce_time=1, persist=False, **kwargs):
+        super().__init__(*args, **kwargs)
         self.debounce_time = debounce_time
         self.persist = persist
 
@@ -103,7 +104,7 @@ class PhraseCounterPlugin(PluginBase):
         return event
 
 
-    def run(self, *args, _children=None, _ctx={}, **kwargs):
+    def run(self, *args, _children=None, **kwargs):
         if len(_children) != 1:
             raise ValueError('Requires a template child')
         template = _children[0]

ovtk_audiencekit/plugins/PluginBase.py

@@ -0,0 +1,133 @@
+from dataclasses import asdict
+from abc import ABC, abstractmethod
+from functools import reduce
+from operator import getitem
+from string import Formatter
+import logging
+import asyncio
+
+import kdl
+from quart import Blueprint
+
+from ovtk_audiencekit.core.Config import kdl_parse_config
+
+
+class PluginError(Exception):
+    def __init__(self, source, message, fatal=True):
+        self.source = source
+        self.message = message
+        self.fatal = fatal
+
+    def __str__(self):
+        return self.message
+
+
+class GetitemFormatter(Formatter):
+    def get_field(self, field_name, args, kwargs):
+        keys = field_name.split('.')
+        field = reduce(getitem, keys, kwargs)
+        return (field, keys[0])
+
+
+class PluginBase(ABC):
+    plugins = {}
+
+    def __init__(self, chat_processes, event_queue, name, _children=None, **kwargs):
+        super().__init__(**kwargs)
+        self.chats = chat_processes
+        self._event_queue = event_queue
+        self._name = name
+
+        self.logger = logging.getLogger(f'plugin.{self._name}')
+        self.plugins[name] = self
+        self.blueprint = Blueprint(self._name, __name__)
+        if _children:
+            raise ValueError('Module does not accept children')
+
+    def __del__(self):
+        if self.plugins.get(self._name) == self:
+            del self.plugins[self._name]
+
+    # Base class helpers
+    def broadcast(self, event):
+        """Send event to every active chat"""
+        for proc in self.chats.values():
+            if proc.readonly:
+                continue
+            proc.control_pipe.send(event)
+
+    @staticmethod
+    def _kdl_arg_formatter(text, fragment, args):
+        key = fragment.fragment[1:-1]
+        try:
+            if '{' in key:
+                return GetitemFormatter().format(key, **args).replace(r'\"', '"')
+            else:
+                return reduce(getitem, key.split('.'), args)
+        except (KeyError, IndexError) as e:
+            raise ValueError(f'Invalid arg string - "{key}": {e}') from e
+
+    def fill_context(self, actionnode, ctx):
+        config = asdict(kdl_parse_config)
+        config['valueConverters'] = {
+            **config['valueConverters'],
+            'arg': lambda text, fragment, args=ctx: self._kdl_arg_formatter(text, fragment, args),
+        }
+        config = kdl.ParseConfig(**config)
+
+        newnode = kdl.parse(str(actionnode), config).nodes[0]
+        return newnode
+
+    async def call_plugin_from_kdl(self, node, *args, _ctx={}, **kwargs):
+        """
+        Calls some other plugin as configured by the passed KDL node
+        If this was done in response to an event, pass it as event in _ctx!
+        """
+        node = self.fill_context(node, _ctx)
+        target = self.plugins.get(node.name)
+        if target is None:
+            self.logger.warning(f'Could not find plugin or builtin with name {node.name}')
+        else:
+            return await target._run(*node.args, *args, **node.props, _ctx=_ctx, **kwargs, _children=node.nodes)
+
+    def send_to_bus(self, event):
+        """
+        Send an event to the event bus
+        WARNING: This will cause the event to be processed by other plugins - be careful not to cause an infinite loop!
+        """
+        self._event_queue.put_nowait(event)
+
+    async def _run(self, *args, **kwargs):
+        try:
+            res = self.run(*args, **kwargs)
+            if asyncio.iscoroutinefunction(self.run):
+                return await res
+            else:
+                return res
+        except Exception as e:
+            if isinstance(e, KeyboardInterrupt):
+                raise e
+            raise PluginError(self._name, str(e)) from e
+
+    # User-defined
+    async def tick(self, dt):
+        """Called at least every half second - perform time-dependent updates here!"""
+        pass
+
+    async def on_bus_event(self, event):
+        """Called for every event from the chats"""
+        return event
+
+    async def on_control_event(self, event):
+        """
+        Called for events targeting this plugin name specifically.
+        This is normally used for other applications to communicate with this one over the websocket interface
+        """
+        pass
+
+    @abstractmethod
+    async def run(self, _children=None, _ctx={}, **kwargs):
+        """
+        Run plugin action, either due to a definition in the config, or due to another plugin
+        """
+        pass

ovtk_audiencekit/plugins/Shoutout/ShoutoutPlugin.py

@@ -2,15 +2,17 @@ from argparse import ArgumentError
 
 from requests.exceptions import HTTPError
 
-from ovtk_audiencekit.core import PluginBase
+from ovtk_audiencekit.plugins import PluginBase
 from ovtk_audiencekit.plugins.builtins.Command import Command, CommandTypes
 from ovtk_audiencekit.events.Message import Message, SysMessage, USER_TYPE
 from ovtk_audiencekit.chats.Twitch import Process as Twitch
 
 
 class ShoutoutPlugin(PluginBase):
-    def setup(self, command='so', min_level='vip',
-                 text='Check out {link}!~ They were last streaming {last_game}'):
+    def __init__(self, *args, command='so', min_level='vip',
+                 text='Check out {link}!~ They were last streaming {last_game}',
+                 **kwargs):
+        super().__init__(*args, **kwargs)
         self.text = text
         if command:
             self.command = Command(name=command, help='Shoutout another user', required_level=min_level)
@@ -47,7 +49,7 @@ class ShoutoutPlugin(PluginBase):
         if isinstance(event, Message):
             if self.command and self.command.invoked(event):
                 try:
-                    args, _ = self.command.parse(event.text)
+                    args = self.command.parse(event.text)
                 except ArgumentError as e:
                     msg = SysMessage(self._name, str(e), replies_to=event)
                     self.chats[event.via].send(msg)

ovtk_audiencekit/plugins/TTS/TTS.py

@@ -0,0 +1,93 @@
+import uuid
+import os
+import asyncio
+
+from TTS.utils.synthesizer import Synthesizer
+from TTS.utils.manage import ModelManager
+from TTS.config import load_config
+
+from ovtk_audiencekit.plugins import PluginBase
+from ovtk_audiencekit.events import Message, SysMessage
+from ovtk_audiencekit.core import Clip
+from ovtk_audiencekit.core.Config import CACHE_DIR
+
+
+class TextToSpeechPlugin(PluginBase):
+    def __init__(self, *args, output=None, cuda=None,
+                 engine="tts_models/en/ljspeech/tacotron2-DDC", speaker_wav=None,
+                 _children=None, **kwargs):
+        super().__init__(*args, _children=_children)
+
+        self.speaker_wav = speaker_wav
+
+        self._output_index = Clip.find_output_index(output)
+
+        conf_overrides = {k[2:]: v for k, v in kwargs.items() if k.startswith('o_')}
+
+        self._cache = os.path.join(CACHE_DIR, 'tts')
+        os.makedirs(os.path.dirname(self._cache), exist_ok=True)
+
+        self.cuda = cuda
+
+        manager = ModelManager(output_prefix=CACHE_DIR) # HACK: coqui automatically adds 'tts' subdir
+        model_path, config_path, model_item = manager.download_model(engine)
+
+        if model_item["default_vocoder"]:
+            vocoder_path, vocoder_config_path, _ = manager.download_model(model_item["default_vocoder"])
+        else:
+            vocoder_path, vocoder_config_path = None, None
+
+        if conf_overrides:
+            override_conf_path = os.path.join(self._cache, f'{self._name}_override.json')
+
+            config = load_config(config_path)
+            for key, value in conf_overrides.items():
+                config[key] = value
+            config.save_json(override_conf_path)
+
+            config_path = override_conf_path
+
+        self.synthesizer = Synthesizer(
+            model_path,
+            config_path,
+            vocoder_checkpoint=vocoder_path,
+            vocoder_config=vocoder_config_path,
+            use_cuda=self.cuda,
+        )
+
+    def make_tts_wav(self, text, filename=None):
+        if filename is None:
+            filename = os.path.join(self._cache, f'{uuid.uuid1()}.wav')
+
+        if self.speaker_wav:
+            wav = self.synthesizer.tts(text, None, 'en', self.speaker_wav)
+        else:
+            wav = self.synthesizer.tts(text)
+
+        self.synthesizer.save_wav(wav, filename)
+        return filename
+
+    async def run(self, text, *args, _ctx={}, wait=True, **kwargs):
+        try:
+            # Force punctuation (keep AI from spinning off into random noises)
+            if not any([text.endswith(punc) for punc in '.!?:']):
+                text += '.'
+            filename = self.make_tts_wav(text)
+            # TODO: Play direct from memory
+            clip = Clip(filename, self._output_index, force_stereo=False)
+            if wait:
+                async def play():
+                    await clip.aplay()
+                    clip.close()
+                asyncio.create_task(play())
+            else:
+                clip.play()
+                clip.close()
+        except Exception as e:
+            print(e)
+            if source_event := _ctx.get('event'):
+                msg = SysMessage(self._name, 'Failed to make speech from input!!')
+
+                if isinstance(source_event, Message):
+                    msg.replies_to = source_event
+                self.chats[source_event.via].send(msg)

ovtk_audiencekit/plugins/__init__.py

@@ -0,0 +1,3 @@
+from .PluginBase import PluginBase, PluginError
+
+__all__ = ['PluginBase', 'PluginError']

ovtk_audiencekit/plugins/builtins/Chance.py

@@ -1,6 +1,6 @@
 import random
 
-from ovtk_audiencekit.core import PluginBase
+from ovtk_audiencekit.plugins import PluginBase
 
 
 class ChancePlugin(PluginBase):
@@ -16,7 +16,5 @@ class ChancePlugin(PluginBase):
             raise ValueError('Chance must be a string (optionally ending in %) or number')
 
         if random.random() < chance / 100:
-            await self.execute_kdl([child for child in _children if child.name != 'or'], _ctx=_ctx)
-        else:
-            if elsenode := next((child for child in _children if child.name == 'or'), None):
-                await self.execute_kdl(elsenode.nodes, _ctx=_ctx)
+            for node in _children:
+                await self.call_plugin_from_kdl(node, _ctx=_ctx)

ovtk_audiencekit/plugins/builtins/Choice.py

@@ -1,7 +1,7 @@
 import subprocess
 import random
 
-from ovtk_audiencekit.core import PluginBase
+from ovtk_audiencekit.plugins import PluginBase
 from ovtk_audiencekit.events import SysMessage
 
 
@@ -12,4 +12,4 @@ class ChoicePlugin(PluginBase):
 
     async def run(self, _children=None, _ctx={}, **kwargs):
         node = random.choice(_children)
-        await self.execute_kdl([node], _ctx=_ctx)
+        await self.call_plugin_from_kdl(node, _ctx=_ctx)

ovtk_audiencekit/plugins/builtins/Command.py

@@ -7,7 +7,7 @@ import sys
 
 from multipledispatch import dispatch
 
-from ovtk_audiencekit.core import PluginBase
+from ovtk_audiencekit.plugins import PluginBase
 from ovtk_audiencekit.events import Message, SysMessage
 from ovtk_audiencekit.events.Message import USER_TYPE
 
@@ -87,7 +87,7 @@ class Command:
         args = emoteless.split()[1:]
         parsed, unknown = self._parser.parse_known_args(args)
         parsed_asdict = vars(parsed)
-        return parsed_asdict, unknown
+        return parsed_asdict
 
 
 class CommandPlugin(PluginBase):
@@ -127,11 +127,11 @@ class CommandPlugin(PluginBase):
                     continue
                 if command.invoked(event):
                     try:
-                        args, unknown = command.parse(event.text)
+                        args = command.parse(event.text)
                         self.logger.debug(f"Parsed args for {command.name}: {args}")
-                        self.logger.debug(f"Remaining text: {unknown}")
-                        ctx = dict(event=event, rest=' '.join(unknown), **args)
-                        await self.execute_kdl(actionnode.nodes, _ctx=ctx)
+                        ctx = dict(event=event, **args)
+                        for node in actionnode.nodes:
+                            await self.call_plugin_from_kdl(node, _ctx=ctx)
                     except argparse.ArgumentError as e:
                         msg = SysMessage(self._name, f"{e}. See !help {command.name}", replies_to=event)
                         self.chats[event.via].send(msg)
@@ -139,7 +139,7 @@ class CommandPlugin(PluginBase):
 
             if self.help_cmd.invoked(event):
                 try:
-                    args, _ = self.help_cmd.parse(event.text)
+                    args = self.help_cmd.parse(event.text)
                 except argparse.ArgumentError as e:
                     msg = SysMessage(self._name, f"{e}. See !help {self.help_cmd.name}", replies_to=event)
                     self.chats[event.via].send(msg)

ovtk_audiencekit/plugins/builtins/Cue.py

@@ -0,0 +1,51 @@
+import maya
+
+from ovtk_audiencekit.plugins import PluginBase
+
+
+class CueEvent:
+    def __init__(self, oneshot, at=None, **kwargs):
+        self.oneshot = oneshot
+
+        if at:
+            self._next_run = maya.parse(at)
+            self._interval = None
+        else:
+            self._next_run = maya.now().add(**kwargs)
+            self._interval = kwargs
+
+    def check(self):
+        if self._next_run <= maya.now():
+            if self._interval:
+                self._next_run = maya.now().add(**self._interval)
+            return True
+        return False
+
+
+class CuePlugin(PluginBase):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.cue_events = {}
+
+    def run(self, name=None, _children=None, **kwargs):
+        if not _children:
+            raise ValueError('Cue defined without any events')
+
+        if name is None:
+            name = f"cue-{len(self.cue_events.keys())}"
+
+        for eventnode in _children:
+            at = eventnode.args[0] if len(eventnode.args) == 1 else None
+            oneshot = eventnode.name in ['once', 'after']
+            cue_event = CueEvent(oneshot, at=at, **eventnode.props)
+
+            actions = [lambda node=node: self.call_plugin_from_kdl(node) for node in eventnode.nodes];
+            self.cue_events[name] = (cue_event, actions)
+
+    async def tick(self, dt):
+        for key, (event, actions) in list(self.cue_events.items()):
+            if event.check():
+                for action in actions:
+                    await action()
+                if event.oneshot:
+                    del self.cue_events[key]

ovtk_audiencekit/plugins/builtins/Exec.py

@@ -1,6 +1,6 @@
 import subprocess
 
-from ovtk_audiencekit.core import PluginBase
+from ovtk_audiencekit.plugins import PluginBase
 from ovtk_audiencekit.events import SysMessage
 
 
@@ -9,7 +9,7 @@ class ExecPlugin(PluginBase):
         super().__init__(*args, **kwargs)
         self.warned = False
 
-    def run(self, cmd, reply=False, to_arg=None, _ctx={}, **kwargs):
+    def run(self, cmd, reply=False, _ctx={}, **kwargs):
         if not self.warned:
             self.logger.warning('Executing unchecked input is potentially dangerous! Check your (arg) inputs, if any, *very* carefully')
             self.warned = True
@@ -17,13 +17,11 @@ class ExecPlugin(PluginBase):
         out = subprocess.run(cmd.split(' '), capture_output=True, text=True)
 
         if out.returncode != 0:
-            self.logger.warning(f'Command returned {out.returncode}: {out.stderr}')
+            self.logger.error(f'Command retruned {out.returncode}: {out.stderr}')
             return
         else:
-            self.logger.debug(f'Command returned {out.returncode}: {out.stdout}')
+            self.logger.info(f'Command retruned {out.returncode}: {out.stdout}')
 
         if reply and _ctx['event']:
             msg = SysMessage(self._name, out.stdout)
             self.chats[_ctx['event'].via].send(msg)
-        if to_arg:
-            _ctx[to_arg] = out.stdout

ovtk_audiencekit/plugins/builtins/Midi.py

@@ -0,0 +1,17 @@
+import mido
+
+from ovtk_audiencekit.plugins import PluginBase
+
+
+class MidiPlugin(PluginBase):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.output_port = mido.open_output()
+
+    def run(self, type, _ctx={}, **kwargs):
+        if type == 'sysex':
+            data = kwargs['data']
+            msg = mido.Message('sysex', data=bytes(data, encoding='utf-8'), time=0)
+            self.output_port.send(msg)
+        else:
+            raise NotImplementedError('TODO: note on/off and cc')

ovtk_audiencekit/plugins/builtins/Remember.py

@@ -0,0 +1,24 @@
+from ovtk_audiencekit.plugins import PluginBase
+
+
+class RememberPlugin(PluginBase):
+    """Saves the response of other plugins in the local context (can be fetched with the custom arg type)"""
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+    async def run(self, name, _children=None, _ctx={}, **kwargs):
+        if _children is None:
+            self.logger.warn('No children - this does nothing!')
+            return
+
+        responses = []
+
+        for child in _children:
+            res = await self.call_plugin_from_kdl(child, _ctx=_ctx)
+            responses.append(res)
+
+        if len(responses) == 1:
+            responses = responses[0]
+
+        _ctx[name] = responses

ovtk_audiencekit/plugins/builtins/Reply.py

@@ -1,4 +1,4 @@
-from ovtk_audiencekit.core import PluginBase
+from ovtk_audiencekit.plugins import PluginBase
 from ovtk_audiencekit.events import SysMessage, Message
 
 

ovtk_audiencekit/plugins/builtins/Trigger.py

@@ -2,7 +2,7 @@ import re
 from dataclasses import dataclass
 import typing
 
-from ovtk_audiencekit.core import PluginBase
+from ovtk_audiencekit.plugins import PluginBase
 from ovtk_audiencekit.events import Message
 
 
@@ -108,15 +108,16 @@ class TriggerPlugin(PluginBase):
                 unknown_args[key] = value
         trigger = Trigger(**args, attr_checks=unknown_args)
 
-        handler = lambda ctx: self.execute_kdl(_children, _ctx=ctx)
+        actions = [lambda ctx, node=node: self.call_plugin_from_kdl(node, _ctx=ctx) for node in _children]
 
-        self.triggers.append((trigger, handler, _ctx))
+        self.triggers.append((trigger, actions, _ctx))
 
     async def on_bus_event(self, event):
-        for trigger, handler, ctx in self.triggers:
+        for trigger, actions, ctx in self.triggers:
             if trigger.matches(event, self.last_msg):
                 _ctx = {**ctx, 'event': event}
-                await handler(_ctx)
+                for action in actions:
+                    await action(_ctx)
         if isinstance(event, Message):
             self.last_msg = event
         return event

ovtk_audiencekit/plugins/builtins/WebSocket.py

@@ -0,0 +1,19 @@
+import subprocess
+
+import websockets
+
+from ovtk_audiencekit.plugins import PluginBase
+from ovtk_audiencekit.events import SysMessage
+from ovtk_audiencekit.utils import make_sync
+
+@make_sync
+async def send(ws, data):
+    async with websockets.connect(ws) as websocket:
+        await websocket.send(data)
+
+class WebSocketPlugin(PluginBase):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+    def run(self, endpoint, data, _ctx={}, **kwargs):
+        send(endpoint, data)

ovtk_audiencekit/plugins/builtins/Write.py

@@ -1,6 +1,6 @@
 import os
 
-from ovtk_audiencekit.core import PluginBase
+from ovtk_audiencekit.plugins import PluginBase
 from ovtk_audiencekit.events import SysMessage, Message
 
 
@@ -10,8 +10,7 @@ class WritePlugin(PluginBase):
 
     def run(self, text, target, append=False, **kwargs):
         text += '\n'
-
-        base_name = os.path.dirname(target)
+        base_name = os.path.dirname(text)
         if base_name:
             os.makedirs(base_name, exist_ok=True)
 

ovtk_audiencekit/plugins/builtins/__init__.py

@@ -0,0 +1,13 @@
+from .Trigger import TriggerPlugin as trigger
+from .Reply import ReplyPlugin as reply
+from .Command import CommandPlugin as command
+from .Cue import CuePlugin as cue
+from .Write import WritePlugin as write
+from .Exec import ExecPlugin as exec
+from .Chance import ChancePlugin as chance
+from .Choice import ChoicePlugin as choice
+from .Midi import MidiPlugin as midi
+from .WebSocket import WebSocketPlugin as ws
+from .Remember import RememberPlugin as remember
+
+__all__ = ['trigger', 'reply', 'command', 'cue', 'write', 'exec', 'chance', 'choice', 'midi', 'ws', 'remember']

ovtk_audiencekit/utils/NonBlockingWebsocket.py

@@ -1,4 +1,5 @@
 from multiprocessing import Process, Pipe
+from traceback import format_exception
 import logging
 import asyncio
 
@@ -43,18 +44,18 @@ class NonBlockingWebsocket(Process):
             self._pipe.send(data)
 
     def run(self):
-        loop = asyncio.new_event_loop()
         # Setup ws client
-        loop.run_until_complete(self._setup())
+        asyncio.get_event_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)
+        asyncio.get_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())
+        asyncio.get_event_loop().create_task(self._send(pipe_ready))
+        asyncio.get_event_loop().create_task(self._read())
 
         # Keep the asyncio code running in this thread until explicitly stopped
         try:
-            loop.run_forever()
+            asyncio.get_event_loop().run_forever()
         except KeyboardInterrupt:
             return 0

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

pyproject.toml

@@ -1,60 +0,0 @@
-[project]
-name = "ovtk_audiencekit"
-version = "0.1.0"
-description = ""
-authors = [
-  {name = "Skeh", email = "im@skeh.site"},
-]
-dependencies = [
-    "click",
-    "kdl-py",
-    "quart==0.18.*",
-    "werkzeug==2.3.7",
-    "hypercorn",
-    "requests",
-    "websockets==11.0.3",
-    "aioprocessing",
-    "aioscheduler",
-    "pyaudio==0.2.*",
-    "librosa==0.8.*",
-    "pytsmod",
-    "numpy",
-    "multipledispatch",
-    "blessed",
-    "appdirs",
-    "maya",
-]
-requires-python = ">=3.10,<3.11"
-readme = "README.md"
-license = {text = "GPLv2"}
-
-[project.optional-dependencies]
-tts = [
-  "TTS==0.9.*",
-  "torch==1.13.*",
-]
-phrasecounter = ["num2words"]
-jail = ["owoify-py==2.*"]
-twitch = ["miniirc"]
-midi = [
-  "mido",
-  "python-rtmidi",
-]
-obs = ["simpleobsws"]
-osc = ["python-osc>=1.9.0"]
-yt-dlp = ["yt-dlp"]
-
-[build-system]
-requires = ["pdm-backend"]
-build-backend = "pdm.backend"
-
-[tool.pdm.scripts]
-start = "python audiencekit.py start"
-ws = "python audiencekit.py ws"
-debug = "python audiencekit.py --debug"
-
-[tool.pdm]
-[[tool.pdm.source]]
-url = "https://pypi.org/simple"
-verify_ssl = true
-name = "pypi"

src/ovtk_audiencekit/__init__.py

@@ -1 +0,0 @@
-__path__ = __import__('pkgutil').extend_path(__path__, __name__)

src/ovtk_audiencekit/chats/YoutubeLive/YoutubeLive.py

@@ -1,183 +0,0 @@
-import os
-import json
-import webbrowser
-from enum import Enum, auto
-
-import requests
-
-from ovtk_audiencekit.chats import ChatProcess
-from ovtk_audiencekit.events.Message import Message, SysMessage, USER_TYPE
-
-
-class STATES(Enum):
-    WAITING_FOR_BROADCAST = auto()
-    POLLING = auto()
-    REFRESH = auto()
-    UNAUTHORIZED = auto()
-    FAILURE = auto()
-
-
-class YoutubeLivePollProcess(ChatProcess):
-    class CONTROL_MESSAGES(str, Enum):
-        RESTART = "restart"
-
-    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):
-        if client_secrets_path is None or not os.path.exists(client_secrets_path):
-            raise ValueError('Missing client secrets')
-
-        with open(client_secrets_path, 'r') as f:
-            client_secrets = json.load(f).get('installed')
-            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)
-
-        self._consent_code = None
-        self._refresh_token = None
-        self._access_token = None
-        self._stream_title = None
-        self._live_chat_id = None
-        self._page_token = None
-
-    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)
-
-    def on_unauthorized(self, next_state):
-        self.request_oauth_consent()
-        response = self.safe_input('Allow access in your browser and paste response code here: ')
-        self.setup_oauth_consent(response['token'])
-        return next_state
-
-    def on_failure(self, next_state):
-        pass
-
-    def on_refresh(self, next_state):
-        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}'})
-
-        if (response.status_code == 401):
-            return STATES.REFRESH, self.state
-
-        items = response.json()['items']
-        if len(items) == 1:
-            stream_snippet = items[0]['snippet']
-            self._live_chat_id = stream_snippet['liveChatId']
-            self._stream_title = stream_snippet['title']
-            return STATES.POLLING
-
-        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}'}
-                                )
-
-        if (response.status_code == 401):
-            return STATES.REFRESH, self.state
-
-        if response.status_code == requests.codes.ok:
-            data = response.json()
-            self._page_token = data['nextPageToken']
-            if len(data['items']):
-                for message in data['items']:
-                    normalized = self.normalize_message(message)
-                    self.publish(normalized)
-
-            sleep_milis = max(data['pollingIntervalMillis'], 5000)
-            return sleep_milis / 1000
-        else:
-            print(response.json())
-            return STATES.FAILURE
-
-    def on_state_enter(self, new_state):
-        status_messages = {
-            STATES.WAITING_FOR_BROADCAST: 'Waiting for active broadcast...',
-            STATES.POLLING: f'''Tuning into "{self._stream_title}" - ready to rumble ~\n''',
-            STATES.UNAUTHORIZED: 'Unauthorized - see terminal',
-            STATES.FAILURE: 'YouTube API returned a bad status, polling stopped - see terminal',
-        }
-
-        message = status_messages.get(new_state)
-        if message is not None:
-            sys_msg = SysMessage(self._name, message)
-            self.publish(sys_msg)
-
-    def process_messages(self, message_type, args, next_state):
-        if message_type == self.__class__.CONTROL_MESSAGES.RESTART:
-            if self.state == STATES.POLLING:
-                self.publish(SysMessage(self._name, 'Restart requested, but service is in healthy state! Ignoring...'))
-            else:
-                self.publish(SysMessage(self._name, 'Restarting service...'))
-                return None, None
-
-    def normalize_message(self, message):
-        if message['authorDetails']['isChatOwner']:
-            author_type = USER_TYPE.OWNER
-        elif message['authorDetails']['isChatModerator']:
-            author_type = USER_TYPE.MODERATOR
-        elif message['authorDetails']['isChatSponsor']:
-            author_type = USER_TYPE.PATRON
-        else:
-            author_type = USER_TYPE.USER
-
-        text = message['snippet']['displayMessage']
-        author_name = message['authorDetails']['displayName']
-        author_id = message['authorDetails']['channelId']
-        return Message(self._name, text, author_name, author_id, author_type)
-
-    def request_oauth_consent(self):
-        params = {
-            'client_id': self._client_secrets['client_id'],
-            'redirect_uri': 'urn:ietf:wg:oauth:2.0:oob',
-            'scope': 'https://www.googleapis.com/auth/youtube',
-            'response_type': 'code',
-        }
-        param_str = '&'.join(f'{k}={v}' for (k, v) in params.items())
-        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={
-            'code': consent_code,
-            'client_id': self._client_secrets['client_id'],
-            'client_secret': self._client_secrets['client_secret'],
-            'redirect_uri': 'urn:ietf:wg:oauth:2.0:oob',
-            'grant_type': 'authorization_code',
-        })
-        response.raise_for_status()
-        auth = response.json()
-        with open('refresh_token.secret', 'w') as f:
-            f.write(auth['refresh_token'])
-        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={
-            'client_id': self._client_secrets['client_id'],
-            'client_secret': self._client_secrets['client_secret'],
-            'refresh_token': self._refresh_token,
-            'grant_type': 'refresh_token'
-        })
-        response.raise_for_status()
-        auth = response.json()
-        self._access_token = auth['access_token']

src/ovtk_audiencekit/cli/start.py

@@ -1,31 +0,0 @@
-import logging
-import asyncio
-
-import click
-
-from ovtk_audiencekit.core import MainProcess
-from .group import cli
-
-logger = logging.getLogger(__name__)
-
-
-@cli.command()
-@click.argument('config_file', type=click.Path('r'), default='config.kdl')
-@click.option('--bus-bind', default='localhost')
-@click.option('--bus-port', default='8080')
-@click.option('--web-bind', default='localhost')
-@click.option('--web-port', default='8000')
-@click.option('--parallel', default=5)
-def start(config_file, bus_bind=None, bus_port=None, web_bind=None, web_port=None, parallel=None):
-    """Start audiencekit server"""
-    logger.info('Hewwo!!')
-    main = MainProcess(config_file,
-                       bus_conf=(bus_bind, bus_port),
-                       web_conf=(web_bind, web_port),
-                       max_concurrent=parallel)
-    try:
-        asyncio.run(main.run())
-    except KeyboardInterrupt:
-        pass
-    finally:
-        logger.info('Suya~')

src/ovtk_audiencekit/core/Config.py

@@ -1,155 +0,0 @@
-from functools import reduce
-from operator import getitem
-from string import Formatter
-from dataclasses import dataclass, field
-from abc import ABC, abstractmethod
-import os
-
-import kdl
-import kdl.types
-
-from ovtk_audiencekit.utils import format_exception
-
-
-@dataclass
-class Dynamic(ABC):
-    source: str
-    parser: any
-
-    def to_kdl(self):
-        return self.source
-
-    @abstractmethod
-    def compute():
-        pass
-
-    def __repr__(self):
-        return str(self.source)
-
-def get(instance, key):
-        if isinstance(instance, dict):
-            return getitem(instance, key)
-        else:
-            try:
-                return getattr(instance, key)
-            except KeyError:
-                return getitem(instance, key)
-
-class Arg(Dynamic):
-    class GetitemFormatter(Formatter):
-        def convert_field(self, value, conversion):
-            if conversion == 'i':
-                return value.replace('\n', '')
-            else:
-                return super().convert_field(value, conversion)
-
-        def get_field(self, field_name, args, kwargs):
-            keys = field_name.split('.')
-            field = reduce(get, keys, kwargs)
-            return (field, keys[0])
-
-    def compute(self, *args, _ctx={}):
-        key = self.parser.fragment[1:-1]
-        try:
-            if '{' in key:
-                return Arg.GetitemFormatter().format(key, **_ctx).replace(r'\"', '"')
-            else:
-                return reduce(get, key.split('.'), _ctx)
-        except (KeyError, IndexError) as e:
-            raise self.parser.error(f'Invalid arg string: {e}') from e
-        except Exception as e:
-            raise self.parser.error(f'Exception raised during arg inject: {format_exception(e, traceback=False)}') from e
-
-
-class Eval(Dynamic):
-    def compute(self, *args, **kwargs):
-        contents = self.parser.fragment[1:-1]
-        try:
-            return eval(contents, kwargs)
-        except Exception as e:
-            raise self.parser.error(f'Exception raised during eval: {format_exception(e, traceback=False)}') from e
-
-
-def csv_parser(text, parser):
-    text = parser.fragment[1:-1]
-    return [field.strip() for field in text.split(',')]
-
-def semisv_parser(text, parser):
-    text = parser.fragment[1:-1]
-    return [field.strip() for field in text.split(';')]
-
-customValueParsers = {
-    'arg': Arg,
-    't': Arg,
-    'eval': Eval,
-    'list': csv_parser,
-    'csv': csv_parser,
-    'semisv': semisv_parser,
-}
-
-@dataclass
-class KdlscriptNode(kdl.Node):
-    alias: str | None = field(default=None, init=False)
-    sub: str | None = field(default=None, init=False)
-
-    def __post_init__(self):
-        lhs, *sub = self.name.split('.')
-        if len(sub) == 0:
-            sub = None
-        alias, *name = lhs.split(':')
-        if len(name) == 0:
-            name = alias
-            alias = None
-        elif len(name) == 1:
-            name = name[0]
-        else:
-            raise ValueError("Invalid node name")
-
-        self.name = name
-        self.alias = alias
-        self.sub = sub
-
-        self._args = self.args
-        self._props = self.props
-
-    def compute(self, *args, **kwargs):
-        args = []
-        for arg in self._args:
-            if isinstance(arg, Dynamic):
-                arg = arg.compute(*args, **kwargs)
-            args.append(arg)
-        props = {}
-        for key, prop in self._props.items():
-            if isinstance(prop, Dynamic):
-                prop = prop.compute(*args, **kwargs)
-            props[key] = prop
-
-        self.args = args
-        self.props = props
-
-
-# HACK: Gross disgusting monkey patch
-kdl.types.Node = KdlscriptNode
-
-kdl_parse_config = kdl.ParseConfig(valueConverters=customValueParsers)
-kdl_reserved = ['secrets', 'chat', 'plugin', 'import']
-
-
-def parse_kdl_deep(path, relativeto=None):
-    if relativeto:
-        path = os.path.normpath(os.path.join(relativeto, path))
-
-    with open(path, 'r') as f:
-        try:
-            config = kdl.parse(f.read(), kdl_parse_config)
-            for node in config.nodes:
-                node.compute()
-        except kdl.errors.ParseError as e:
-            e.file = path
-            raise e
-
-    for node in config.nodes:
-        if node.name == 'import':
-            yield from parse_kdl_deep(node.args[0], relativeto=os.path.dirname(path))
-        else:
-            yield node

src/ovtk_audiencekit/core/MainProcess.py

@@ -1,344 +0,0 @@
-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
-
-import kdl
-from click import progressbar
-from aioscheduler import TimedScheduler
-from quart import Quart
-import hypercorn
-import hypercorn.asyncio
-import hypercorn.logging
-
-from ovtk_audiencekit.core.Config import parse_kdl_deep, kdl_reserved
-from ovtk_audiencekit.core.Plugins import PluginError
-from ovtk_audiencekit.core.WebsocketServerProcess import WebsocketServerProcess
-from ovtk_audiencekit.events import Event, Delete
-from ovtk_audiencekit.chats.ChatProcess import ShutdownRequest
-from ovtk_audiencekit.plugins import builtins
-from ovtk_audiencekit.utils import format_exception
-
-logger = logging.getLogger(__name__)
-
-class HypercornLoggingShim(hypercorn.logging.Logger):
-    """Force bog-standard loggers for Hypercorn"""
-    def __init__(self, config):
-        self.access_logger = logging.getLogger('hypercorn.access')
-        self.error_logger = logging.getLogger('hypercorn.error')
-        self.access_log_format = config.access_log_format
-
-def import_or_reload_mod(module_name, default_package=None, external=False):
-    if external:
-        package = None
-        import_fragment = module_name
-    else:
-        package = default_package
-        import_fragment = f'.{module_name}'
-    fq_module = f'{package}{import_fragment}' if not external else import_fragment
-
-    if module := sys.modules.get(fq_module):
-        for submod in [mod for name, mod in sys.modules.items() if name.startswith(fq_module)]:
-            importlib.reload(submod)
-        importlib.reload(module)
-    else:
-        module = importlib.import_module(import_fragment, package=package)
-    return module
-
-
-class MainProcess:
-    def __init__(self, config_path, bus_conf, web_conf, max_concurrent=5):
-        self._running = False
-        self.config_path = config_path
-        self.bus_conf = bus_conf
-        self.web_conf = web_conf
-        self.max_concurrent = max_concurrent
-
-        self.chat_processes = {}
-        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
-
-    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)
-        self.webserver.config['SERVER_NAME'] = listen
-        @self.webserver.context_processor
-        async def update_ctx():
-            return { 'EVBUS': 'ws://' + ':'.join(self.bus_conf) }
-        self.webserver.jinja_options = {
-            'block_start_string': '<%', 'block_end_string': '%>',
-            'variable_start_string': '<<', 'variable_end_string': '>>',
-            'comment_start_string': '<#', 'comment_end_string': '#>',
-        }
-
-        async def serve_coro():
-            config = hypercorn.config.Config()
-            config.bind = listen
-            config.use_reloader = False
-            config.logger_class = HypercornLoggingShim
-            try:
-                await hypercorn.asyncio.serve(self.webserver, config, shutdown_trigger=self.shutdown_ev.wait)
-            except Exception as e:
-                logger.critical(f'Failure in web process - {e}')
-                logger.debug(format_exception(e))
-                raise e
-            # MAGIC: As root tasks are supposed to be infinte loops, raise an exception if hypercorn shut down
-            raise asyncio.CancelledError()
-
-        return serve_coro()
-
-    async def handle_events(self):
-        while True:
-            event = await self.event_queue.get()
-            logger.info(event)
-
-            if isinstance(event, Event):
-                for plugin_name, plugin in list(self.plugins.items()):
-                    try:
-                        event = plugin.on_bus_event(event)
-                        if asyncio.iscoroutinefunction(plugin.on_bus_event):
-                            event = await event
-                    except PluginError as e:
-                        if e.fatal:
-                            logger.critical(f'Failure when processing {e.source} ({e}) - disabling...')
-                        else:
-                            logger.warning(f'Encounterd error when processing {e.source} ({e})')
-                        logger.debug(format_exception(e))
-                        if e.fatal:
-                            self._unload_plugin(e.source)
-                    except Exception as e:
-                        logger.critical(f'Failure when processing {plugin_name} ({e}) - disabling...')
-                        logger.debug(format_exception(e))
-                        self._unload_plugin(plugin_name)
-                    if event is None:
-                        break
-                else:
-                    self.server_process.message_pipe.send(event)
-                logger.debug(f'Event after plugin chain - {event}')
-            elif isinstance(event, Delete):
-                self.server_process.message_pipe.send(event)
-            else:
-                logger.error(f'Unknown data in event loop - {event}')
-
-    async def tick_plugins(self):
-        while True:
-            await asyncio.sleep(0.5)
-            for plugin_name, plugin in list(self.plugins.items()):
-                try:
-                    res = await plugin._tick(0.5) # Not necesarily honest!
-                except Exception as e:
-                    logger.critical(f'Failure during background processing for {plugin_name} ({e}) - disabling...')
-                    logger.debug(format_exception(e))
-                    self._unload_plugin(plugin_name)
-
-    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'):
-            for module in node.nodes:
-                fields = secrets.get(module.name, {})
-                for node in module.nodes:
-                    fields[node.name] = node.args[0] if len(node.args) == 1 else node.args
-                secrets[module.name] = fields
-
-        # Dynamically import chats
-        with progressbar(list(config.getAll('chat')), label="Preparing modules (chats)", item_show_func=lambda i: i and i.args[0]) as bar:
-            for node in bar:
-                if len(node.args) == 0:
-                    continue
-                module_name = node.args[0]
-                chat_name = node.alias or module_name
-                if chat_name in self.plugins:
-                    raise ValueError(f'Definition "{chat_name}" already exists - rename using alias syntax: `alias:chat ...`')
-                secrets_for_mod = secrets.get(module_name, {})
-                try:
-                    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)
-                    self.chat_processes[chat_name] = chat_process
-                except Exception as e:
-                    raise ValueError(f'Failed to initalize {module_name} module "{chat_name}" - {e}')
-
-        if len(self.chat_processes.keys()) == 0:
-            logger.warning('No chats configured!')
-
-        # Start chat processes
-        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))
-
-        # Load plugins
-        global_ctx = {}
-        ## Builtins
-        for node_name in builtins.__all__:
-            plugin = builtins.__dict__[node_name](self.chat_processes, self.event_queue, node_name, global_ctx)
-            self.plugins[node_name] = plugin
-            self.webserver.register_blueprint(plugin.blueprint)
-
-        ## Dynamic
-        with progressbar(list(config.getAll('plugin')), label="Preparing modules (plugins)", item_show_func=lambda i: i and i.args[0]) as bar:
-            for node in bar:
-                if len(node.args) == 0:
-                    continue
-                module_name = node.args[0]
-                plugin_name = node.alias or module_name
-                if plugin_name in self.plugins:
-                    raise ValueError(f'Definition "{plugin_name}" already exists - rename using alias syntax: `alias:plugin ...`')
-                secrets_for_mod = secrets.get(module_name, {})
-                try:
-                    plugin_module = import_or_reload_mod(module_name,
-                                                         default_package='ovtk_audiencekit.plugins',
-                                                         external=False)
-                    plugin = plugin_module.Plugin(self.chat_processes, self.event_queue, plugin_name, global_ctx)
-                    self.plugins[plugin_name] = plugin
-                    await plugin._setup(*node.args[1:], **node.props, **secrets_for_mod)
-                    # Register UI with webserver
-                    self.webserver.register_blueprint(plugin.blueprint)
-                except Exception as e:
-                    raise ValueError(f'Failed to initalize {module_name} plugin "{plugin_name}" - {e}')
-
-        # Run plugin definitions
-        with progressbar(list(config.nodes), label=f"Executing {self.config_path}") as bar:
-            for node in bar:
-                if node.name in kdl_reserved:
-                    continue
-                plugin_name = node.name
-                plugin_module = self.plugins.get(plugin_name)
-                if plugin_module is None:
-                    logger.error(f'Unknown plugin: {node.name}')
-                else:
-                    await plugin_module._kdl_call(node, global_ctx)
-
-    async def user_shutdown(self):
-        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,
-            #       as the webserver will still have its quart blueprint attached
-            self._unload_plugin(plugin_name)
-        sys.path = self._initial_syspath
-
-    async def run(self):
-        self.shutdown_ev = asyncio.Event()
-        self.reload_ev = asyncio.Event()
-        loop = asyncio.get_event_loop()
-        user_tasks = set()
-
-        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()
-            ## UI server
-            serve_coro = self._setup_webserver()
-            self.webserver_task = loop.create_task(serve_coro)
-            logger.debug(f'Listening on: {":".join(self.web_conf)} (UI) and {":".join(self.bus_conf)} (event bus)')
-
-            # User (plugin / chat) mode (reloading allowed)
-            while True:
-                async with self.webserver.app_context():
-                    await self.user_setup()
-                    # Start plumbing tasks
-                    user_tasks.add(loop.create_task(self.tick_plugins()))
-                    for i in range(0, self.max_concurrent):
-                        user_tasks.add(loop.create_task(self.handle_events()))
-
-                    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)
-
-                    if reload_task in done:
-                        logger.warn('Reloading (some events may be missed!)')
-                        logger.debug('Teardown...')
-                        self.reload_ev.clear()
-                        # Shutdown plugins / chats
-                        await self.user_shutdown()
-                        # Stop event plumbing
-                        for task in user_tasks:
-                            task.cancel()
-                        user_tasks.clear()
-                        # HACK: Restart webserver to workaround quart's inability to remove blueprints
-                        # Stop
-                        await self.webserver.shutdown()
-                        self.shutdown_ev.set()
-                        try:
-                            await self.webserver_task
-                        except asyncio.CancelledError:
-                            self.shutdown_ev.clear()
-                        # Start
-                        logger.debug('Startup...')
-                        serve_coro = self._setup_webserver()
-                        self.webserver_task = loop.create_task(serve_coro)
-                    else:
-                        break
-        except KeyboardInterrupt:
-            pass
-        except kdl.errors.ParseError as e:
-            try:
-                logger.critical(f'Invalid configuration in {e.file}: line {e.line}, character {e.col} - {e.msg}')
-            except AttributeError:
-                logger.critical(f'Invalid configuration - {e.msg}')
-        except Exception as e:
-            logger.critical(f'Failure in core process - {e}')
-            logger.debug(format_exception(e))
-        finally:
-            logger.warn('Closing up shop...')
-            for task in user_tasks:
-                task.cancel()
-            await self.user_shutdown()
-            self.webserver_task.cancel()
-            self.server_process.terminate()

src/ovtk_audiencekit/core/Plugins.py

@@ -1,200 +0,0 @@
-from abc import ABC, abstractmethod
-import logging
-import asyncio
-import os.path
-import sys
-import copy
-
-import kdl
-import quart
-
-from ovtk_audiencekit.core.Config import kdl_parse_config
-from ovtk_audiencekit.utils import format_exception
-
-
-
-class PluginError(Exception):
-    def __init__(self, source, message, fatal=True):
-        self.source = source
-        self.message = message
-        self.fatal = fatal
-
-    def __str__(self):
-        return self.message
-
-class OvtkBlueprint(quart.Blueprint):
-    def url_for(self, endpoint, *args, **kwargs):
-        """url_for method that understands blueprint-relative names under non-request contexts"""
-        if endpoint.startswith('.'):
-            endpoint = self.name + endpoint
-        return quart.url_for(endpoint, *args, **kwargs)
-
-    def render(self, name, **kwargs):
-        """render_template that prefers the plugin-specific templates"""
-        full = self.template_folder / name
-        if os.path.exists(full):
-            template_string = None
-            with open(full, 'r') as template_file:
-                template_string = template_file.read()
-            return quart.render_template_string(template_string, **kwargs)
-        else:
-            return quart.render_template(name, **kwargs)
-
-
-class PluginBase(ABC):
-    plugins = {}
-    hooks = {} # the hookerrrrrrrr
-
-    def __init__(self, chat_processes, event_queue, name, global_ctx, _children=None, **kwargs):
-        super().__init__(**kwargs)
-        self.chats = chat_processes
-        self._event_queue = event_queue
-        self._name = name
-        self._global_ctx = global_ctx
-
-        self.plugins[name] = self
-
-        self.logger = logging.getLogger(f'plugin.{self._name}')
-
-        # HACK: This is kinda gross, and probably wont be true for frozen modules
-        plugin_dir = os.path.dirname(sys.modules[self.__class__.__module__].__file__)
-        self.blueprint = OvtkBlueprint(self._name, __name__,
-                                       url_prefix=f'/{self._name}',
-                                       static_url_path='static',
-                                       static_folder=os.path.join(plugin_dir, 'static'),
-                                       template_folder=os.path.join(plugin_dir, 'templates'))
-
-        if _children:
-            raise ValueError('Module does not accept children')
-
-    def __del__(self):
-        if self.plugins.get(self._name) == self:
-            del self.plugins[self._name]
-            if self._name in self.hooks:
-                del self.hooks[self._name]
-
-    async def _kdl_call(self, node, _ctx):
-        node.compute(_ctx=_ctx)
-        subroutine = node.sub
-
-        if subroutine:
-            func = self
-            for accessor in subroutine:
-                func = getattr(func, accessor)
-        else:
-            func = self.run
-
-        for hook in self.hooks.values():
-            try:
-                res = hook(self._name, node, _ctx)
-                if asyncio.iscoroutinefunction(hook):
-                    await res
-            except Exception as e:
-                self.logger.warning(f'Failed to run plugin hook: {e}')
-                self.logger.debug(format_exception(e))
-
-        try:
-            result = func(*node.args, _children=node.nodes, _ctx=_ctx, **node.props)
-            if asyncio.iscoroutinefunction(func):
-                result = await result
-        except Exception as e:
-            if isinstance(e, KeyboardInterrupt):
-                raise e
-            raise PluginError(self._name, str(e)) from e
-
-        if node.alias:
-            _ctx[node.alias] = result
-
-    async def _tick(self, *args, **kwargs):
-        try:
-            res = self.tick(*args, **kwargs)
-            if asyncio.iscoroutinefunction(self.tick):
-                return await res
-            else:
-                return res
-        except Exception as e:
-            if isinstance(e, KeyboardInterrupt):
-                raise e
-            raise PluginError(self._name, str(e)) from e
-
-    async def _setup(self, *args, **kwargs):
-        try:
-            res = self.setup(*args, **kwargs)
-            if asyncio.iscoroutinefunction(self.setup):
-                return await res
-            else:
-                return res
-        except Exception as e:
-            if isinstance(e, KeyboardInterrupt):
-                raise e
-            raise PluginError(self._name, str(e)) from e
-
-
-    # Base class helpers
-    def broadcast(self, event):
-        """Send event to every active chat"""
-        for proc in self.chats.values():
-            if proc.readonly:
-                continue
-            proc.control_pipe.send(event)
-
-    def register_hook(self, hook):
-        self.hooks[self._name] = hook
-
-    async def execute_kdl(self, nodes, _ctx={}):
-        """
-        Run other plugins as configured by the passed KDL nodes collection
-        If this was done in response to an event, pass it as 'event' in _ctx!
-        """
-        _ctx = copy.deepcopy({**self._global_ctx, **_ctx})
-        for node in nodes:
-            try:
-                target = self.plugins.get(node.name)
-                if target is None:
-                    self.logger.warning(f'Could not find plugin or builtin with name {node.name}')
-                    break
-                await target._kdl_call(node, _ctx)
-            except Exception as e:
-                self.logger.warning(f'Failed to execute defered KDL: {e}')
-                self.logger.debug(format_exception(e))
-                break
-
-
-    def send_to_bus(self, event):
-        """
-        Send an event to the event bus
-        WARNING: This will cause the event to be processed by other plugins - be careful not to cause an infinite loop!
-        """
-        self._event_queue.put_nowait(event)
-
-
-    # User-defined
-    async def setup(self, *args, **kwargs):
-        """Called when plugin is being loaded."""
-        pass
-
-    def close(self):
-        """Called when plugin is about to be unloaded. Use this to safely close any resouces if needed"""
-        pass
-
-    async def tick(self, dt):
-        """Called at least every half second - perform time-dependent updates here!"""
-        pass
-
-    async def on_bus_event(self, event):
-        """Called for every event from the chats"""
-        return event
-
-    async def on_control_event(self, event):
-        """
-        Called for events targeting this plugin name specifically.
-        This is normally used for other applications to communicate with this one over the websocket interface
-        """
-        pass
-
-    @abstractmethod
-    async def run(self, *args, _children=None, _ctx={}, **kwargs):
-        """
-        Run plugin action, either due to a definition in the config, or due to another plugin
-        """
-        pass

src/ovtk_audiencekit/events/Control.py

@@ -1,47 +0,0 @@
-from dataclasses import dataclass, field
-import json
-
-import click
-
-from ovtk_audiencekit.events import Event
-
-
-@dataclass
-class Control(Event):
-    """Generic inter-bus communication"""
-    target: str = None
-    data: dict = field(default_factory=dict)
-
-    def __init__(self, via, target, **kwargs):
-        super().__init__(via)
-        self.target = target
-        self.data = kwargs
-
-    def __repr__(self):
-        return f"Control message : target = {self.target}, data = {self.data}"
-
-    def serialize(self):
-        return json.dumps({
-            'type': [cls.__name__ for cls in self.__class__.__mro__],
-            'data': {**self.data, **self.to_dict()},
-        })
-
-    @classmethod
-    def _cli(cls, cmd):
-        def parse(ctx, param, value):
-            if value:
-                return json.loads(value)
-
-        super_cb = cmd.callback
-
-        @click.pass_context
-        def expand(ctx, *args, json=None, **kwargs):
-            if json:
-                kwargs = {**json, **kwargs}
-            return ctx.invoke(super_cb, *args, **kwargs)
-
-        cmd.params.append(
-            click.Option(['--json', '-d'], callback=parse, help="Add arbitrary data in JSON format")
-        )
-        cmd.callback = expand
-        return cmd

src/ovtk_audiencekit/plugins/AudioAlert/AudioAlert.py

@@ -1,82 +0,0 @@
-import asyncio
-from collections import deque
-
-import maya
-
-from ovtk_audiencekit.core import PluginBase
-from ovtk_audiencekit.core import Clip, Stream
-
-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
-        self.timeout_min = timeout_min
-        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)
-        key = f'{path}@{speed}{"X" if keep_pitch else "x"}'
-        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)
-            self.clips[key] = [clip, maya.now()]
-        else:
-            self.clips[key][1] = maya.now()
-
-        stream_dq = self.streams.get(key, None)
-        if stream_dq is None:
-            stream_dq = deque(maxlen=poly)
-            self.streams[key] = stream_dq
-
-        if stream_dq.maxlen != poly:
-            self.logger.warn('Cannot change poly while streams are active!')
-
-        if len(stream_dq) == stream_dq.maxlen:
-            stream_dq.rotate(1)
-            stream = stream_dq[0]
-        else:
-            stream = Stream(clip, self.output_index,
-                            buffer_length=self.buffer_length)
-            stream_dq.append(stream)
-
-
-        if wait:
-            await stream.aplay()
-        else:
-            task = asyncio.create_task(stream.aplay())
-            task.add_done_callback(self.tasks.discard)
-            self.tasks.add(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()
-
-    async def _cleanup(self):
-        while True:
-            await asyncio.sleep(60)
-            now = maya.now()
-            for key, [clip, last_used] in list(self.clips.items()):
-                if now >= last_used.add(minutes=self.timeout_min, seconds=clip.length):
-                    self.logger.debug(f'Dropping {key}')
-                    streams = self.streams.get(key, [])
-                    for stream in streams:
-                        stream.close()
-                    del self.streams[key]
-                    del self.clips[key]

src/ovtk_audiencekit/plugins/Midi/Midi.py

@@ -1,60 +0,0 @@
-import asyncio
-
-import mido
-
-from ovtk_audiencekit.core import PluginBase
-
-
-def matches(msg, attrs):
-    for attr, match_val in attrs.items():
-        msg_val = getattr(msg, attr)
-        if msg_val != match_val:
-            return False
-    return True
-
-
-class MidiPlugin(PluginBase):
-    def setup(self):
-        loop = asyncio.get_event_loop()
-        def callback(msg):
-            asyncio.run_coroutine_threadsafe(self.recv_callback(msg), loop)
-
-        self.output_port = mido.open_output()
-        self.input_port = mido.open_input(callback=callback)
-        self.listeners = {
-            'note_off': [],
-            'note_on': [],
-            'control_change': [],
-            'program_change': [],
-            'sysex': [],
-            'song_select': [],
-            'start': [],
-            'continue': [],
-            'stop': [],
-        }
-
-    def close(self):
-        self.input_port.close()
-        self.output_port.close()
-
-    def run(self, type, _ctx={}, _children=None, **kwargs):
-        if type == 'sysex':
-            data = kwargs['data']
-            msg = mido.Message('sysex', data=bytes(data, encoding='utf-8'), time=0)
-        else:
-            msg = mido.Message(type, **kwargs, time=0)
-        self.output_port.send(msg)
-
-    async def recv_callback(self, msg):
-        if hasattr(msg, 'channel'):
-            msg.channel += 1 # Channels in mido are 0-15, but in spec are 1-16. Adjust to spec
-        self.logger.debug(f"Recv: {msg}")
-        for params, handler, ctx in self.listeners[msg.type]:
-            if matches(msg, params):
-                _ctx = {**ctx, 'midi': msg}
-                await handler(_ctx)
-
-    def listen(self, type, _ctx={}, _children=None, **kwargs):
-        kwargs = {k:int(v) for k, v in kwargs.items()}
-        handler = lambda ctx: self.execute_kdl(_children, _ctx=ctx)
-        self.listeners[type].append((kwargs, handler, _ctx))

src/ovtk_audiencekit/plugins/Midi/__init__.py

@@ -1 +0,0 @@
-from .Midi import MidiPlugin as Plugin

src/ovtk_audiencekit/plugins/OSC/__init__.py

@@ -1 +0,0 @@
-from .osc import OSCPlugin as Plugin

src/ovtk_audiencekit/plugins/OSC/osc.py

@@ -1,16 +0,0 @@
-from pythonosc.udp_client import SimpleUDPClient
-
-from ovtk_audiencekit.core import PluginBase
-
-
-class OSCPlugin(PluginBase):
-    def setup(self, ip='localhost', port=None):
-        if port is None:
-            raise RuntimeError('A unique port must be specified')
-        self.client = SimpleUDPClient(ip, int(port))
-
-    async def run(self, endpoint, *data, _children=None, _ctx={}, **kwargs):
-        if len(data) == 1:
-            self.client.send_message(endpoint, data[0])
-        else:
-            self.client.send_message(endpoint, data)

src/ovtk_audiencekit/plugins/TTS/TTS.py

@@ -1,114 +0,0 @@
-import uuid
-import os
-import asyncio
-
-from TTS.utils.synthesizer import Synthesizer
-from TTS.utils.manage import ModelManager
-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.Data import CACHE_DIR
-
-
-class TextToSpeechPlugin(PluginBase):
-    def setup(self, output=None, cuda=None,
-                 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
-
-        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.cuda = cuda
-
-        manager = ModelManager(output_prefix=CACHE_DIR) # HACK: coqui automatically adds 'tts' subdir
-        model_path, config_path, model_item = manager.download_model(engine)
-
-        if model_item["default_vocoder"]:
-            vocoder_path, vocoder_config_path, _ = manager.download_model(model_item["default_vocoder"])
-        else:
-            vocoder_path, vocoder_config_path = None, None
-
-        if conf_overrides:
-            override_conf_path = os.path.join(self.cache_dir, f'{self._name}_override.json')
-
-            config = load_config(config_path)
-            for key, value in conf_overrides.items():
-                config[key] = value
-            config.save_json(override_conf_path)
-
-            config_path = override_conf_path
-
-        self.synthesizer = Synthesizer(
-            model_path,
-            config_path,
-            vocoder_checkpoint=vocoder_path,
-            vocoder_config=vocoder_config_path,
-            use_cuda=self.cuda,
-        )
-
-        self.tasks = set()
-
-    def close(self):
-        for task in self.tasks:
-            task.cancel()
-
-    def make_tts_wav(self, text, filename=None):
-        # 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)
-        else:
-            wav = self.synthesizer.tts(text)
-
-        self.synthesizer.save_wav(wav, filename)
-        self.logger.info(f'Done - saved as {filename}')
-        return filename
-
-    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)
-
-            # 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()
-                finally:
-                    stream.close()
-                    os.remove(os.path.join(self.cache_dir, filename))
-            task = asyncio.create_task(play())
-            self.tasks.add(task)
-            task.add_done_callback(self.tasks.discard)
-
-            if wait:
-                await task
-        except Exception as e:
-            self.logger.error(f"Failed to make speech from input: {e}")
-            if source_event := _ctx.get('event'):
-                msg = SysMessage(self._name, 'Failed to make speech from input!!')
-
-                if isinstance(source_event, Message):
-                    msg.replies_to = source_event
-                self.chats[source_event.via].send(msg)

src/ovtk_audiencekit/plugins/__init__.py

@@ -1 +0,0 @@
-__path__ = __import__('pkgutil').extend_path(__path__, __name__)

src/ovtk_audiencekit/plugins/builtins/Cue.py

@@ -1,128 +0,0 @@
-import asyncio
-import logging
-import datetime
-import uuid
-
-import maya
-import aioscheduler
-
-from ovtk_audiencekit.core import PluginBase
-from ovtk_audiencekit.utils import format_exception
-
-logger = logging.getLogger(__name__)
-
-TIME_SEGMENTS = ['seconds', 'minutes', 'hours', 'days']
-
-class Cue:
-    def __init__(self, repeat, at=None, **kwargs):
-        self.repeat = repeat
-        self.enabled = True
-
-        if at:
-            self._next = maya.when(at)
-        else:
-            self._next = maya.now().add(**kwargs)
-        self._interval = kwargs
-
-    @property
-    def next(self):
-        return self._next.datetime(to_timezone='UTC', naive=True)
-
-    def is_obsolete(self):
-        if self.repeat:
-            return False
-        if self._next >= maya.now():
-            return False
-        return True
-
-    def reschedule(self, fresh=False):
-        if fresh:
-            self._next = maya.now().add(**self._interval)
-        else:
-            self._next = self._next.add(**self._interval)
-            # HACK: Compare epochs directly, as maya comps are only second accurate
-            if not fresh and self._next._epoch <= maya.now()._epoch:
-                offset = maya.now()._epoch - self._next._epoch
-                logger.warn(f'Cannot keep up with configured interval - {underrun} underrun. Repetition may fail!')
-                self._next = maya.now().add(**self._interval)
-
-
-class CuePlugin(PluginBase):
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.cues = {}
-        self.tasks = {}
-
-        self.scheduler = aioscheduler.TimedScheduler()
-        self.scheduler.start()
-
-        self._cleanup_task = asyncio.create_task(self._cleanup())
-
-    def run(self, name=None, repeat=False, enabled=None, _children=[], _ctx={}, **kwargs):
-        if name is None:
-            name = str(uuid.uuid4())
-
-        first_set = self.cues.get(name) is None
-
-        if len(_children) > 0:
-            has_interval = any(kwargs.get(segment) is not None for segment in TIME_SEGMENTS)
-            if kwargs.get('at') is None and not has_interval:
-                raise ValueError('Provide a concrete time with `at` or a timer length with `seconds`, `hours`, etc')
-            if kwargs.get('at') is not None and repeat and not has_interval:
-                raise ValueError('`repeat` can not be used with solely a concrete time')
-
-            cue = Cue(repeat, **kwargs)
-
-            async def handler():
-                # Repetition management
-                if not cue.enabled:
-                    return
-                if cue.repeat:
-                    cue.reschedule()
-                    self.tasks[name] = self.scheduler.schedule(handler(), cue.next)
-                # Run configured actions
-                try:
-                    await self.execute_kdl(_children, _ctx=_ctx)
-                except Exception as e:
-                    self.logger.error(f'Failed to complete cue {name}: {e}')
-                    self.logger.debug(format_exception(e))
-
-            self.cues[name] = (cue, handler)
-            self.schedule_exec(name, cue.next, handler())
-
-        if enabled is not None:
-            entry = self.cues.get(name)
-            if entry is None:
-                self.logger.warn(f'Cannot find cue with name "{name}"')
-                return
-            cue, handler = entry
-
-            cue.enabled = enabled
-
-            if enabled and not first_set:
-                cue.reschedule(fresh=True)
-                self.schedule_exec(name, cue.next, handler())
-
-    def schedule_exec(self, name, at, coro):
-        if existing_task := self.tasks.get(name):
-            self.scheduler.cancel(existing_task)
-        try:
-            self.tasks[name] = self.scheduler.schedule(coro, at)
-        except ValueError as e:
-            self.logger.error(f'Cannot schedule cue {name} at {at}: {e}')
-
-    def close(self):
-        self._cleanup_task.cancel()
-        for task in self.tasks.values():
-            self.scheduler.cancel(task)
-        self.scheduler._task.cancel()
-
-    async def _cleanup(self):
-        while True:
-            await asyncio.sleep(60)
-            for name, (cue, _) in list(self.cues.items()):
-                if cue.is_obsolete():
-                    del self.cues[name]
-                    if task := self.tasks.get(name):
-                        self.scheduler.cancel(task)
-                        del self.tasks[name]

src/ovtk_audiencekit/plugins/builtins/Extend.py

@@ -1,32 +0,0 @@
-import os
-
-from ovtk_audiencekit.core import PluginBase
-from ovtk_audiencekit.events import SysMessage, Message
-
-
-def compute_recursive(node, _ctx):
-    node.compute(_ctx=_ctx)
-    for child in node.nodes:
-        compute_recursive(child, _ctx)
-
-class ExtendPlugin(PluginBase):
-    """Extend a KDL file with the contained node"""
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-
-    async def run(self, target, compute=False, run=False, _children=None, _ctx={}):
-        if _children is None:
-            return
-
-        if target:
-            base_name = os.path.dirname(target)
-            if base_name:
-                os.makedirs(base_name, exist_ok=True)
-
-            with open(target, 'a') as f:
-                for node in _children:
-                    if compute:
-                        compute_recursive(node, _ctx)
-                    f.write(str(node))
-        if run:
-            await self.execute_kdl(_children, _ctx=_ctx)

src/ovtk_audiencekit/plugins/builtins/Log.py

@@ -1,14 +0,0 @@
-from ovtk_audiencekit.core import PluginBase
-
-import logging
-
-level_names = ['critical', 'error', 'warning',  'info', 'debug']
-
-class LogPlugin(PluginBase):
-    def run(self, msg, level="info", **kwargs):
-        try:
-            int_level = next((getattr(logging, level_name.upper()) for level_name in level_names if level_name.startswith(level.lower())))
-        except StopIteration:
-            self.logger.debug(f'Using default log level for KDL log call since user level "{level}" is not recognized')
-            int_level = logging.INFO
-        self.logger.log(int_level, msg)

src/ovtk_audiencekit/plugins/builtins/Mkevent.py

@@ -1,53 +0,0 @@
-import dataclasses
-
-from enum import Enum
-
-from ovtk_audiencekit.core import PluginBase
-from ovtk_audiencekit.events import Event
-from ovtk_audiencekit.utils import get_subclasses
-
-
-class EvFactory:
-    def __init__(self, ev_class):
-        self.target_class = ev_class
-        self.arg_convs = []
-        self.kwarg_convs = {}
-        for field in dataclasses.fields(self.target_class):
-            required = all(isinstance(default, dataclasses.MISSING.__class__) for default in [field.default, field.default_factory])
-            if issubclass(field.type, Enum):
-                conv = lambda value: field.type.__members__[value]
-            else:
-                conv = None
-            if required:
-                self.arg_convs.append(conv)
-            else:
-                self.kwarg_convs[field.name] = conv
-
-    def make(self, via, *args, **kwargs):
-        args = [
-            conv(arg) if conv else arg
-            for arg, conv in zip(args, self.arg_convs)
-        ]
-        kwargs = {
-            key: self.kwarg_convs[key](value) if self.kwarg_convs.get(key) else value
-            for key, value in kwargs.items()
-        }
-        return self.target_class(via, *args, **kwargs)
-
-
-class MakeEventPkugin(PluginBase):
-    """Create a new event and send it to the event bus"""
-
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.event_factories = {}
-        for event_class in get_subclasses(Event):
-            event_name = event_class.__name__
-            self.event_factories[event_name] = EvFactory(event_class)
-
-    def run(self, event_name, *args, _children=[], _ctx={}, **kwargs):
-        event_factory = self.event_factories.get(event_name)
-        if event_factory is None:
-            raise ValueError(f'Unknown event type "{event_name}"')
-        ev = event_factory.make('kdl', *args, **kwargs)
-        self.send_to_bus(ev)

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

@@ -1,153 +0,0 @@
-from dataclasses import dataclass, field
-import asyncio
-from typing import Callable
-import quart
-import json
-
-import kdl
-
-from ovtk_audiencekit.core import PluginBase
-from ovtk_audiencekit.utils import format_exception
-
-
-@dataclass
-class Scene:
-    name: str
-    group: str
-    oneshot: bool
-    enter: Callable
-    exit: Callable
-    entry_context: dict = field(default_factory=dict)
-    tasks: list[asyncio.Task] = field(default_factory=list)
-
-
-class ScenePlugin(PluginBase):
-    """Allows for creating modal configurations"""
-
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-
-        self.scenes = {}
-        self.active = {}
-
-        self._scene_state_changed = asyncio.Event()
-
-        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)
-
-    async def run(self, name, _children=None, _ctx={}, active=None, group=None, oneshot=False, **kwargs):
-        if _children is None:
-            raise UsageError('Empty scene definition! Did you mean scene.set?')
-
-        await self.define(name, group, _children, default_active=active, oneshot=oneshot, ctx=_ctx)
-
-    async def set(self, name, _children=None, _ctx={}, active=True, wait=False):
-        await self.switch(name, active, is_immediate=not wait, ctx=_ctx)
-
-    async def define(self, name, group, children, default_active=False, oneshot=False, ctx={}):
-        if self.scenes.get(name) is not None:
-            raise UsageError(f'Scene with name "{name}" already exists!')
-
-        # Categorize nodes
-        enter_nodes = []
-        exit_nodes = []
-        for child in children:
-            if child.name == 'exit':
-                exit_nodes.extend(child.nodes)
-            else:
-                enter_nodes.append(child)
-        # Make transisition functions
-        async def enter(ctx):
-            await self.execute_kdl(enter_nodes, _ctx=ctx)
-            scene.entry_context = ctx
-        async def exit(ctx):
-            await self.execute_kdl(exit_nodes, _ctx=ctx)
-
-        scene = Scene(name, group, oneshot, enter, exit)
-        self.scenes[name] = scene
-
-        if default_active:
-            await self.switch(name, default_active, is_immediate=True, ctx=ctx)
-
-    async def switch(self, name, active, is_immediate=True, ctx={}):
-        scene = self.scenes.get(name)
-        if scene is None:
-            raise UsageError(f'No defined scene with name "{name}"')
-
-        if scene.oneshot:
-            await self._execute(scene, 'enter', is_immediate, ctx)
-        elif active:
-            if current := self.active.get(scene.group):
-                if current == scene:
-                    return
-                await self._execute(current, 'exit', is_immediate, ctx)
-            self.active[scene.group] = scene
-            await self._execute(scene, 'enter', is_immediate, ctx)
-        else:
-            if self.active.get(scene.group) == scene:
-                self.active[scene.group] = None
-                await self._execute(scene, 'exit', is_immediate, ctx)
-
-        self._scene_state_changed.set()
-        self._scene_state_changed.clear()
-
-    async def _execute(self, scene, mode, immediate, ctx):
-        ctx = {**ctx} # HACK: Copy to avoid leakage from previous group item exit
-
-        scene_transision_fn = getattr(scene, mode)
-
-        # Wrap to handle context at exec time
-        async def context_wrapper(ctx):
-            if mode == 'exit':
-                ctx = {
-                    **scene.entry_context,
-                    'caller_ctx': ctx,
-                }
-            await scene_transision_fn(ctx)
-            if mode == 'enter':
-                scene.entry_context = ctx
-        coro = context_wrapper(ctx)
-        # Wrap to finish any other pending tasks before running this
-        if len(scene.tasks):
-            async def exec_order_wrap(other):
-                await asyncio.gather(*scene.tasks)
-                scene.tasks = []
-                await other
-            coro = exec_order_wrap(coro)
-
-        # Run (or schedule for execution)
-        if immediate:
-            try:
-                await coro
-            except Exception as e:
-                self.logger.error(f'Failed to handle "{scene.name}" {mode} transistion: {e}')
-                self.logger.debug(format_exception(e))
-        else:
-            scene.tasks.append(asyncio.create_task(coro))
-
-    def _get_state(self):
-        groups = {}
-        for scene_name, scene in self.scenes.items():
-            active = self.active.get(scene.group) == scene
-            group = scene.group or "default group"
-            if groups.get(group) is None:
-                groups[group] = {}
-            groups[group][scene_name] = active
-        return groups
-
-    async def ui_ctrlpanel(self):
-        groups = self._get_state()
-        return await self.blueprint.render('index.html', init_state=json.dumps(groups))
-
-    async def ui_setscene(self, name=None, cmd=None):
-        active = cmd == 'activate'
-        await self.switch(name, active, is_immediate=True)
-        return quart.Response(status=200)
-
-    async def ui_monitor_ws(self):
-        await quart.websocket.accept()
-        while True:
-            groups = self._get_state()
-            await quart.websocket.send(json.dumps(groups))
-            await self._scene_state_changed.wait()

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

@@ -1 +0,0 @@
-from .Plugin import ScenePlugin, Scene

src/ovtk_audiencekit/plugins/builtins/Scene/templates/index.html

@@ -1,85 +0,0 @@
-<!DOCTYPE html>
-<html lang="en" dir="ltr">
-  <head>
-    <meta charset="utf-8">
-    <title>Scene control</title>
-    <script type="importmap">
-      {
-        "imports": { "vue": "https://unpkg.com/vue@3/dist/vue.esm-browser.js" }
-      }
-    </script>
-
-    <script type="module">
-      import { createApp, ref, onMounted } from 'vue'
-
-      createApp({
-        setup() {
-          const groups = ref(JSON.parse('<< init_state|safe >>'))
-          const inflight = ref([])
-          onMounted(() => {
-            const websock = new WebSocket('<<url_for(".monitor")>>');
-            websock.addEventListener('message', (msg) => {
-              groups.value = JSON.parse(msg.data)
-              inflight.value = []
-            })
-          })
-          const toggle = async (group_name, scene_name) => {
-            if (!groups.value[group_name][scene_name].oneshot) {
-              if (inflight.value.includes(scene_name)) return
-              inflight.value.push(scene_name)
-            }
-            const next_state = !groups.value[group_name][scene_name]
-            await fetch(`${scene_name}/${next_state ? 'activate' : 'deactivate'}`, { method: 'GET' })
-          }
-          return { groups, inflight, toggle }
-        },
-      }).mount('#app')
-    </script>
-  </head>
-
-  <body id="root">
-    <div id="app">
-      <div v-for="(group, group_name) in groups" class="group">
-        <h3>{{ group_name }}</h3>
-        <div v-for="(active, scene_name) in group" v-on:click="toggle(group_name, scene_name)"
-          :class="{ active, pending: inflight.includes(scene_name), scene: true }"
-        >
-          <p>{{ scene_name }}</p>
-        </div>
-      </div>
-    </div>
-
-    <style type="text/css">
-      #app {
-        display: flex;
-        flex-direction: row;
-        flex-wrap: wrap;
-        font-family: sans-serif;
-        gap: 8px;
-      }
-      p {
-        text-align: center;
-      }
-      h3 {
-        margin-right: 1em;
-      }
-      .group {
-        display: flex;
-        flex-direction: column;
-        gap: 4px;
-      }
-      .scene {
-        padding: 12px 24px;
-        user-select: none;
-        background-color: lightgray;
-        flex: 1;
-      }
-      .scene.pending {
-        background-color: lightgoldenrodyellow;
-      }
-      .scene.active {
-        background-color: lightgreen;
-      }
-    </style>
-  </body>
-</html>

src/ovtk_audiencekit/plugins/builtins/Set.py

@@ -1,39 +0,0 @@
-from ovtk_audiencekit.core import PluginBase
-
-
-class SetPlugin(PluginBase):
-    """Set arbitrary data in the local context (can be fetched with the custom arg type)"""
-
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-
-    def run(self, *args, _children=[], _ctx={}, **kwargs):
-        self.proc_node(_ctx, *args, _children=_children, _stack=[], **kwargs)
-        self.logger.debug(_ctx)
-
-    def proc_node(self, target, *args, _children=[], _stack=[], **props):
-        if len(args) > 0 and len(props) > 0:
-            raise ValueError("Cannot use both item/list and keyword forms at the same time")
-        if _children and (len(args) > 0 or len(props) > 0):
-            raise ValueError("Cannot define value as something and dict at the same time")
-        if len(args) > 0 and len(_stack) == 0:
-            raise ValueError("Cannot use item/list short form on top level set")
-
-        if len(props) > 0:
-            for key, value in props.items():
-                if target.get(key) is not None:
-                    fullkey = '.'.join([n for n, t in _stack] + [key])
-                    self.logger.debug(f'Shadowing {fullkey}')
-                target[key] = value
-        elif _children:
-            for child in _children:
-                sub = dict()
-                target[child.name] = sub
-                stack = [*_stack, (child.name, target)]
-                key = '.'.join(s for s, t in stack)
-
-                child.compute(_ctx=stack[0][1])
-                self.proc_node(sub, *child.args, _children=child.nodes, _stack=stack, **child.props)
-        elif len(args) > 0:
-            name, target = _stack[-1]
-            target[name] = args[0] if len(args) == 1 else args

src/ovtk_audiencekit/plugins/builtins/Wait.py

@@ -1,9 +0,0 @@
-from ovtk_audiencekit.core import PluginBase
-
-import asyncio
-
-
-class WaitPlugin(PluginBase):
-    """Halt execution for a bit"""
-    async def run(self, seconds=0, minutes=0, **kwargs): # If you want `hours`, why??????
-        await asyncio.sleep(seconds + (minutes * 60))

src/ovtk_audiencekit/plugins/builtins/__init__.py

@@ -1,19 +0,0 @@
-from .Trigger import TriggerPlugin as trigger
-from .Reply import ReplyPlugin as reply
-from .Command import CommandPlugin as command
-from .Cue import CuePlugin as cue
-from .Write import WritePlugin as write
-from .Exec import ExecPlugin as exec
-from .Chance import ChancePlugin as chance
-from .Choice import ChoicePlugin as choice
-from .Set import SetPlugin as set
-from .Scene import ScenePlugin as scene
-from .Log import LogPlugin as log
-from .Mkevent import MakeEventPkugin as mkevent
-from .Wait import WaitPlugin as wait
-from .Extend import ExtendPlugin as extend
-
-__all__ = [
-    'trigger', 'reply', 'command', 'cue', 'write', 'exec', 'chance', 'choice',
-    'set', 'scene', 'log', 'mkevent', 'wait', 'extend'
-]

src/ovtk_audiencekit/plugins/yt-dlp.py

@@ -1,86 +0,0 @@
-import asyncio
-import functools
-
-from yt_dlp import YoutubeDL, DownloadCancelled
-
-from ovtk_audiencekit.core import PluginBase
-
-
-filter_aliases = {
-    'views': 'view_count',
-    'likes': 'like_count',
-    'comments': 'comment_count',
-    'followers': 'channel_follower_count',
-    'subscribers': 'channel_follower_count', 'subs': 'channel_follower_count',
-    'length': 'duration',
-}
-
-
-class Plugin(PluginBase):
-    async def run(self, source_url, destination, audio_only=False, format=None,
-                  _children=None, _ctx={}, **kwargs):
-        filter_fn, fail_node = self.make_filter(_children) if _children else (None, None)
-
-        if format == 'ogg':
-            format = 'vorbis'
-
-        params = {
-            'fragment_retries': 10, 'retries': 10,
-            'noplaylist': True, 'noprogress': True,
-            'logger': self.logger,
-            'match_filter': filter_fn, 'break_on_reject': True,
-            'outtmpl': {'default': destination},
-            'final_ext': format,
-            'postprocessors': [{
-                'key': 'FFmpegExtractAudio', 'preferredcodec': format,
-            } if audio_only else {
-                'key': 'FFmpegVideoConvertor', 'preferedformat': format,
-            }],
-            **kwargs,
-        }
-
-        loop = asyncio.get_running_loop()
-        try:
-            with YoutubeDL(params) as ytdl:
-                info = await loop.run_in_executor(None, ytdl.extract_info, source_url)
-        except DownloadCancelled as e:
-            if fail_node:
-                await self.execute_kdl(fail_node.nodes, _ctx=_ctx)
-                if fail_node.props.get("continue"):
-                    return
-            raise e
-
-        filename = info['requested_downloads'][0]['filepath']
-        return filename
-
-    def make_filter(self, nodes):
-        fail_node = None
-        nodes = nodes.copy()
-        if index := next((index for index, node in enumerate(nodes) if node.name == 'on_fail'), None):
-            fail_node = nodes.pop(index)
-
-        def filter(_nodes, info, *args):
-            for node in _nodes:
-                field = node.name
-                if real_field := filter_aliases.get(field):
-                    field = real_field
-
-                value = info.get(field)
-                if value is None:
-                    if node.props.get('strict'):
-                        return f"Missing metadata: {field}"
-                    else:
-                        continue
-
-                if target := node.props.get('min'):
-                    if value < target:
-                        return f"Failed {field} check: min {target}, got {value}"
-                if target := node.props.get('max'):
-                    if value >= target:
-                        return f"Failed {field} check: max {target}, got {value}"
-                if len(node.args) == 1:
-                    target = node.args[0]
-                    if value != target:
-                        return f"Failed {field} check: wanted {target}, got {value}"
-
-        return functools.partial(filter, nodes), fail_node