diff --git a/modalapi/mod.py b/modalapi/mod.py
index efeedc66..4b21e6d9 100755
--- a/modalapi/mod.py
+++ b/modalapi/mod.py
@@ -27,6 +27,9 @@
import modalapi.pedalboard as Pedalboard
import common.parameter as Parameter
import modalapi.wifi as Wifi
+from modalapi.websocket_bridge import AsyncWebSocketBridge
+from modalapi.ws_protocol import parse_message, LoadingEndMessage, PedalSnapshotMessage, WebSocketMessage
+from modalapi.pedalboard_monitor import FileChangeMonitor, read_pedalboard_bundle
from pistomp.analogmidicontrol import AnalogMidiControl
from pistomp.footswitch import Footswitch
@@ -131,17 +134,30 @@ def __init__(self, audiocard, homedir):
self.current = None # pointer to Current class
self.deep = None # pointer to current Deep class
+ # Stores snapshot index from loading_end until pedalboard change is detected
+ self.next_pedalboard_preset_index = None
+
self.selected_menu_index = 0
self.menu_items = None
self.current_menu = MenuType.MENU_NONE
- # This file is modified when the pedalboard is changed via MOD UI
- self.pedalboard_modification_file = "/home/pistomp/data/last.json"
- self.pedalboard_change_timestamp = os.path.getmtime(self.pedalboard_modification_file)\
- if Path(self.pedalboard_modification_file).exists() else 0
-
+ self.data_dir = "/home/pistomp/data"
+ self.last_json_monitor = FileChangeMonitor(os.path.join(self.data_dir, "last.json"))
self.wifi_manager = Wifi.WifiManager()
+ # WebSocket bridge for MOD-UI communication
+ self.ws_bridge = None
+ try:
+ self.ws_bridge = AsyncWebSocketBridge(
+ ws_url='ws://localhost:80/websocket',
+ backpressure_threshold=8192 # 8 KB
+ )
+ self.ws_bridge.start()
+ logging.info("WebSocket bridge started")
+ except Exception as e:
+ logging.warning(f"Failed to initialize WebSocket bridge: {e}")
+
+
# Callback function map. Key is the user specified name, value is function from this handler
# Used for calling handler callbacks pointed to by names which may be user set in the config file
self.callbacks = {"set_mod_tap_tempo": self.set_mod_tap_tempo,
@@ -153,10 +169,15 @@ def __del__(self):
logging.info("Handler cleanup")
if self.wifi_manager:
del self.wifi_manager
+ if self.ws_bridge is not None:
+ self.ws_bridge.stop()
def cleanup(self):
if self.lcd is not None:
self.lcd.cleanup()
+ if self.ws_bridge is not None:
+ self.ws_bridge.stop()
+ logging.info("WebSocket bridge stopped")
# Container for dynamic data which is unique to the "current" pedalboard
# The self.current pointed above will point to this object which gets
@@ -166,7 +187,7 @@ class Current:
def __init__(self, pedalboard):
self.pedalboard = pedalboard
self.presets = {}
- self.preset_index = 0
+ self.preset_index = 0 # Assumes pedalboard loads at snapshot 0 (default behavior)
self.analog_controllers = {} # { type: (plugin_name, param_name) }
class Deep:
@@ -445,27 +466,46 @@ def poll_wifi(self):
def poll_system_info(self):
pass
- def poll_modui_changes(self):
- # This poll looks for changes made via the MOD UI and tries to sync the pi-Stomp hardware
-
- # Look for a change of pedalboard
- #
- # If the pedalboard_modification_file timestamp has changed, extract the bundle path and set current pedalboard
- #
- # TODO this is an interim solution until better MOD-UI to pi-stomp event communication is added
- #
- if Path(self.pedalboard_modification_file).exists():
- ts = os.path.getmtime(self.pedalboard_modification_file)
- if ts == self.pedalboard_change_timestamp:
- return
+ def _handle_ws_message(self, msg: WebSocketMessage):
+ """Handle incoming WebSocket message from MOD-UI"""
+ if isinstance(msg, LoadingEndMessage):
+ logging.debug(f"WebSocket: Pedalboard loading finished, snapshot={msg.snapshot_id}")
+ self.next_pedalboard_preset_index = msg.snapshot_id
+
+ elif isinstance(msg, PedalSnapshotMessage):
+ if self.next_pedalboard_preset_index is not None:
+ logging.debug(f"WebSocket: Pre-switch snapshot changed to {msg.snapshot_id}")
+ self.next_pedalboard_preset_index = msg.snapshot_id
+ else:
+ assert self.current is not None, "Received snapshot message but no current pedalboard is set"
+ logging.debug(f"WebSocket: Snapshot changed to {msg.snapshot_id} ({msg.snapshot_name})")
- # Timestamp changed
- self.pedalboard_change_timestamp = ts
+ if msg.snapshot_id not in self.current.presets:
+ self.current.presets[msg.snapshot_id] = msg.snapshot_name
+
+ self.current.preset_index = msg.snapshot_id
+ self.update_lcd_title()
+
+ def poll_modui_changes(self):
+ """Poll for changes from MOD-UI: websockets and file watching"""
+ if self.ws_bridge is not None:
+ messages = self.ws_bridge.get_received_messages()
+ for msg in messages:
+ try:
+ self._handle_ws_message(parse_message(msg))
+ except Exception as e:
+ logging.error(f"Error handling WebSocket message '{msg}': {e}")
+
+ # Check for pedalboard change via last.json
+ if self.last_json_monitor.check_for_change():
self.lcd.draw_info_message("Loading...")
- mod_bundle = self.get_pedalboard_bundle_from_mod()
- if mod_bundle:
- logging.info("Pedalboard changed via MOD from: %s to: %s" %
- (self.current.pedalboard.bundle, mod_bundle))
+ mod_bundle = read_pedalboard_bundle(self.last_json_monitor.path)
+ if mod_bundle and mod_bundle != self.current.pedalboard.bundle:
+ logging.info(f"Pedalboard changed via MOD from: {self.current.pedalboard.bundle} to: {mod_bundle}")
+
+ if mod_bundle not in self.pedalboards:
+ self.load_pedalboards()
+
pb = self.pedalboards[mod_bundle]
self.set_current_pedalboard(pb)
@@ -503,19 +543,8 @@ def load_pedalboards(self):
#logging.debug("Preset: %s %d" % (bund, self.host.pedalboard_preset)) # this value not initialized
#logging.debug("Preset: %s" % self.get_current_preset_name())
- def get_pedalboard_bundle_from_mod(self):
- # Assumes the caller has already checked for existence of the file
- mod_bundle = None
- with open(self.pedalboard_modification_file, 'r') as file:
- j = json.load(file)
- mod_bundle = util.DICT_GET(j, 'pedalboard')
- return mod_bundle
-
def get_current_pedalboard_bundle_path(self):
- mod_bundle = None
- if Path(self.pedalboard_modification_file).exists():
- mod_bundle = self.get_pedalboard_bundle_from_mod()
- return mod_bundle
+ return read_pedalboard_bundle(self.last_json_monitor.path)
def set_current_pedalboard(self, pedalboard):
# Delete previous "current"
@@ -524,6 +553,10 @@ def set_current_pedalboard(self, pedalboard):
# Create a new "current"
self.current = self.Current(pedalboard)
+ if self.next_pedalboard_preset_index is not None:
+ self.current.preset_index = self.next_pedalboard_preset_index
+ self.next_pedalboard_preset_index = None
+
# Load Pedalboard specific config (overrides default set during initial hardware init)
config_file = Path(pedalboard.bundle) / "config.yml"
cfg = None
diff --git a/modalapi/modhandler.py b/modalapi/modhandler.py
index 42fb73e0..da83ad62 100755
--- a/modalapi/modhandler.py
+++ b/modalapi/modhandler.py
@@ -13,8 +13,6 @@
# You should have received a copy of the GNU General Public License
# along with pi-stomp. If not, see .
-from pistomp.handler import Handler
-
import json
import logging
import os
@@ -33,6 +31,9 @@
from pistomp.lcd320x240 import Lcd
from pistomp.hardware import Controller, Hardware
import pistomp.settings as Settings
+from modalapi.websocket_bridge import AsyncWebSocketBridge
+from modalapi.ws_protocol import parse_message, LoadingEndMessage, PedalSnapshotMessage, WebSocketMessage
+from modalapi.pedalboard_monitor import FileChangeMonitor, read_pedalboard_bundle
from pistomp.analogmidicontrol import AnalogMidiControl
from pistomp.encodermidicontrol import EncoderMidiControl
@@ -75,6 +76,9 @@ def __init__(self, audiocard, homedir):
self._lcd: Lcd | None = None
self._hardware: Hardware | None = None
+ # Stores snapshot index from loading_end until pedalboard change is detected
+ self.next_pedalboard_preset_index = None
+
# Backup
self.backup_dir = "/media/usb0/backups"
self.backup_file = "pistomp_backup.zip"
@@ -82,17 +86,26 @@ def __init__(self, audiocard, homedir):
# Banks
self.banks_file = os.path.join(self.data_dir, "banks.json")
- self.banks_file_timestamp = os.path.getmtime(self.banks_file) if Path(self.banks_file).exists() else 0
self.banks = {}
self.current_bank = None
- # This file is modified when the pedalboard is changed via MOD UI
- self.pedalboard_modification_file = os.path.join(self.data_dir, "last.json")
- self.pedalboard_change_timestamp = os.path.getmtime(self.pedalboard_modification_file)\
- if Path(self.pedalboard_modification_file).exists() else 0
+ self.last_json_monitor = FileChangeMonitor(os.path.join(self.data_dir, "last.json"))
+ self.banks_monitor = FileChangeMonitor(self.banks_file)
self.wifi_manager = Wifi.WifiManager()
+ # WebSocket bridge for MOD-UI communication
+ self.ws_bridge = None
+ try:
+ self.ws_bridge = AsyncWebSocketBridge(
+ ws_url='ws://localhost:80/websocket',
+ backpressure_threshold=8192 # 8 KB
+ )
+ self.ws_bridge.start()
+ logging.info("WebSocket bridge started")
+ except Exception as e:
+ logging.warning(f"Failed to initialize WebSocket bridge: {e}")
+
# Callback function map. Key is the user specified name, value is function from this handler
# Used for calling handler callbacks pointed to by names which may be user set in the config file
self.callbacks = {"set_mod_tap_tempo": self.set_mod_tap_tempo,
@@ -106,12 +119,17 @@ def __del__(self):
logging.info("Handler cleanup")
if self.wifi_manager:
del self.wifi_manager
+ if self.ws_bridge is not None:
+ self.ws_bridge.stop()
def cleanup(self):
if self._lcd is not None:
self._lcd.cleanup()
if self._hardware is not None:
self._hardware.cleanup()
+ if self.ws_bridge is not None:
+ self.ws_bridge.stop()
+ logging.info("WebSocket bridge stopped")
# Container for dynamic data which is unique to the "current" pedalboard
# The self.current pointed above will point to this object which gets
@@ -121,7 +139,7 @@ class Current:
def __init__(self, pedalboard: Pedalboard.Pedalboard):
self.pedalboard: Pedalboard.Pedalboard = pedalboard
self.presets: dict[int, str] = {}
- self.preset_index: int = 0
+ self.preset_index: int = 0 # Assumes pedalboard loads at snapshot 0 (default behavior)
self.analog_controllers: dict[str, dict[str, Any]] = {} # { type: (plugin_name, param_name) }
def _rest_get(self, url: str) -> Response | None:
@@ -240,36 +258,76 @@ def universal_encoder_sw(self, value, obj=None):
if self._lcd is not None:
self._lcd.enc_sw(value)
+ def _handle_ws_message(self, msg: WebSocketMessage):
+ """Handle incoming WebSocket message from MOD-UI."""
+ if isinstance(msg, LoadingEndMessage):
+ logging.debug(f"WebSocket: Pedalboard loading finished, snapshot={msg.snapshot_id}")
+ # Sometimes mod-ui sends us -1 for preset index, but shows 0 anyway ("Default")
+ self.next_pedalboard_preset_index = max(0, msg.snapshot_id)
+
+ elif isinstance(msg, PedalSnapshotMessage):
+ if self.next_pedalboard_preset_index is not None:
+ # Check if we're still on the same pedalboard (stale flag from previous load)
+ mod_bundle = read_pedalboard_bundle(self.last_json_monitor.path)
+ if mod_bundle and self.current and mod_bundle == self.current.pedalboard.bundle:
+ # Same pedalboard - this is a new snapshot on current board, not a pre-switch
+ logging.debug(f"WebSocket: Snapshot changed to {msg.snapshot_id} ({msg.snapshot_name}) - clearing stale pre-switch flag")
+ self.next_pedalboard_preset_index = None
+
+ if msg.snapshot_id not in self.current.presets:
+ self.current.presets[msg.snapshot_id] = msg.snapshot_name
+
+ self.current.preset_index = msg.snapshot_id
+ self._handle_collage_mode_snapshot_change(msg.snapshot_id)
+ self.lcd.draw_title()
+ else:
+ # Different pedalboard pending - this is a legitimate pre-switch update
+ logging.debug(f"WebSocket: Pre-switch snapshot changed to {msg.snapshot_id}")
+ self.next_pedalboard_preset_index = msg.snapshot_id
+ else:
+ assert self.current is not None, "Received snapshot message but no current pedalboard is set"
+ logging.debug(f"WebSocket: Snapshot changed to {msg.snapshot_id} ({msg.snapshot_name})")
+
+ if msg.snapshot_id not in self.current.presets:
+ self.current.presets[msg.snapshot_id] = msg.snapshot_name
+
+ self.current.preset_index = msg.snapshot_id
+ self.lcd.draw_title()
+
def poll_modui_changes(self):
- # This poll looks for changes made via the MOD UI and tries to sync the pi-Stomp hardware
-
- # Look for a change of pedalboard
- #
- # If the pedalboard_modification_file timestamp has changed, extract the bundle path and set current pedalboard
- #
- # TODO this is an interim solution until better MOD-UI to pi-stomp event communication is added
- #
- if Path(self.pedalboard_modification_file).exists():
- ts = os.path.getmtime(self.pedalboard_modification_file)
- if ts != self.pedalboard_change_timestamp:
- # Timestamp changed
- self.pedalboard_change_timestamp = ts
- self.lcd.draw_info_message("Loading...")
- mod_bundle = self.get_pedalboard_bundle_from_mod()
- if mod_bundle:
- logging.info("Pedalboard changed via MOD from: %s to: %s" %
- (self.current.pedalboard.bundle if self.current else None, mod_bundle))
- pb = self.reload_pedalboard(mod_bundle)
- self.set_current_pedalboard(pb)
+ """Poll for changes from MOD-UI: websockets and file watching"""
+ if self.ws_bridge is not None:
+ messages = self.ws_bridge.get_received_messages()
+ for msg in messages:
+ try:
+ self._handle_ws_message(parse_message(msg))
+ except Exception as e:
+ logging.error(f"Error handling WebSocket message '{msg}': {e}")
+
+ # Check for pedalboard change via last.json
+ if self.last_json_monitor.check_for_change():
+ self.lcd.draw_info_message("Loading...")
+ mod_bundle = read_pedalboard_bundle(self.last_json_monitor.path)
+ if mod_bundle and self.current and mod_bundle != self.current.pedalboard.bundle:
+ logging.info(f"Pedalboard changed via MOD from: {self.current.pedalboard.bundle} to: {mod_bundle}")
+
+ if mod_bundle not in self.pedalboards:
+ self.load_pedalboards()
+
+ pb = self.reload_pedalboard(mod_bundle)
+ self.set_current_pedalboard(pb)
+ elif mod_bundle and self.next_pedalboard_preset_index is not None:
+ # Same pedalboard reloaded with a pending snapshot - apply it now
+ logging.info(f"Applying pending snapshot {self.next_pedalboard_preset_index} to current pedalboard")
+ self.current.preset_index = self.next_pedalboard_preset_index
+ self._handle_collage_mode_snapshot_change(self.next_pedalboard_preset_index)
+ self.next_pedalboard_preset_index = None
+ self.lcd.draw_title()
# Look for a change in banks file
- if Path(self.banks_file).exists():
- ts = os.path.getmtime(self.banks_file)
- if ts != self.banks_file_timestamp:
- # Timestamp changed
- logging.info("Reloading banks file: %s" % self.banks_file)
- self.banks_file_timestamp = ts
- self.load_banks()
+ if self.banks_monitor.check_for_change():
+ logging.info("Reloading banks file: %s" % self.banks_file)
+ self.load_banks()
#
# Bank Stuff
@@ -341,19 +399,8 @@ def reload_pedalboard(self, bundle):
return pedalboard
- def get_pedalboard_bundle_from_mod(self):
- # Assumes the caller has already checked for existence of the file
- mod_bundle = None
- with open(self.pedalboard_modification_file, 'r') as file:
- j = json.load(file)
- mod_bundle = util.DICT_GET(j, 'pedalboard')
- return mod_bundle
-
def get_current_pedalboard_bundle_path(self):
- mod_bundle = None
- if Path(self.pedalboard_modification_file).exists():
- mod_bundle = self.get_pedalboard_bundle_from_mod()
- return mod_bundle
+ return read_pedalboard_bundle(self.last_json_monitor.path)
def set_current_pedalboard(self, pedalboard):
# Delete previous "current"
@@ -362,6 +409,10 @@ def set_current_pedalboard(self, pedalboard):
# Create a new "current"
self.current = self.Current(pedalboard)
+ if self.next_pedalboard_preset_index is not None:
+ self.current.preset_index = self.next_pedalboard_preset_index
+ self.next_pedalboard_preset_index = None
+
# Load Pedalboard specific config (overrides default set during initial hardware init)
config_file = Path(pedalboard.bundle) / "config.yml"
cfg = None
diff --git a/modalapi/pedalboard_monitor.py b/modalapi/pedalboard_monitor.py
new file mode 100644
index 00000000..0487f7d8
--- /dev/null
+++ b/modalapi/pedalboard_monitor.py
@@ -0,0 +1,59 @@
+# This file is part of pi-stomp.
+#
+# pi-stomp is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# pi-stomp is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with pi-stomp. If not, see .
+
+"""
+Monitors a file for modification-time changes.
+
+Used to detect when MOD-UI writes last.json (pedalboard change) or
+banks.json (bank list change) without polling the WebSocket.
+"""
+
+import json
+import logging
+import os
+from pathlib import Path
+from typing import Optional
+
+
+class FileChangeMonitor:
+ """Detects when a file has been modified since the last check."""
+
+ def __init__(self, file_path: str):
+ self.path = file_path
+ self._last_timestamp = self._current_timestamp()
+
+ def _current_timestamp(self) -> float:
+ p = Path(self.path)
+ return os.path.getmtime(self.path) if p.exists() else 0.0
+
+ def check_for_change(self) -> bool:
+ """Return True (and update baseline) if the file was modified since last call."""
+ ts = self._current_timestamp()
+ if ts != self._last_timestamp:
+ self._last_timestamp = ts
+ return True
+ return False
+
+
+def read_pedalboard_bundle(last_json_path: str) -> Optional[str]:
+ """Read the current pedalboard bundle name from last.json."""
+ if not Path(last_json_path).exists():
+ return None
+ try:
+ with open(last_json_path, "r") as f:
+ return json.load(f).get("pedalboard")
+ except (json.JSONDecodeError, IOError) as e:
+ logging.warning(f"Failed to read {last_json_path}: {e}")
+ return None
diff --git a/modalapi/ws_protocol.py b/modalapi/ws_protocol.py
new file mode 100644
index 00000000..3812f4d0
--- /dev/null
+++ b/modalapi/ws_protocol.py
@@ -0,0 +1,179 @@
+# This file is part of pi-stomp.
+#
+# pi-stomp is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# pi-stomp is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with pi-stomp. If not, see .
+
+"""
+Type-safe WebSocket protocol for MOD-UI communication.
+
+Defines message types received from mod-ui WebSocket server.
+"""
+
+from dataclasses import dataclass
+from typing import Union
+import logging
+
+
+@dataclass
+class LoadingStartMessage:
+ """Pedalboard loading started."""
+
+ is_default: bool
+
+
+@dataclass
+class LoadingEndMessage:
+ """Pedalboard loading finished."""
+
+ snapshot_id: int
+
+
+@dataclass
+class PedalSnapshotMessage:
+ """Snapshot changed within current pedalboard."""
+
+ snapshot_id: int
+ snapshot_name: str
+
+
+@dataclass
+class SizeMessage:
+ """Pedalboard canvas size."""
+
+ width: int
+ height: int
+
+
+@dataclass
+class AddHwPortMessage:
+ """Hardware port appeared (JACK)."""
+
+ port_name: str
+ port_type: str # "audio" or "midi"
+ is_output: bool
+ title: str
+ index: int
+
+
+@dataclass
+class RemoveHwPortMessage:
+ """Hardware port disappeared."""
+
+ port_name: str
+
+
+@dataclass
+class TrueBypassMessage:
+ """True bypass state changed."""
+
+ left: int
+ right: int
+
+
+@dataclass
+class UnknownMessage:
+ """Message type we don't handle yet."""
+
+ raw: str
+
+
+# Union of all message types
+WebSocketMessage = Union[
+ LoadingStartMessage,
+ LoadingEndMessage,
+ PedalSnapshotMessage,
+ SizeMessage,
+ AddHwPortMessage,
+ RemoveHwPortMessage,
+ TrueBypassMessage,
+ UnknownMessage,
+]
+
+
+def parse_message(raw_message: str) -> WebSocketMessage:
+ """Parse raw WebSocket message string into typed message object."""
+ try:
+ match raw_message.split(" ", 2):
+ # Format: loading_start {isDefault}
+ case ["loading_start", flag, *_]:
+ return LoadingStartMessage(is_default=bool(int(flag)))
+ case ["loading_start", *_]:
+ return LoadingStartMessage(is_default=False)
+
+ # Format: loading_end {snapshotId}
+ case ["loading_end", sid, *_]:
+ return LoadingEndMessage(snapshot_id=int(sid))
+ case ["loading_end"]:
+ return LoadingEndMessage(snapshot_id=0)
+
+ # Format: pedal_snapshot {snapshotId} {snapshotName}
+ case ["pedal_snapshot", sid, name]:
+ return PedalSnapshotMessage(snapshot_id=int(sid), snapshot_name=name)
+ case ["pedal_snapshot", sid]:
+ return PedalSnapshotMessage(snapshot_id=int(sid), snapshot_name="")
+ case ["pedal_snapshot"]:
+ return PedalSnapshotMessage(snapshot_id=0, snapshot_name="")
+
+ # Format: size {width} {height}
+ case ["size", w, h_trailing]:
+ return SizeMessage(width=int(w), height=int(h_trailing.split()[0]))
+ case ["size", w]:
+ return SizeMessage(width=int(w), height=0)
+ case ["size"]:
+ return SizeMessage(width=0, height=0)
+
+ # Format: add_hw_port /graph/{name} {type} {isOutput} {title} {index}
+ case ["add_hw_port", port_name, rest]:
+ match rest.split(" ", 3):
+ case [port_type, is_out, title, index]:
+ return AddHwPortMessage(
+ port_name=port_name,
+ port_type=port_type,
+ is_output=bool(int(is_out)),
+ title=title,
+ index=int(index),
+ )
+ case [port_type, is_out, title]:
+ return AddHwPortMessage(
+ port_name=port_name, port_type=port_type, is_output=bool(int(is_out)), title=title, index=0
+ )
+ case [port_type, is_out]:
+ return AddHwPortMessage(
+ port_name=port_name, port_type=port_type, is_output=bool(int(is_out)), title="", index=0
+ )
+ case [port_type]:
+ return AddHwPortMessage(
+ port_name=port_name, port_type=port_type, is_output=False, title="", index=0
+ )
+ case ["add_hw_port", port_name]:
+ return AddHwPortMessage(port_name=port_name, port_type="", is_output=False, title="", index=0)
+
+ # Format: remove_hw_port /graph/{name}
+ case ["remove_hw_port", port_name, *_]:
+ return RemoveHwPortMessage(port_name=port_name)
+ case ["remove_hw_port"]:
+ return RemoveHwPortMessage(port_name="")
+
+ # Format: truebypass {left} {right}
+ case ["truebypass", left, right_trailing]:
+ return TrueBypassMessage(left=int(left), right=int(right_trailing.split()[0]))
+ case ["truebypass", left]:
+ return TrueBypassMessage(left=int(left), right=0)
+ case ["truebypass"]:
+ return TrueBypassMessage(left=0, right=0)
+
+ except (ValueError, IndexError) as e:
+ logging.warning(f"Failed to parse WebSocket message '{raw_message}': {e}")
+ return UnknownMessage(raw=raw_message)
+
+ return UnknownMessage(raw=raw_message)