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()