From 58fe9548a66443e740d09e8bbe2d30adc6dafd15 Mon Sep 17 00:00:00 2001 From: space Date: Tue, 9 Jun 2026 22:06:31 +0200 Subject: [PATCH] Add GoXLR utility plugin --- README.md | 1 + TODO.md | 10 + plugins/goxlr.py | 462 ++++++++++++++++++++++++++++++++++++++++++ tests/test_backend.py | 120 +++++++++++ 4 files changed, 593 insertions(+) create mode 100644 TODO.md create mode 100644 plugins/goxlr.py diff --git a/README.md b/README.md index e4a3339..1d8cc82 100644 --- a/README.md +++ b/README.md @@ -100,6 +100,7 @@ Pre-shipped plugin actions include: - HTTP Requests: call webhooks, local services, and automation endpoints with templated request data - Media Controls: send media keys and repeated volume adjustments - Clipboard Tools: copy preset text, paste snippets, and transform clipboard text +- GoXLR Utility: control channel volume, faders, mute state, routing, profiles, sampler banks, and raw API requests - OBS Integration: sample OBS-style scene and stream actions for plugin development - WLED: control WLED device power, brightness, colors, effects, pixel ranges, and combined updates diff --git a/TODO.md b/TODO.md new file mode 100644 index 0000000..f4b0018 --- /dev/null +++ b/TODO.md @@ -0,0 +1,10 @@ +## P2 +- [ ] Add live GoXLR device discovery to the plugin UI instead of manual serial entry. +> The plugin can auto-pick the first detected mixer, but multi-device setups would be safer with a dropdown sourced from `GetStatus`. + +- [ ] Add GoXLR state-aware toggle actions backed by websocket or status reads. +> The utility publishes state updates, which would let buttons truly toggle mute and FX states instead of only setting explicit values. + +## P3 +- [ ] Add broader GoXLR action coverage for mic tuning, submix, and lighting. +> The raw request action is the current escape hatch; typed actions for those areas would make setup faster and less error-prone. diff --git a/plugins/goxlr.py b/plugins/goxlr.py new file mode 100644 index 0000000..cf78bf9 --- /dev/null +++ b/plugins/goxlr.py @@ -0,0 +1,462 @@ +from __future__ import annotations + +import json +import urllib.error +import urllib.parse +import urllib.request +from typing import Any + + +def _select_field(field_id: str, label: str, default: str, options: list[tuple[str, str]]) -> dict[str, Any]: + return { + "id": field_id, + "label": label, + "type": "select", + "default": default, + "options": [{"label": option_label, "value": option_value} for option_label, option_value in options], + } + + +def _connection_fields() -> list[dict[str, Any]]: + return [ + { + "id": "base_url", + "label": "GoXLR Base URL", + "type": "url", + "default": "http://127.0.0.1:14564", + "placeholder": "http://127.0.0.1:14564", + }, + { + "id": "device_serial", + "label": "Device Serial", + "type": "text", + "placeholder": "Optional: blank uses the first detected mixer", + }, + { + "id": "timeout", + "label": "Timeout Seconds", + "type": "number", + "default": 5, + }, + ] + + +CHANNEL_OPTIONS = [ + ("Mic", "Mic"), + ("Line In", "LineIn"), + ("Console", "Console"), + ("System", "System"), + ("Game", "Game"), + ("Chat", "Chat"), + ("Sample", "Sample"), + ("Music", "Music"), + ("Headphones", "Headphones"), + ("Mic Monitor", "MicMonitor"), + ("Line Out", "LineOut"), +] + +INPUT_DEVICE_OPTIONS = [ + ("Microphone", "Microphone"), + ("Chat", "Chat"), + ("Music", "Music"), + ("Game", "Game"), + ("Console", "Console"), + ("Line In", "LineIn"), + ("System", "System"), + ("Samples", "Samples"), +] + +OUTPUT_DEVICE_OPTIONS = [ + ("Headphones", "Headphones"), + ("Broadcast Mix", "BroadcastMix"), + ("Chat Mic", "ChatMic"), + ("Sampler", "Sampler"), + ("Line Out", "LineOut"), + ("Stream Mix 2", "StreamMix2"), +] + +FADER_OPTIONS = [("A", "A"), ("B", "B"), ("C", "C"), ("D", "D")] + +MUTE_STATE_OPTIONS = [ + ("Unmuted", "Unmuted"), + ("Muted To X", "MutedToX"), + ("Muted To All", "MutedToAll"), +] + +EFFECT_PRESET_OPTIONS = [ + ("Preset 1", "Preset1"), + ("Preset 2", "Preset2"), + ("Preset 3", "Preset3"), + ("Preset 4", "Preset4"), + ("Preset 5", "Preset5"), + ("Preset 6", "Preset6"), +] + +SAMPLER_BANK_OPTIONS = [("Bank A", "A"), ("Bank B", "B"), ("Bank C", "C")] + +EFFECT_TOGGLE_OPTIONS = [ + ("Main FX", "SetFXEnabled"), + ("Megaphone", "SetMegaphoneEnabled"), + ("Robot", "SetRobotEnabled"), + ("HardTune", "SetHardTuneEnabled"), +] + + +class GoXLRPlugin: + name = "GoXLR Utility" + desc = "Control a GoXLR Utility daemon over its JSON API." + version = "0.1.0" + actions = [ + { + "id": "set_volume", + "name": "Set Channel Volume", + "desc": "Set a GoXLR channel volume from 0 to 255.", + "fields": _connection_fields() + + [ + _select_field("channel", "Channel", "Chat", CHANNEL_OPTIONS), + {"id": "volume", "label": "Volume 0-255", "type": "number", "default": 180}, + ], + }, + { + "id": "set_fader", + "name": "Set Fader Assignment", + "desc": "Assign fader A, B, C, or D to a GoXLR channel.", + "fields": _connection_fields() + + [ + _select_field("fader", "Fader", "A", FADER_OPTIONS), + _select_field("channel", "Channel", "Mic", CHANNEL_OPTIONS), + ], + }, + { + "id": "set_fader_mute", + "name": "Set Fader Mute State", + "desc": "Set a fader mute state.", + "fields": _connection_fields() + + [ + _select_field("fader", "Fader", "A", FADER_OPTIONS), + _select_field("state", "Mute State", "MutedToAll", MUTE_STATE_OPTIONS), + ], + }, + { + "id": "set_cough_mute", + "name": "Set Cough Mute State", + "desc": "Set the cough button mute state.", + "fields": _connection_fields() + [_select_field("state", "Mute State", "MutedToAll", MUTE_STATE_OPTIONS)], + }, + { + "id": "set_effect_state", + "name": "Enable Or Disable Effect", + "desc": "Turn the main FX, Megaphone, Robot, or HardTune effect on or off.", + "fields": _connection_fields() + + [ + _select_field("effect", "Effect", "SetFXEnabled", EFFECT_TOGGLE_OPTIONS), + {"id": "enabled", "label": "Enabled", "type": "boolean", "default": True}, + ], + }, + { + "id": "set_active_effect_preset", + "name": "Set Active Effect Preset", + "desc": "Switch the active GoXLR effect preset.", + "fields": _connection_fields() + [_select_field("preset", "Preset", "Preset1", EFFECT_PRESET_OPTIONS)], + }, + { + "id": "set_active_sampler_bank", + "name": "Set Active Sampler Bank", + "desc": "Switch the active sampler bank.", + "fields": _connection_fields() + [_select_field("bank", "Sampler Bank", "A", SAMPLER_BANK_OPTIONS)], + }, + { + "id": "load_profile", + "name": "Load Device Profile", + "desc": "Load a GoXLR device profile by name.", + "fields": _connection_fields() + + [ + {"id": "profile_name", "label": "Profile Name", "type": "text", "required": True, "placeholder": "Headphones"}, + {"id": "persist", "label": "Persist Profile", "type": "boolean", "default": False}, + ], + }, + { + "id": "load_mic_profile", + "name": "Load Mic Profile", + "desc": "Load a GoXLR microphone profile by name.", + "fields": _connection_fields() + + [ + {"id": "profile_name", "label": "Mic Profile Name", "type": "text", "required": True, "placeholder": "ShureSM7B"}, + {"id": "persist", "label": "Persist Profile", "type": "boolean", "default": False}, + ], + }, + { + "id": "set_monitor_mix", + "name": "Set Monitor Mix", + "desc": "Choose which output mix the monitor bus follows.", + "fields": _connection_fields() + [_select_field("output_device", "Output Device", "Headphones", OUTPUT_DEVICE_OPTIONS)], + }, + { + "id": "set_router", + "name": "Set Router Link", + "desc": "Enable or disable one router path from an input source to an output mix.", + "fields": _connection_fields() + + [ + _select_field("input_device", "Input Device", "Microphone", INPUT_DEVICE_OPTIONS), + _select_field("output_device", "Output Device", "BroadcastMix", OUTPUT_DEVICE_OPTIONS), + {"id": "enabled", "label": "Enabled", "type": "boolean", "default": True}, + ], + }, + { + "id": "send_raw_request", + "name": "Send Raw API Request", + "desc": "Send a raw GoXLR Utility API request body exactly as JSON.", + "fields": _connection_fields() + + [ + { + "id": "payload_json", + "label": "Raw JSON Payload", + "type": "json", + "required": True, + "placeholder": '"GetStatus"\n\nor\n\n{\n "Command": ["ABC123", { "SetFXEnabled": true }]\n}', + } + ], + }, + ] + + def on_load(self, ctx): + ctx.db.add_event("plugin.loaded", {"plugin": self.name}) + + def execute_action(self, ctx, action_id, config, event): + if action_id == "set_volume": + self._send_command( + ctx, + action_id, + config, + {"SetVolume": [self._required_str(config, "channel", "Channel"), self._int_range(config.get("volume", 180), 0, 255, "Volume")]}, + ) + return + + if action_id == "set_fader": + self._send_command( + ctx, + action_id, + config, + { + "SetFader": [ + self._required_str(config, "fader", "Fader"), + self._required_str(config, "channel", "Channel"), + ] + }, + ) + return + + if action_id == "set_fader_mute": + self._send_command( + ctx, + action_id, + config, + { + "SetFaderMuteState": [ + self._required_str(config, "fader", "Fader"), + self._required_str(config, "state", "Mute State"), + ] + }, + ) + return + + if action_id == "set_cough_mute": + self._send_command( + ctx, + action_id, + config, + {"SetCoughMuteState": self._required_str(config, "state", "Mute State")}, + ) + return + + if action_id == "set_effect_state": + effect_command = self._required_str(config, "effect", "Effect") + if effect_command not in {value for _label, value in EFFECT_TOGGLE_OPTIONS}: + raise ValueError(f"Unsupported effect toggle '{effect_command}'.") + self._send_command(ctx, action_id, config, {effect_command: bool(config.get("enabled", True))}) + return + + if action_id == "set_active_effect_preset": + self._send_command( + ctx, + action_id, + config, + {"SetActiveEffectPreset": self._required_str(config, "preset", "Preset")}, + ) + return + + if action_id == "set_active_sampler_bank": + self._send_command( + ctx, + action_id, + config, + {"SetActiveSamplerBank": self._required_str(config, "bank", "Sampler Bank")}, + ) + return + + if action_id == "load_profile": + self._send_command( + ctx, + action_id, + config, + { + "LoadProfile": [ + self._required_str(config, "profile_name", "Profile Name"), + bool(config.get("persist", False)), + ] + }, + ) + return + + if action_id == "load_mic_profile": + self._send_command( + ctx, + action_id, + config, + { + "LoadMicProfile": [ + self._required_str(config, "profile_name", "Mic Profile Name"), + bool(config.get("persist", False)), + ] + }, + ) + return + + if action_id == "set_monitor_mix": + self._send_command( + ctx, + action_id, + config, + {"SetMonitorMix": self._required_str(config, "output_device", "Output Device")}, + ) + return + + if action_id == "set_router": + self._send_command( + ctx, + action_id, + config, + { + "SetRouter": [ + self._required_str(config, "input_device", "Input Device"), + self._required_str(config, "output_device", "Output Device"), + bool(config.get("enabled", True)), + ] + }, + ) + return + + if action_id == "send_raw_request": + payload = self._parse_json(config.get("payload_json", ""), "Raw JSON Payload") + response = self._request(config, payload) + self._log_request(ctx, action_id, config, payload, response) + return + + raise ValueError(f"Unknown GoXLR action '{action_id}'.") + + def _send_command(self, ctx, action_id: str, config: dict[str, Any], command: dict[str, Any]) -> None: + serial = self._resolve_device_serial(config) + payload = {"Command": [serial, command]} + response = self._request(config, payload) + self._log_request(ctx, action_id, config, payload, response, serial=serial) + + def _resolve_device_serial(self, config: dict[str, Any]) -> str: + configured = str(config.get("device_serial", "") or "").strip() + if configured: + return configured + + status = self._request(config, "GetStatus") + mixers = status.get("Status", {}).get("mixers") if isinstance(status, dict) else None + if not isinstance(mixers, dict) or not mixers: + raise RuntimeError("GoXLR status did not include any mixers. Set Device Serial explicitly if needed.") + return next(iter(mixers)) + + def _request(self, config: dict[str, Any], payload: Any) -> Any: + url = f"{self._base_url(config)}/api/command" + data = json.dumps(payload, separators=(",", ":")).encode("utf-8") + request = urllib.request.Request( + url, + data=data, + headers={"Content-Type": "application/json"}, + method="POST", + ) + timeout = self._timeout(config) + try: + with urllib.request.urlopen(request, timeout=timeout) as response: + body = response.read(32768).decode("utf-8", errors="replace") + except urllib.error.HTTPError as exc: + error_body = exc.read(32768).decode("utf-8", errors="replace") + raise RuntimeError(f"GoXLR request failed with {exc.code}: {error_body[:300]}") from exc + except urllib.error.URLError as exc: + raise RuntimeError(f"GoXLR request failed: {exc.reason}") from exc + + try: + parsed = json.loads(body) + except json.JSONDecodeError as exc: + raise RuntimeError("GoXLR response was not valid JSON.") from exc + if isinstance(parsed, dict) and "Error" in parsed: + raise RuntimeError(f"GoXLR request failed: {parsed['Error']}") + return parsed + + def _base_url(self, config: dict[str, Any]) -> str: + base_url = str(config.get("base_url", "http://127.0.0.1:14564") or "").strip() + if not base_url: + raise ValueError("GoXLR Base URL is required.") + if "://" not in base_url: + base_url = f"http://{base_url}" + parsed = urllib.parse.urlparse(base_url) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + raise ValueError("GoXLR Base URL must be an http:// or https:// address.") + return urllib.parse.urlunparse((parsed.scheme, parsed.netloc, parsed.path.rstrip("/"), "", "", "")).rstrip("/") + + def _timeout(self, config: dict[str, Any]) -> float: + return max(1.0, min(30.0, float(config.get("timeout", 5) or 5))) + + def _int_range(self, value: Any, minimum: int, maximum: int, label: str) -> int: + try: + parsed = int(float(str(value).strip())) + except (TypeError, ValueError) as exc: + raise ValueError(f"{label} must be a number.") from exc + if parsed < minimum or parsed > maximum: + raise ValueError(f"{label} must be between {minimum} and {maximum}.") + return parsed + + def _required_str(self, config: dict[str, Any], key: str, label: str) -> str: + value = str(config.get(key, "") or "").strip() + if not value: + raise ValueError(f"{label} is required.") + return value + + def _parse_json(self, value: Any, label: str) -> Any: + raw = str(value or "").strip() + if not raw: + raise ValueError(f"{label} is required.") + try: + return json.loads(raw) + except json.JSONDecodeError as exc: + raise ValueError(f"{label} is invalid JSON: {exc.msg}") from exc + + def _log_request( + self, + ctx, + action_id: str, + config: dict[str, Any], + payload: Any, + response: Any, + *, + serial: str | None = None, + ) -> None: + ctx.db.add_event( + "plugin.goxlr_request", + { + "plugin": self.name, + "action_id": action_id, + "base_url": self._base_url(config), + "device_serial": serial, + "payload": payload, + "response": response, + }, + ) + + +PLUGIN = GoXLRPlugin() diff --git a/tests/test_backend.py b/tests/test_backend.py index 4876869..8a5b2e8 100644 --- a/tests/test_backend.py +++ b/tests/test_backend.py @@ -11,6 +11,7 @@ from backend.services.actions import ActionEngine from backend.services.pico import parse_pico_line from backend.services.plugins import PluginContext, PluginManager from plugins.clipboard_tools import ClipboardToolsPlugin +from plugins.goxlr import GoXLRPlugin from plugins.http_requests import HTTPRequestsPlugin from plugins.wled import WLEDPlugin @@ -371,6 +372,125 @@ def test_wled_mega_update_combines_effect_color_range_and_extra_json(tmp_path: P ] +def test_goxlr_plugin_resolves_first_device_and_sets_volume(tmp_path: Path): + received = [] + + class Handler(BaseHTTPRequestHandler): + def do_POST(self): + length = int(self.headers.get("Content-Length", "0")) + payload = json.loads(self.rfile.read(length).decode("utf-8")) + received.append(payload) + self.send_response(200) + self.end_headers() + if payload == "GetStatus": + self.wfile.write(b'{"Status":{"mixers":{"SERIAL123":{"hardware":{"device_type":"Mini"}}}}}') + return + self.wfile.write(b'"Ok"') + + def log_message(self, *_args): + return None + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + db = Database(tmp_path / "streamdeck.sqlite") + GoXLRPlugin().execute_action( + PluginContext(DummyApp(db)), + "set_volume", + { + "base_url": f"http://127.0.0.1:{server.server_port}", + "channel": "Chat", + "volume": 180, + "timeout": 3, + }, + None, + ) + finally: + server.shutdown() + thread.join(timeout=5) + + assert received == [ + "GetStatus", + {"Command": ["SERIAL123", {"SetVolume": ["Chat", 180]}]}, + ] + + +def test_goxlr_plugin_load_profile_uses_explicit_serial(tmp_path: Path): + received = [] + + class Handler(BaseHTTPRequestHandler): + def do_POST(self): + length = int(self.headers.get("Content-Length", "0")) + received.append(json.loads(self.rfile.read(length).decode("utf-8"))) + self.send_response(200) + self.end_headers() + self.wfile.write(b'"Ok"') + + def log_message(self, *_args): + return None + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + db = Database(tmp_path / "streamdeck.sqlite") + GoXLRPlugin().execute_action( + PluginContext(DummyApp(db)), + "load_profile", + { + "base_url": f"127.0.0.1:{server.server_port}", + "device_serial": "MIXER-42", + "profile_name": "Headphones", + "persist": True, + "timeout": 3, + }, + None, + ) + finally: + server.shutdown() + thread.join(timeout=5) + + assert received == [ + {"Command": ["MIXER-42", {"LoadProfile": ["Headphones", True]}]}, + ] + + +def test_goxlr_plugin_raw_request_surfaces_api_error(tmp_path: Path): + class Handler(BaseHTTPRequestHandler): + def do_POST(self): + self.send_response(200) + self.end_headers() + self.wfile.write(b'{"Error":"No such device"}') + + def log_message(self, *_args): + return None + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + db = Database(tmp_path / "streamdeck.sqlite") + try: + GoXLRPlugin().execute_action( + PluginContext(DummyApp(db)), + "send_raw_request", + { + "base_url": f"http://127.0.0.1:{server.server_port}", + "payload_json": '{"Command":["SERIAL123",{"SetFXEnabled":true}]}', + "timeout": 3, + }, + None, + ) + except RuntimeError as exc: + assert "No such device" in str(exc) + else: + raise AssertionError("Expected GoXLR API error to raise.") + finally: + server.shutdown() + thread.join(timeout=5) + + def test_clipboard_plugin_copy_renders_event_tokens(tmp_path: Path): db = Database(tmp_path / "streamdeck.sqlite") plugin = ClipboardToolsPlugin()