.coveragerc

@@ -1,26 +0,0 @@
-# .coveragerc to control coverage.py
-[run]
-branch = True
-
-[report]
-# Regexes for lines to exclude from consideration
-exclude_lines =
-    # Have to re-enable the standard pragma
-    pragma: no cover
-
-    # Don't complain about missing debug-only code:
-    def __repr__
-    if self\.debug
-
-    # Don't complain if tests don't hit defensive assertion code:
-    raise AssertionError
-    raise NotImplementedError
-
-    # Don't complain if non-runnable code isn't run:
-    if 0:
-    if __name__ == .__main__.:
-
-    # Don't complain about abstract methods, they aren't run:
-    @(abc\.)?abstractmethod
-
-ignore_errors = True

README.md

@@ -1,37 +1,18 @@
 # Croaker
-
+A command-and-control web application for icecast / liquidaudio, with helpers for D&D session music.
-A shoutcast audio player designed for serving D&D session music.
-
-### Features
-
-* Native streaming of MP3 sources direct to your shoutcast / icecast server
-* Transcoding of anything your local `ffmpeg` installation can convert to mp3
-* Playlists are built using symlinks
-* Randomizes playlist order the first time it is cached
-* Always plays `_theme.mp3` first upon switching to a playlist, if it exists
-* Falls back to silence if the stream encounters an error
-
-### Requirements
-
-* A functioning shoutcast / icecast server
-* Python >= 3.10
-* ffmpeg
-* libshout3-dev
-
 
 ## What? Why?
 
-Because I run an online D&D game, which includes a background music stream for my players. The stream used to be served by liquidsoap and controlled by a bunch of bash scripts I cobbled together which are functional but brittle, and liquidsoap is a nightmare for the small use case. Also, this required me to have a terminal window open to my media server to control liquidsoap directly, and I'd rather integrate the music controls directly with the rest of my DM tools, all of which run on my laptop.
+Because I run an online D&D game, which includes a background music stream for my players. The stream is controlled by a bunch of bash scripts I cobbled together which are functional but brittle. Also, this currently requires me to have a terminal window open to my media server to control liquidsoap directly, and I'd rather integrate the music controls directly with the rest of my DM tools, all of which run on my laptop. A web-based commmand-and-control app lets me use vanilla HTTP requests to control liquidsoap.
 
 *Now that is a powerful yak! -- Aesop Rock (misquoted)*
 
 
 ## Quick Start (Server)
 
-This assumes you have a functioning icecast2/whatever installation already.
+This assumes you have a functioning icecast2 installation already.
 
 ```
-% sudo apt install libshout3-dev
 % mkdir -p ~/.dnd/croaker
 % croaker setup > ~/.dnd/croaker/defaults
 % vi ~/.dnd/croaker/defaults  # adjust to taste
@@ -41,120 +22,35 @@ This assumes you have a functioning icecast2/whatever installation already.
 
 Now start the server, which will begin streaming the `session_start` playlist:
 
-## Controlling The Server
-
 ```
 % croaker start
-INFO Daemonizing controller on (localhost, 8003); pidfile and logs in ~/.dnd/croaker
+Daemonizing webserver on http://0.0.0.0:8003, pidfile and output in ~/.dnd/croaker
 ```
 
-Connnect to the command & control server:
+## Quick Start (Client)
-
-```bash
-% telnet localhost 8003
-Trying 127.0.0.1...
-Connected to croaker.local.
-Escape character is '^]'.
-
-help
-
-PLAY PLAYLIST    - Switch to the specified playlist.
-LIST [PLAYLIST]  - List playlists or contents of the specified list.
-FFWD             - Skip to the next track in the playlist.
-HELP             - Display command help.
-KTHX             - Close the current connection.
-STOP             - Stop the current track and stream silence.
-STFU             - Terminate the Croaker server.
 ```
-
+% mkdir -p ~/.dnd/croaker
-List available playlists:
+% croaker setup > ~/.dnd/croaker/defaults  # only the client config is required
-
+% vi ~/.dnd/croaker/defaults  # adjust to taste
-```
-list
-
-battle
-adventure
-session_start
 ```
 
 Switch to battle music -- roll initiative!
+
 ```
-play battle
+% croaker play battle
 OK
 ```
 
 Skip this track and move on to the next:
 
 ```
-ffwd
+% croaker skip
 OK
 ```
 
-Stop the music:
+Stop the server:
 
 ```
