space/custom-streamdeck
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add GoXLR utility plugin

Browse Source
This commit is contained in:
Space
2026-06-09 22:06:31 +02:00
parent bb5571784a
commit 58fe9548a6
4 changed files with 593 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+1
View File
@@ -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
TODO.md
+10
View File
@@ -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.
plugins/goxlr.py
+462
View File
@@ -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()
tests/test_backend.py
+120
View File
@@ -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()