-stop
+% croaker stop
 OK
 ```
-
-Disconnect:
-
-```
-kthx
-KBAI
-Connection closed by foreign host.
-```
-
-## Python Client Implementation
-
-Here's a sample client using Ye Olde Socket Library:
-
-```python
-import socket
-from dataclasses import dataclass
-from functools import cached_property
-
-
-@dataclass
-class CroakerClient():
-    host: str
-    port: int
-
-    @cached_property
-    def playlists(self):
-        return self.send("LIST").split("\n")
-
-    def list(self, *args):
-        if not args:
-            return self.playlists
-        return self.send(f"LIST {args[0]}")
-
-    def play(self, *args):
-        if not args:
-            return "Error: Must specify the playlist to play."
-        return self.send(f"PLAY {args[0]}")
-
-    def ffwd(self, *args):
-        return self.send("FFWD")
-
-    def stop(self, *args):
-        return self.send("STOP")
-
-    def send(self, msg: str):
-        BUFSIZE = 4096
-        data = bytearray()
-        with socket.create_connection((self.host, self.port)) as sock:
-            sock.sendall(f"{msg}\n".encode())
-            while True:
-                buf = sock.recv(BUFSIZE)
-                data.extend(buf)
-                if len(buf) < BUFSIZE:
-                    break
-            sock.sendall(b'KTHX\n')
-        return data.decode()
-
-
-if __name__ == '__main__':
-    client = CroakerClient(host='localhost', port=1234)
-    client.play('session_start')
-```

croaker/cli.py

@@ -0,0 +1,207 @@
+import io
+import logging
+import os
+import sys
+from pathlib import Path
+from textwrap import dedent
+from typing import List, Optional
+
+import typer
+from dotenv import load_dotenv
+from typing_extensions import Annotated
+
+import croaker.path
+from croaker import client, controller, server
+from croaker.exceptions import ConfigurationError
+from croaker.playlist import Playlist
+
+SETUP_HELP = """
+# Root directory for croaker configuration and logs. See also croaker --root.
+CROAKER_ROOT=~/.dnd/croaker
+
+## COMMAND AND CONTROL WEBSERVER
+
+# Please make sure you set SECRET_KEY in your environment if you are running
+# the command and control webserver. Clients do not need this.
+SECRET_KEY=
+
+# Where the record the webserver daemon's PID
+PIDFILE=~/.dnd/croaker/croaker.pid
+
+# Web interface configuration
+HOST=127.0.0.1
+PORT=8003
+
+## CONTROLLER CLIENT
+
+# The host and port to use when connecting to the websever.
+CONTROLLER_HOST=127.0.0.1
+CONTROLLER_PORT=8003
+
+## MEDIA
+
+# where to store playlist sources
+PLAYLIST_ROOT=~/.dnd/croaker/playlists
+
+# where to cache transcoded media files
+CACHE_ROOT=~/.dnd/croaker/cache
+
+# the kinds of files to add to playlists
+MEDIA_GLOB=*.mp3,*.flac,*.m4a
+
+# If defined, transcode media before streaming it, and cache it to disk. The
+# strings INFILE and OUTFILE will be replaced with the media source file and
+# the cached output location, respectively.
+TRANSCODER=/usr/bin/ffmpeg -i INFILE '-hide_banner -loglevel error -codec:v copy -codec:a libmp3lame -q:a 2' OUTFILE
+
+## LIQUIDSOAP AND ICECAST
+
+# The liquidsoap executable
+LIQUIDSOAP=/usr/bin/liquidsoap
+
+# Icecast2 configuration for Liquidsoap
+ICECAST_PASSWORD=
+ICECAST_MOUNT=
+ICECAST_HOST=
+ICECAST_PORT=
+ICECAST_URL=
+"""
+
+
+app = typer.Typer()
+app_state = {}
+
+
+@app.callback()
+def main(
+    context: typer.Context,
+    root: Optional[Path] = typer.Option(
+        Path("~/.dnd/croaker"),
+        help="Path to the Croaker environment",
+    ),
+    host: Optional[str] = typer.Option(
+        None,
+        help="bind address",
+    ),
+    port: Optional[int] = typer.Option(
+        None,
+        help="bind port",
+    ),
+    debug: Optional[bool] = typer.Option(None, help="Enable debugging output"),
+):
+    load_dotenv(root.expanduser() / Path("defaults"))
+    load_dotenv(stream=io.StringIO(SETUP_HELP))
+    if host:
+        os.environ["HOST"] = host
+    if port:
+        os.environ["PORT"] = port
+    if debug is not None:
+        if debug:
+            os.environ["DEBUG"] = 1
+        else:
+            del os.environ["DEBUG"]
+
+    logging.basicConfig(
+        format="%(message)s",
+        level=logging.DEBUG if debug else logging.INFO,
+    )
+
+    try:
+        croaker.path.media_root()
+        croaker.path.cache_root()
+    except ConfigurationError as e:
+        sys.stderr.write(f"{e}\n\n{SETUP_HELP}")
+        sys.exit(1)
+
+    app_state["client"] = client.Client(
+        host=os.environ["CONTROLLER_HOST"],
+        port=os.environ["CONTROLLER_PORT"],
+    )
+
+    if not context.invoked_subcommand:
+        return play(context)
+
+
+@app.command()
+def setup(context: typer.Context):
+    """
+    (Re)Initialize Croaker.
+    """
+    sys.stderr.write("Interactive setup is not yet available. Sorry!\n")
+    print(dedent(SETUP_HELP))
+
+
+@app.command()
+def start(
+    context: typer.Context,
+    daemonize: bool = typer.Option(True, help="Daemonize the webserver."),
+):
+    """
+    Start the Croaker command and control webserver.
+    """
+    controller.start()
+    if daemonize:
+        server.daemonize()
+    else:
+        server.start()
+
+
+@app.command()
+def stop():
+    """
+    Terminate the webserver process and liquidsoap.
+    """
+    controller.stop()
+    server.stop()
+
+
+@app.command()
+def play(
+    playlist: str = typer.Argument(
+        ...,
+        help="Playlist name",
+    )
+):
+    """
+    Begin playing tracks from the directory $PLAYLIST_ROOT/[NAME].
+    """
+    res = app_state["client"].play(playlist)
+    if res.status_code == 200:
+        print("OK")
+
+
+@app.command()
+def skip():
+    """
+    Play the next track on the current playlist.
+    """
+    res = app_state["client"].skip()
+    if res.status_code == 200:
+        print("OK")
+
+
+@app.command()
+def add(
+    playlist: str = typer.Argument(
+        ...,
+        help="Playlist name",
+    ),
+    theme: Optional[bool] = typer.Option(False, help="Make the first track the theme song."),
+    tracks: Annotated[Optional[List[Path]], typer.Argument()] = None,
+):
+    """
+    Recursively add one or more paths to the specified playlist. Tracks can be
+    any combination of individual audio files and directories containing audio
+    files; anything not already on the playlist will be added to it.
+
+    If --theme is specified, the first track will be designated the playlist
+    "theme." Theme songs get played first whenever the playlist is loaded,
+    after which the playlist order is randomized.
+    """
+    pl = Playlist(name=playlist)
+    pl.add(tracks, make_theme=theme)
+    print(pl)
+
+
+if __name__ == "__main__":
+    app.main()

croaker/client.py

@@ -0,0 +1,43 @@
+import logging
+from dataclasses import dataclass
+from functools import cached_property
+
+import bottle
+import requests
+
+# needs to be imported to attach routes to the default app
+from croaker import routes
+
+assert routes
+
+
+@dataclass
+class Client:
+    host: str
+    port: int
+
+    @cached_property
+    def _session(self):
+        return requests.Session()
+
+    @property
+    def _routes(self):
+        return [r.callback.__name__ for r in bottle.default_app().routes]
+
+    def get(self, uri: str, *args, **params):
+        url = f"http://{self.host}:{self.port}/{uri}"
+        if args:
+            url += "/" + "/".join(args)
+        res = self._session.get(url, params=params)
+        logging.debug(f"{url = }, {res = }")
+        return res
+
+    def __getattr__(self, attr):
+        if attr in self._routes:
+
+            def dispatch(*args, **kwargs):
+                logging.debug(f"calling attr, {args = }, {kwargs = }")
+                return self.get(attr, *args, **kwargs)
+
+            return dispatch
+        return self.__getattribute__(attr)

croaker/controller.py

@@ -0,0 +1,141 @@
+import logging
+import os
+from pathlib import Path
+from subprocess import Popen
+from time import sleep
+
+from Exscript.protocols import Telnet
+
+from croaker import path
+from croaker.pidfile import pidfile
+from croaker.playlist import Playlist, load_playlist
+
+NOW_PLAYING = None
+
+LIQUIDSOAP_CONFIG = """
+set("server.telnet",true)
+set("request.grace_time", 1.0)
+set("init.daemon.pidfile.path", "{pidfile.path}")
+set("decoder.ffmpeg.codecs.alac", ["alac"])
+
+
+# deeebuggin
+set("log.file.path","{debug_log}")
+
+# set up the stream
+stream = crossfade(normalize(playlist.safe(
+    id='stream',
+    reload_mode='watch',
+    mode='normal',
+    '{playlist_root}/now_playing',
+)))
+
+# if source files don't contain metadata tags, use the filename
+def apply_metadata(m) =
+    title = m["title"]
+    print("Now Playing: #{{m['filename']}}")
+    if (title == "") then
+        [("title", "#{{path.remove_extension(path.basename(m['filename']))}}")]
+    else
+        [("title", "#{{title}}")]
+    end
+end
+
+# apply the metadata parser
+stream = map_metadata(apply_metadata, stream)
+
+# define the source. ignore errors and provide no infallibale fallback. yolo.
+radio = fallback(track_sensitive=false, [stream])
+
+# transcode to icecast
+output.icecast(
+  %mp3.vbr(quality=3),
+  name='Croaker Radio',
+  description='Background music for The Frog Hat Club',
+  host="{icecast_host}",
+  port={icecast_port},
+  password="{icecast_password}",
+  mount="{icecast_mount}",
+  icy_metadata="true",
+  url="{icecast_url}",
+  fallible=true,
+  radio
+)
+"""
+
+
+def generate_liquidsoap_config():
+    log = path.root() / "liquidsoap.log"
+    if log.exists():
+        log.unlink()
+        log.touch()
+    ls_config = path.root() / "croaker.liq"
+    with ls_config.open("wt") as fh:
+        fh.write(
+            LIQUIDSOAP_CONFIG.format(
+                pidfile=_pidfile(terminate_if_running=False),
+                debug_log=log,
+                playlist_root=path.playlist_root(),
+                icecast_host=os.environ.get("ICECAST_HOST"),
+                icecast_port=os.environ.get("ICECAST_PORT"),
+                icecast_mount=os.environ.get("ICECAST_MOUNT"),
+                icecast_password=os.environ.get("ICECAST_PASSWORD"),
+                icecast_url=os.environ.get("ICECAST_URL"),
+            )
+        )
+    path.playlist_root().mkdir(exist_ok=True)
+
+
+def start_liquidsoap():
+    logging.debug("Staring liquidsoap...")
+    pf = _pidfile(terminate_if_running=False)
+    pid = pf.read_pid()
+    if not pid:
+        logging.info("Liquidsoap does not appear to be running. Starting it...")
+        generate_liquidsoap_config()
+        Popen([os.environ["LIQUIDSOAP"], "--daemon", path.root() / "croaker.liq"])
+        sleep(1)
+
+
+def start():
+    play_next("session_start")
+
+
+def stop():
+    _pidfile(terminate_if_running=True)
+
+
+def play_next(playlist_name: str = None):
+    start_liquidsoap()
+    if playlist_name:
+        pl = load_playlist(playlist_name)
+        logging.debug(f"Loaded playlist {pl = }")
+        if NOW_PLAYING != pl.name:
+            _switch_to(pl)
+    _send_liquidsoap_command("skip")
+
+
+def _pidfile(terminate_if_running: bool = True):
+    pf = os.environ.get("LIQUIDSOAP_PIDFILE", None)
+    if pf:
+        pf = Path(pf)
+    else:
+        pf = path.root() / "liquidsoap.pid"
+    return pidfile(pf, terminate_if_running=terminate_if_running)
+
+
+def _switch_to(playlist: Playlist):
+    logging.debug(f"Switching to {playlist = }")
+    np = path.playlist_root() / Path("now_playing")
+    with np.open("wt") as fh:
+        for track in playlist.tracks:
+            fh.write(f"{track}\n")
+    playlist.name
+
+
+def _send_liquidsoap_command(command: str):
+    conn = Telnet()
+    conn.connect("localhost", port=1234)
+    conn.send(f"Croaker_Radio.{command}\r")
+    conn.send("quit\r")
+    conn.close()

croaker/exceptions.py

@@ -0,0 +1,16 @@
+class APIHandlingException(Exception):
+    """
+    An API reqeust could not be encoded or decoded.
+    """
+
+
+class ConfigurationError(Exception):
+    """
+    An error was discovered with the Groove on Demand configuration.
+    """
+
+
+class InvalidPathError(Exception):
+    """
+    The specified path was invalid -- either it was not the expected type or wasn't accessible.
+    """

croaker/path.py

@@ -0,0 +1,47 @@
+import logging
+import os
+from pathlib import Path
+
+from croaker.exceptions import ConfigurationError
+
+_setup_hint = "You may be able to solve this error by running 'croaker setup' or specifying the --root parameter."
+_reinstall_hint = "You might need to reinstall Groove On Demand to fix this error."
+
+
+def root():
+    return Path(os.environ.get("CROAKER_ROOT", "~/.dnd/croaker")).expanduser()
+
+
+def media_root():
+    path = os.environ.get("MEDIA_ROOT", None)
+    if not path:
+        raise ConfigurationError(f"MEDIA_ROOT is not defined in your environment.\n\n{_setup_hint}")
+    path = Path(path).expanduser()
+    if not path.exists() or not path.is_dir():
+        raise ConfigurationError(
+            "The media_root directory (MEDIA_ROOT) doesn't exist, or isn't a directory.\n\n{_setup_hint}"
+        )
+    logging.debug(f"Media root is {path}")
+    return path
+
+
+def cache_root():
+    path = Path(os.environ.get("CACHE_ROOT", root() / Path("cache"))).expanduser()
+    logging.debug(f"Media cache root is {path}")
+    return path
+
+
+def playlist_root():
+    path = Path(os.environ.get("PLAYLIST_ROOT", root() / Path("playlsits"))).expanduser()
+    logging.debug(f"Playlist root is {path}")
+    return path
+
+
+def media(relpath):
+    path = media_root() / Path(relpath)
+    return path
+
+
+def transcoded_media(relpath):
+    path = cache_root() / Path(relpath + ".webm")
+    return path

croaker/pidfile.py

@@ -5,17 +5,15 @@ from pathlib import Path
 
 from daemon import pidfile as _pidfile
 
-logger = logging.getLogger("daemon")
 
-
+def pidfile(pidfile_path: Path, terminate_if_running: bool = True):
-def pidfile(pidfile_path: Path, sig=signal.SIGQUIT, terminate_if_running: bool = True):
     pf = _pidfile.TimeoutPIDLockFile(str(pidfile_path.expanduser()), 30)
     pid = pf.read_pid()
     if pid and terminate_if_running:
         try:
-            logger.debug(f"Stopping PID {pid}")
+            logging.debug(f"Stopping PID {pid}")
-            os.kill(pid, sig)
+            os.kill(pid, signal.SIGTERM)
         except ProcessLookupError:
-            logger.debug(f"PID {pid} not running; breaking lock.")
+            logging.debug(f"PID {pid} not running; breaking lock.")
             pf.break_lock()
     return pf

croaker/playlist.py

@@ -9,8 +9,6 @@ from typing import List
 
 import croaker.path
 
-logger = logging.getLogger("playlist")
-
 playlists = {}
 
 
@@ -24,20 +22,21 @@ def _stripped(name):
 class Playlist:
     name: str
     theme: Path = Path("_theme.mp3")
+    current_track: int = 0
 
     @cached_property
     def path(self):
-        return self._get_path()
+        return croaker.path.playlist_root() / Path(self.name)
 
     @cached_property
     def tracks(self):
         if not self.path.exists():
-            raise RuntimeError(f"Playlist {self.name} not found at {self.path}.")  # pragma: no cover
+            raise RuntimeError(f"Playlist {self.name} not found at {self.path}.")
 
         entries = []
         theme = self.path / self.theme
         if theme.exists():
-            entries.append(theme)
+            entries[0] = theme
         files = [e for e in self.get_audio_files() if e.name != "_theme.mp3"]
         if files:
             shuffle(files)
@@ -49,12 +48,13 @@ class Playlist:
             path = self.path
         logging.debug(f"Getting files matching {os.environ['MEDIA_GLOB']} from {path}")
         pats = os.environ["MEDIA_GLOB"].split(",")
-        return chain(*[list(path.rglob(pat)) for pat in pats])
+        return chain(*[list(path.glob(pat)) for pat in pats])
 
-    def _get_path(self):
+    def _add_track(self, target: Path, source: Path, make_theme: bool = False):
-        return croaker.path.playlist_root() / self.name
+        if source.is_dir():
-
+            for file in self.get_audio_files(source):
-    def _add_track(self, target: Path, source: Path):
+                self._add_track(self.path / _stripped(file.name), file)
+            return
         if target.exists():
             if not target.is_symlink():
                 logging.warning(f"{target}: target already exists and is not a symlink; skipping.")
@@ -62,26 +62,16 @@ class Playlist:
             target.unlink()
         target.symlink_to(source)
 
-    def add(self, paths: List[Path], make_theme: bool = False):
+    def add(self, tracks: List[Path], make_theme: bool = False):
-        logger.debug(f"Adding everything from {paths = }")
         self.path.mkdir(parents=True, exist_ok=True)
-        for path in paths:
-            if path.is_dir():
-                files = list(self.get_audio_files(path))
         if make_theme:
-                    logger.debug(f"Adding first file from dir as theme: {files[0] = }")
+            if source.is_dir():
-                    self._add_track(self.path / "_theme.mp3", files.pop(0))
+                raise RuntimeError(f"Cannot create a playlist theme from a directory: {source}")
-                    make_theme = False
+            target = self.path / "_theme.mp3"
-                for file in files:
+            source = tracks.pop(0)
-                    logger.debug(f"Adding {file = }")
+            self._add_track(target, source, make_theme=True)
-                    self._add_track(target=self.path / _stripped(file.name), source=file)
+        for track in tracks:
-            elif make_theme:
+            self._add_track(target=self.path / _stripped(track.name), source=track)
-                logger.debug(f"Adding path as theme: {path = }")
-                self._add_track(self.path / "_theme.mp3", path)
-                make_theme = False
-            else:
-                logger.debug(f"Adding {path = }")
-                self._add_track(target=self.path / _stripped(path.name), source=path)
         return sorted(self.get_audio_files())
 
     def __repr__(self):
@@ -90,7 +80,7 @@ class Playlist:
         return "\n".join(lines)
 
 
-def load_playlist(name: str):  # pragma: no cover
+def load_playlist(name: str):
     if name not in playlists:
         playlists[name] = Playlist(name=name)
     return playlists[name]

croaker/routes.py

@@ -0,0 +1,17 @@
+from bottle import route
+
+from croaker import controller
+
+
+@route("/play/<playlist_name>")
+def play(playlist_name=None):
+    if not controller.play_next(playlist_name):
+        return
+    return "OK"
+
+
+@route("/skip")
+def skip():
+    if not controller.play_next():
+        return
+    return "OK"

croaker/server.py

@@ -0,0 +1,43 @@
+import logging
+import os
+from pathlib import Path
+
+import bottle
+import daemon
+
+from croaker import path, routes
+from croaker.pidfile import pidfile
+
+assert routes
+app = bottle.default_app()
+
+
+def _pidfile(terminate_if_running: bool = True):
+    pf = os.environ.get("PIDFILE", None)
+    if pf:
+        pf = Path(pf)
+    else:
+        pf = path.root() / "croaker.pid"
+    return pidfile(pf, terminate_if_running=terminate_if_running)
+
+
+def daemonize(host: str = "0.0.0.0", port: int = 8003, debug: bool = False) -> None:  # pragma: no cover
+    logging.info(f"Daemonizing webserver on http://{host}:{port}, pidfile and output in {path.root()}")
+    context = daemon.DaemonContext()
+    context.pidfile = _pidfile()
+    context.stdout = open(path.root() / Path("croaker.out"), "wb")
+    context.stderr = open(path.root() / Path("croaker.err"), "wb", buffering=0)
+    context.open()
+    start(host, port, debug)
+
+
+def stop():
+    _pidfile()
+
+
+def start(host: str = "0.0.0.0", port: int = 8003, debug: bool = False) -> None:  # pragma: no cover
+    """
+    Start the Bottle app.
+    """
+    logging.debug(f"Configuring webserver with host={host}, port={port}, debug={debug}")
+    app.run(host=os.getenv("HOST", host), port=os.getenv("PORT", port), debug=debug, server="paste", quiet=True)

pyproject.toml

@@ -1,41 +1,34 @@
 [tool.poetry]
 name = "croaker"
-version = "0.9.2"
+version = "0.1.1"
 description = ""
 authors = ["evilchili <evilchili@gmail.com>"]
 readme = "README.md"
 packages = [
-    { include = "*", from = "src" }
+    { include = "croaker" }
 ]
 
 [tool.poetry.dependencies]
-python = ">=3.10,<4.0"
+python = "^3.8"
 prompt-toolkit = "^3.0.38"
 typer = "^0.9.0"
 python-dotenv = "^0.21.0"
 rich = "^13.7.0"
 pyyaml = "^6.0.1"
+bottle = "^0.12.25"
 paste = "^3.7.1"
 python-daemon = "^3.0.1"
 requests = "^2.31.0"
 psutil = "^5.9.8"
 exscript = "^2.6.28"
-python-shout = "^0.2.8"
-ffmpeg-python = "^0.2.0"
 
 [tool.poetry.scripts]
 croaker = "croaker.cli:app"
 
-[tool.poetry.group.dev.dependencies]
+[tool.poetry.dev-dependencies]
-pytest = "^8.1.1"
+black = "^23.3.0"
-pytest-cov = "^5.0.0"
+isort = "^5.12.0"
-
+pyproject-autoflake = "^1.0.2"
-[build-system]
-requires = ["poetry-core"]
-build-backend = "poetry.core.masonry.api"
-
-
-### SLAM
 
 [tool.black]
 line-length = 120
@@ -55,8 +48,7 @@ ignore-init-module-imports = true    # exclude __init__.py when removing unused
 remove-duplicate-keys = true         # remove all duplicate keys in objects
 remove-unused-variables = true       # remove unused variables
 
-[tool.pytest.ini_options]
-log_cli_level = "DEBUG"
-addopts = "--cov=src --cov-report=term-missing"
 
-### ENDSLAM
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"

src/croaker/cli.py

@@ -1,129 +0,0 @@
-import io
-import logging
-import os
-import sys
-from pathlib import Path
-from textwrap import dedent
-from typing import List, Optional
-
-import typer
-from dotenv import load_dotenv
-from typing_extensions import Annotated
-
-from croaker import path
-from croaker.playlist import Playlist
-from croaker.server import server
-
-SETUP_HELP = f"""
-# Root directory for croaker configuration and logs. See also croaker --root.
-CROAKER_ROOT={path.root()}
-
-# where to store playlist sources
-#PLAYLIST_ROOT={path.root()}/playlists
-
-# Where the record the daemon's PID
-#PIDFILE={path.root()}/croaker.pid
-
-# Command and Control TCP Server bind address
-HOST=0.0.0.0
-PORT=8003
-
-# the kinds of files to add to playlists
-MEDIA_GLOB=*.mp3,*.flac,*.m4a
-
-# Icecast2 configuration for Liquidsoap
-ICECAST_PASSWORD=
-ICECAST_MOUNT=
-ICECAST_HOST=
-ICECAST_PORT=
-ICECAST_URL=
-"""
-
-app = typer.Typer()
-app_state = {}
-
-logger = logging.getLogger("cli")
-
-
-@app.callback()
-def main(
-    context: typer.Context,
-    root: Optional[Path] = typer.Option(
-        Path("~/.dnd/croaker"),
-        help="Path to the Croaker environment",
-    ),
-    debug: Optional[bool] = typer.Option(None, help="Enable debugging output"),
-):
-    load_dotenv(root.expanduser() / Path("defaults"))
-    load_dotenv(stream=io.StringIO(SETUP_HELP))
-    if debug is not None:
-        if debug:
-            os.environ["DEBUG"] = "1"
-        else:
-            del os.environ["DEBUG"]
-
-    logging.basicConfig(
-        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
-        level=logging.DEBUG if debug else logging.INFO,
-    )
-
-
-@app.command()
-def setup(context: typer.Context):
-    """
-    (Re)Initialize Croaker.
-    """
-
-    sys.stderr.write(
-        "Interactive setup is not available, but you can redirect "
-        "this command's output to a defaults file of your choice.\n"
-    )
-    print(dedent(SETUP_HELP))
-
-
-@app.command()
-def start(
-    context: typer.Context,
-    daemonize: bool = typer.Option(True, help="Daemonize the server."),
-):
-    """
-    Start the Croaker command and control server.
-    """
-    server.start(daemonize=daemonize)
-
-
-@app.command()
-def stop():
-    """
-    Terminate the server.
-    """
-    server.stop()
-
-
-@app.command()
-def add(
-    playlist: str = typer.Argument(
-        ...,
-        help="Playlist name",
-    ),
-    theme: Optional[bool] = typer.Option(False, help="Make the first track the theme song."),
-    tracks: Annotated[Optional[List[Path]], typer.Argument()] = None,
-):
-    """
-    Recursively add one or more paths to the specified playlist.
-
-    Tracks can be any combination of individual audio files and directories
-    containing audio files; anything not already on the playlist will be
-    added to it.
-
-    If --theme is specified, the first track will be designated the playlist
-    "theme." Theme songs get played first whenever the playlist is loaded,
-    after which the playlist order is randomized.
-    """
-    pl = Playlist(name=playlist)
-    pl.add(tracks, make_theme=theme)
-    print(pl)
-
-
-if __name__ == "__main__":
-    app.main()

src/croaker/path.py

@@ -1,14 +0,0 @@
-import os
-from pathlib import Path
-
-_setup_hint = "You may be able to solve this error by running 'croaker setup' or specifying the --root parameter."
-_reinstall_hint = "You might need to reinstall Croaker to fix this error."
-
-
-def root():
-    return Path(os.environ.get("CROAKER_ROOT", "~/.dnd/croaker")).expanduser()
-
-
-def playlist_root():
-    path = Path(os.environ.get("PLAYLIST_ROOT", root() / "playlists")).expanduser()
-    return path

src/croaker/server.py

@@ -1,190 +0,0 @@
-import logging
-import os
-import queue
-import socketserver
-import threading
-from pathlib import Path
-from time import sleep
-
-import daemon
-
-from croaker import path
-from croaker.pidfile import pidfile
-from croaker.playlist import load_playlist
-from croaker.streamer import AudioStreamer
-
-logger = logging.getLogger("server")
-
-
-class RequestHandler(socketserver.StreamRequestHandler):
-    """
-    Instantiated by the TCPServer when a request is received. Implements the
-    command and control protocol and sends commands to the shoutcast source
-    client on behalf of the user.
-    """
-
-    supported_commands = {
-        # command              # help text
-        "PLAY": "PLAYLIST    - Switch to the specified playlist.",
-        "LIST": "[PLAYLIST]  - List playlists or contents of the specified list.",
-        "FFWD": "            - Skip to the next track in the playlist.",
-        "HELP": "            - Display command help.",
-        "KTHX": "            - Close the current connection.",
-        "STOP": "            - Stop the current track and stream silence.",
-        "STFU": "            - Terminate the Croaker server.",
-    }
-
-    should_listen = True
-
-    def handle(self):
-        """
-        Start a command and control session. Commands are read one line at a
-        time; the format is:
-
-        Byte     Definition
-        -------------------
-        0-3      Command
-        4        Ignored
-        5+       Arguments
-        """
-        while True:
-            self.data = self.rfile.readline().strip().decode()
-            logger.debug(f"Received: {self.data}")
-            try:
-                cmd = self.data[0:4].strip().upper()
-                args = self.data[5:]
-            except IndexError:
-                self.send(f"ERR Command not understood '{cmd}'")
-                sleep(0.001)
-                continue
-
-            if not cmd:
-                sleep(0.001)
-                continue
-            elif cmd not in self.supported_commands:
-                self.send(f"ERR Unknown Command '{cmd}'")
-                sleep(0.001)
-                continue
-            elif cmd == "KTHX":
-                return self.send("KBAI")
-
-            handler = getattr(self, f"handle_{cmd}", None)
-            if not handler:
-                self.send(f"ERR No handler for {cmd}.")
-            handler(args)
-            if not self.should_listen:
-                break
-
-    def send(self, msg):
-        return self.wfile.write(msg.encode() + b"\n")
-
-    def handle_PLAY(self, args):
-        self.server.load(args)
-        return self.send("OK")
-
-    def handle_FFWD(self, args):
-        self.server.ffwd()
-        return self.send("OK")
-
-    def handle_LIST(self, args):
-        return self.send(self.server.list(args))
-
-    def handle_HELP(self, args):
-        return self.send("\n".join(f"{cmd} {txt}" for cmd, txt in self.supported_commands.items()))
-
-    def handle_STOP(self, args):
-        return self.server.stop_event.set()
-
-    def handle_STFU(self, args):
-        self.send("Shutting down.")
-        self.server.stop()
-
-
-class CroakerServer(socketserver.TCPServer):
-    """
-    A Daemonized TCP Server that also starts a Shoutcast source client.
-    """
-
-    allow_reuse_address = True
-
-    def __init__(self):
-        self._context = daemon.DaemonContext()
-        self._queue = queue.Queue()
-        self.skip_event = threading.Event()
-        self.stop_event = threading.Event()
-        self.load_event = threading.Event()
-        self._streamer = None
-        self.playlist = None
-
-    def _pidfile(self):
-        return pidfile(path.root() / "croaker.pid")
-
-    @property
-    def streamer(self):
-        if not self._streamer:
-            self._streamer = AudioStreamer(self._queue, self.skip_event, self.stop_event, self.load_event)
-        return self._streamer
-
-    def bind_address(self):
-        return (os.environ["HOST"], int(os.environ["PORT"]))
-
-    def _daemonize(self) -> None:
-        """
-        Daemonize the current process.
-        """
-        logger.info(f"Daemonizing controller; pidfile and output in {path.root()}")
-        self._context.pidfile = self._pidfile()
-        self._context.stdout = open(path.root() / Path("croaker.out"), "wb", buffering=0)
-        self._context.stderr = open(path.root() / Path("croaker.err"), "wb", buffering=0)
-
-        # when open() is called, all open file descriptors will be closed, as
-        # befits a good daemon. However this will also close the socket on
-        # which the TCPServer is listening! So let's keep that one open.
-        self._context.files_preserve = [self.fileno()]
-        self._context.open()
-
-    def start(self, daemonize: bool = True) -> None:
-        """
-        Start the shoutcast controller background thread, then begin listening for connections.
-        """
-        logger.info(f"Starting controller on {self.bind_address()}.")
-        super().__init__(self.bind_address(), RequestHandler)
-        if daemonize:
-            self._daemonize()
-        try:
-            logger.debug("Starting AudioStreamer...")
-            self.streamer.start()
-            self.load("session_start")
-            self.serve_forever()
-        except KeyboardInterrupt:
-            logger.info("Shutting down.")
-            self.stop()
-
-    def stop(self):
-        self._pidfile()
-
-    def ffwd(self):
-        logger.debug("Sending SKIP signal to streamer...")
-        self.skip_event.set()
-
-    def clear_queue(self):
-        logger.debug("Requesting a reload...")
-        self.streamer.load_requested.set()
-        sleep(0.5)
-
-    def list(self, playlist_name: str = None):
-        if playlist_name:
-            return str(load_playlist(playlist_name))
-        return "\n".join([str(p.name) for p in path.playlist_root().iterdir()])
-
-    def load(self, playlist_name: str):
-        logger.debug(f"Switching to {playlist_name = }")
-        if self.playlist:
-            self.clear_queue()
-        self.playlist = load_playlist(playlist_name)
-        logger.debug(f"Loaded new playlist {self.playlist = }")
-        for track in self.playlist.tracks:
-            self._queue.put(str(track).encode())
-
-
-server = CroakerServer()

src/croaker/streamer.py

@@ -1,116 +0,0 @@
-import logging
-import os
-import queue
-import threading
-from functools import cached_property
-from pathlib import Path
-from time import sleep
-
-import shout
-
-from croaker.transcoder import FrameAlignedStream
-
-logger = logging.getLogger("streamer")
-
-
-class AudioStreamer(threading.Thread):
-    """
-    Receive filenames from the controller thread and stream the contents of
-    those files to the icecast server.
-    """
-
-    def __init__(self, queue, skip_event, stop_event, load_event, chunk_size=4096):
-        super().__init__()
-        self.queue = queue
-        self.skip_requested = skip_event
-        self.stop_requested = stop_event
-        self.load_requested = load_event
-        self.chunk_size = chunk_size
-
-    @property
-    def silence(self):
-        return FrameAlignedStream.from_source(Path(__file__).parent / "silence.mp3", chunk_size=self.chunk_size)
-
-    @cached_property
-    def _shout(self):
-        s = shout.Shout()
-        s.name = "Croaker Radio"
-        s.url = os.environ["ICECAST_URL"]
-        s.mount = os.environ["ICECAST_MOUNT"]
-        s.host = os.environ["ICECAST_HOST"]
-        s.port = int(os.environ["ICECAST_PORT"])
-        s.password = os.environ["ICECAST_PASSWORD"]
-        s.protocol = os.environ.get("ICECAST_PROTOCOL", "http")
-        s.format = os.environ.get("ICECAST_FORMAT", "mp3")
-        return s
-
-    def run(self):  # pragma: no cover
-        while True:
-            try:
-                logger.debug(f"Connecting to shoutcast server at {self._shout.host}:{self._shout.port}")
-                self._shout.open()
-            except shout.ShoutException as e:
-                logger.error("Error connecting to shoutcast server. Will sleep and try again.", exc_info=e)
-                sleep(3)
-                continue
-
-            try:
-                self.stream_queued_audio()
-            except Exception as exc:
-                logger.error("Caught exception.", exc_info=exc)
-        self._shout.close()
-
-    def clear_queue(self):
-        logger.debug("Clearing queue...")
-        while not self.queue.empty():
-            self.queue.get()
-
-    def queued_audio_source(self):
-        """
-        Return a filehandle to the next queued audio source, or silence if the queue is empty.
-        """
-        try:
-            track = Path(self.queue.get(block=False).decode())
-            logger.debug(f"Streaming {track.stem = }")
-            return FrameAlignedStream.from_source(track, chunk_size=self.chunk_size), track.stem
-        except queue.Empty:
-            logger.debug("Nothing queued; enqueing silence.")
-        except Exception as exc:
-            logger.error("Caught exception; falling back to silence.", exc_info=exc)
-        return self.silence, "[NOTHING PLAYING]"
-
-    def stream_queued_audio(self):
-        stream = None
-        title = None
-        next_stream = None
-        next_title = None
-
-        while True:
-            stream, title = (next_stream, next_title) if next_stream else self.queued_audio_source()
-            logging.debug(f"Starting stream of {title = }, {stream = }")
-            self._shout.set_metadata({"song": title})
-            next_stream, next_title = self.queued_audio_source()
-
-            for chunk in stream:
-                self._shout.send(chunk)
-                self._shout.sync()
-
-                # play the next source immediately
-                if self.skip_requested.is_set():
-                    logger.debug("Skip was requested.")
-                    self.skip_requested.clear()
-                    break
-
-                # clear the queue
-                if self.load_requested.is_set():
-                    logger.debug("Load was requested.")
-                    self.clear_queue()
-                    self.load_requested.clear()
-                    break
-
-                # Stop streaming and clear the queue
-                if self.stop_requested.is_set():
-                    logger.debug("Stop was requested.")
-                    self.clear_queue()
-                    self.stop_requested.clear()
-                    break

src/croaker/transcoder.py

@@ -1,132 +0,0 @@
-import logging
-import os
-import subprocess
-from dataclasses import dataclass
-from io import BufferedReader
-from pathlib import Path
-
-import ffmpeg
-
-logger = logging.getLogger("transcoder")
-
-
-@dataclass
-class FrameAlignedStream:
-    """
-    Use ffmpeg to transcode a source audio file to mp3 and iterate over the result
-    in frame-aligned chunks. This will ensure that readers will always have a full
-    frame of audio data to parse or emit.
-
-    I learned a lot from https://github.com/pylon/streamp3 figuring this stuff out!
-
-    Usage:
-
-        >>> stream = FrameAlignedStream.from_source(Path('test.flac').open('rb'))
-        >>> for segment in stream:
-            ...
-    """
-
-    source: BufferedReader
-    chunk_size: int = 1024
-    bit_rate: int = 192000
-    sample_rate: int = 44100
-
-    @property
-    def frames(self):
-        while True:
-            frame = self._read_one_frame()
-            if not frame:
-                return
-            yield frame
-
-    def _read_one_frame(self):
-        """
-        Read the next full audio frame from the input source and return it
-        """
-
-        # step through the source a byte at a time and look for the frame sync.
-        header = None
-        buffer = b""
-        while not header:
-            buffer += self.source.read(4 - len(buffer))
-            if len(buffer) != 4:
-                logging.debug("Reached the end of the source stream without finding another framesync.")
-                return False
-            header = buffer[:4]
-            if header[0] != 0b11111111 or header[1] >> 5 != 0b111:
-                logging.debug(f"Expected a framesync but got {buffer} instead; moving fwd 1 byte.")
-                header = None
-                buffer = buffer[1:]
-
-        # Decode the mp3 header. We could derive the bit_rate and sample_rate
-        # here if we had the lookup tables etc. from the MPEG spec, but since
-        # we control the input, we can rely on them being predefined.
-        version_code = (header[1] & 0b00011000) >> 3
-        padding_code = (header[2] & 0b00000010) >> 1
-        version = version_code & 1 if version_code >> 1 else 2
-        is_padded = bool(padding_code)
-
-        # calculate the size of the whole frame
-        frame_size = 1152 if version == 1 else 576
-        frame_size = self.bit_rate // 8 * frame_size // self.sample_rate
-        if is_padded:
-            frame_size += 1
-
-        # read the rest of the frame from the source
-        frame_data = self.source.read(frame_size - len(header))
-        if len(frame_data) != frame_size - len(header):
-            logging.debug("Reached the end of the source stream without finding a full frame.")
-            return None
-
-        # return the entire frame
-        return header + frame_data
-
-    def __iter__(self):
-        """
-        Generate approximately chunk_size segments of audio data by iterating over the
-        frames, buffering them, and then yielding several as a single bytes object.
-        """
-        buf = b""
-        for frame in self.frames:
-            if len(buf) >= self.chunk_size:
-                yield buf
-                buf = b""
-            if not frame:
-                break
-            buf += frame
-        if buf:
-            yield buf
-
-    @classmethod
-    def from_source(cls, infile: Path, **kwargs):
-        """
-        Create a FrameAlignedStream instance by transcoding an audio source on disk.
-        """
-
-        args = [] if os.environ.get("DEBUG") else ["-hide_banner", "-loglevel", "quiet"]
-        ffmpeg_args = (
-            ffmpeg.input(str(infile))
-            .output(
-                "pipe:",
-                map="a",
-                format="mp3",
-                # no ID3 headers -- saves having to decode them later
-                write_xing=0,
-                id3v2_version=0,
-                # force sample and bit rates
-                **{
-                    "b:a": kwargs.get("bit_rate", cls.bit_rate),
-                    "ar": kwargs.get("sample_rate", cls.sample_rate),
-                },
-            )
-            .global_args("-vn", *args)
-            .compile()
-        )
-
-        # Force close STDIN to prevent ffmpeg from trying to read from it. silly ffmpeg.
-        proc = subprocess.Popen(
-            ffmpeg_args, bufsize=kwargs.get("chunk_size", cls.chunk_size), stdout=subprocess.PIPE, stdin=subprocess.PIPE
-        )
-        proc.stdin.close()
-        logger.debug(f"Spawned ffmpeg (PID {proc.pid}) with args {ffmpeg_args = }")
-        return cls(proc.stdout, **kwargs)

test/conftest.py

@@ -1,16 +0,0 @@
-from pathlib import Path
-
-import pytest
-
-
-@pytest.fixture(autouse=True)
-def mock_env(monkeypatch):
-    fixtures = Path(__file__).parent / "fixtures"
-    monkeypatch.setenv("CROAKER_ROOT", str(fixtures))
-    monkeypatch.setenv("MEDIA_GLOB", "*.mp3,*.foo,*.bar")
-    monkeypatch.setenv("ICECAST_URL", "http://127.0.0.1")
-    monkeypatch.setenv("ICECAST_HOST", "localhost")
-    monkeypatch.setenv("ICECAST_MOUNT", "mount")
-    monkeypatch.setenv("ICECAST_PORT", "6523")
-    monkeypatch.setenv("ICECAST_PASSWORD", "password")
-    monkeypatch.setenv("DEBUG", "1")

test/fixtures/playlists/test_playlist/_theme.mp3

@@ -1 +0,0 @@
-_theme.mp3

test/fixtures/playlists/test_playlist/one.mp3

@@ -1 +0,0 @@
-one.mp3

test/fixtures/playlists/test_playlist/two.mp3

@@ -1 +0,0 @@
-two.mp3

test/test_pidfile.py

@@ -1,35 +0,0 @@
-from pathlib import Path
-from unittest.mock import MagicMock
-
-import pytest
-
-from croaker import pidfile
-
-
-@pytest.mark.parametrize(
-    "pid,terminate,kill_result,broken",
-    [
-        ("pid", False, None, False),  # running proc, no terminate
-        ("pid", True, True, False),  # running proc, terminate
-        ("pid", True, ProcessLookupError, True),  # stale pid
-        (None, None, None, False),  # no running proc
-    ],
-)
-def test_pidfile(monkeypatch, pid, terminate, kill_result, broken):
-    monkeypatch.setattr(
-        pidfile._pidfile,
-        "TimeoutPIDLockFile",
-        MagicMock(
-            **{
-                "return_value.read_pid.return_value": pid,
-            }
-        ),
-    )
-    monkeypatch.setattr(
-        pidfile.os,
-        "kill",
-        MagicMock(**{"side_effect": kill_result if type(kill_result) is Exception else [kill_result]}),
-    )
-
-    ret = pidfile.pidfile(pidfile_path=Path("/dev/null"), terminate_if_running=terminate)
-    assert ret.break_lock.called == broken

test/test_playlist.py

@@ -1,44 +0,0 @@
-from unittest.mock import MagicMock
-
-import pytest
-
-import croaker.path
-import croaker.playlist
-
-
-def test_playlist_loading():
-    pl = croaker.playlist.Playlist(name="test_playlist")
-    path = str(pl.path)
-    tracks = [str(t) for t in pl.tracks]
-
-    assert path == str(croaker.path.playlist_root() / pl.name)
-    assert pl.name == "test_playlist"
-    assert tracks[0] == f"{path}/_theme.mp3"
-    assert f"{path}/one.mp3" in tracks
-    assert f"{path}/two.mp3" in tracks
-    assert f"{path}/one.foo" in tracks
-    assert f"{path}/one.baz" not in tracks
-
-
-@pytest.mark.parametrize(
-    "paths, make_theme, expected_count",
-    [
-        (["test_playlist"], True, 4),
-        (["test_playlist"], False, 4),
-        (["test_playlist", "sources/one.mp3"], True, 5),
-        (["test_playlist", "sources/one.mp3"], False, 5),
-    ],
-)
-def test_playlist_creation(monkeypatch, paths, make_theme, expected_count):
-    new_symlinks = []
-
-    def symlink(target):
-        new_symlinks.append(target)
-
-    pl = croaker.playlist.Playlist(name="foo")
-    monkeypatch.setattr(croaker.playlist.Path, "unlink", MagicMock())
-    monkeypatch.setattr(croaker.playlist.Path, "symlink_to", MagicMock(side_effect=symlink))
-    monkeypatch.setattr(croaker.playlist.Path, "mkdir", MagicMock())
-
-    pl.add([croaker.path.playlist_root() / p for p in paths], make_theme)
-    assert len(new_symlinks) == expected_count

test/test_streamer.py

@@ -1,132 +0,0 @@
-import io
-import queue
-import threading
-from pathlib import Path
-from unittest.mock import MagicMock
-
-import pytest
-import shout
-
-from croaker import playlist, streamer
-
-
-def get_stream_output(stream):
-    return stream.read()
-
-
-@pytest.fixture(scope="session")
-def silence_bytes():
-    return (Path(streamer.__file__).parent / "silence.mp3").read_bytes()
-
-
-@pytest.fixture
-def output_stream():
-    return io.BytesIO()
-
-
-@pytest.fixture
-def mock_shout(output_stream, monkeypatch):
-    def handle_send(buf):
-        output_stream.write(buf)
-
-    mm = MagicMock(spec=shout.Shout, **{"return_value.send.side_effect": handle_send})
-    monkeypatch.setattr("shout.Shout", mm)
-    return mm
-
-
-@pytest.fixture
-def input_queue():
-    return queue.Queue()
-
-
-@pytest.fixture
-def skip_event():
-    return threading.Event()
-
-
-@pytest.fixture
-def stop_event():
-    return threading.Event()
-
-
-@pytest.fixture
-def load_event():
-    return threading.Event()
-
-
-@pytest.fixture
-def audio_streamer(mock_shout, input_queue, skip_event, stop_event, load_event):
-    return streamer.AudioStreamer(input_queue, skip_event, stop_event, load_event)
-
-
-def test_streamer_stop(audio_streamer, stop_event, output_stream):
-    stop_event.set()
-    audio_streamer.stream_queued_audio()
-    assert not stop_event.is_set()
-
-
-def test_streamer_skip(audio_streamer, skip_event, output_stream):
-    skip_event.set()
-    audio_streamer.stream_queued_audio()
-    assert not skip_event.is_set()
-
-
-def test_streamer_load(audio_streamer, load_event, output_stream):
-    load_event.set()
-    audio_streamer.stream_queued_audio()
-    assert not load_event.is_set()
-
-
-def test_clear_queue(audio_streamer, input_queue):
-    pl = playlist.Playlist(name="test_playlist")
-    for track in pl.tracks:
-        input_queue.put(bytes(track))
-    assert input_queue.not_empty
-    audio_streamer.clear_queue()
-    assert input_queue.empty
-
-
-@pytest.mark.skip
-def test_streamer_defaults_to_silence(audio_streamer, input_queue, output_stream, silence_bytes):
-    audio_streamer.stream_queued_audio()
-    track = playlist.Playlist(name="test_playlist").tracks[0]
-    input_queue.put(bytes(track))
-    audio_streamer.stream_queued_audio()
-    audio_streamer.stream_queued_audio()
-    assert get_stream_output(output_stream) == silence_bytes + track.read_bytes() + silence_bytes
-
-
-@pytest.mark.skip
-def test_streamer_plays_silence_on_error(monkeypatch, audio_streamer, input_queue, output_stream, silence_bytes):
-    monkeypatch.setattr(audio_streamer.queue, "get", MagicMock(side_effect=Exception))
-    track = playlist.Playlist(name="test_playlist").tracks[0]
-    input_queue.put(bytes(track))
-    audio_streamer.stream_queued_audio()
-    assert get_stream_output(output_stream) == silence_bytes
-
-
-@pytest.mark.skip
-def test_streamer_plays_from_queue(audio_streamer, input_queue, output_stream):
-    pl = playlist.Playlist(name="test_playlist")
-    expected = b""
-    for track in pl.tracks:
-        input_queue.put(bytes(track))
-        expected += track.read_bytes()
-    while not input_queue.empty():
-        audio_streamer.stream_queued_audio()
-    assert get_stream_output(output_stream) == expected
-
-
-def test_streamer_handles_stop_interrupt(audio_streamer, output_stream, stop_event):
-    stop_event.set()
-    audio_streamer.stream_queued_audio()
-    assert get_stream_output(output_stream) == b""
-
-
-def test_streamer_handles_load_interrupt(audio_streamer, input_queue, output_stream, load_event):
-    pl = playlist.Playlist(name="test_playlist")
-    input_queue.put(bytes(pl.tracks[0]))
-    load_event.set()
-    audio_streamer.stream_queued_audio()
-    assert get_stream_output(output_stream) == b""
-    assert input_queue.empty

test/test_transcoder.py

@@ -1,43 +0,0 @@
-from unittest.mock import MagicMock
-
-import ffmpeg
-import pytest
-
-from croaker import playlist, transcoder
-
-
-@pytest.fixture
-def mock_mp3decoder(monkeypatch):
-    def read(stream):
-        return stream.read()
-
-    monkeypatch.setattr(transcoder, "MP3Decoder", MagicMock(**{"__enter__.return_value.read": read}))
-
-
-@pytest.mark.xfail
-@pytest.mark.parametrize(
-    "suffix, expected",
-    [
-        (".mp3", b"_theme.mp3\n"),
-        (".foo", b"transcoding!\n"),
-    ],
-)
-def test_transcoder_open(monkeypatch, mock_mp3decoder, suffix, expected):
-    monkeypatch.setattr(
-        transcoder,
-        "ffmpeg",
-        MagicMock(
-            spec=ffmpeg,
-            **{
-                "input.return_value."
-                "output.return_value."
-                "global_args.return_value."
-                "compile.return_value": ["echo", "transcoding!"],
-            },
-        ),
-    )
-
-    pl = playlist.Playlist(name="test_playlist")
-    track = [t for t in pl.tracks if t.suffix == suffix][0]
-    with transcoder.open(track) as handle:
-        assert handle.read() == expected