""" OPLabs Mixer — Unix domain socket IPC client. Connects to the C++ MixerDaemon over a Unix domain socket using the daemon's newline-terminated JSON protocol. Protocol: Send: JSON payload + newline ('\\n') Receive: JSON payload + newline ('\\n') Each connection handles one request batch (send → receive → close). """ import asyncio import json import logging from typing import Any, Optional logger = logging.getLogger("mixer-ipc") class MixerIpcError(Exception): """Raised when the mixer daemon returns an error response.""" class MixerIpcClient: """Async Unix socket IPC client for the OPLabs Mixer Daemon. Each request opens a fresh connection to the daemon, sends the command as a newline-terminated JSON line, reads the response, and closes. This matches the daemon's line-based protocol where one connection = one request batch. Usage: client = MixerIpcClient(socket_path="/tmp/oplabs-mixer.sock") await client.connect() state = await client.get_state() await client.set_channel_volume(0, -6.0) await client.close() """ def __init__(self, socket_path: str = "/tmp/oplabs-mixer.sock", connect_timeout: float = 5.0, request_timeout: float = 10.0): self._socket_path = socket_path self._connect_timeout = connect_timeout self._request_timeout = request_timeout self._reader: Optional[asyncio.StreamReader] = None self._writer: Optional[asyncio.StreamWriter] = None self._connected = False self._next_id = 1 # ------------------------------------------------------------------ # Connection lifecycle # ------------------------------------------------------------------ async def connect(self) -> bool: """Probe the Unix domain socket to verify daemon is reachable. Opens a test connection to the daemon and immediately closes it. The daemon's protocol is connection-per-request, so we don't hold a persistent connection. """ try: loop = asyncio.get_running_loop() reader, writer = await asyncio.wait_for( asyncio.open_unix_connection(self._socket_path), timeout=self._connect_timeout, ) # Test succeeded — daemon is listening. Close immediately. writer.close() await writer.wait_closed() self._connected = True logger.info("Connected to mixer daemon at %s", self._socket_path) return True except (FileNotFoundError, ConnectionRefusedError, asyncio.TimeoutError, OSError) as exc: self._connected = False logger.warning("Daemon not reachable at %s: %s", self._socket_path, exc) return False async def close(self): """Close the connection.""" await self._disconnect() async def _disconnect(self): """Close the underlying socket.""" if self._writer: try: self._writer.close() await self._writer.wait_closed() except Exception: pass self._writer = None self._reader = None async def _ensure_connection(self): """Open a fresh connection to the daemon.""" await self._disconnect() try: loop = asyncio.get_running_loop() self._reader, self._writer = await asyncio.wait_for( asyncio.open_unix_connection(self._socket_path), timeout=self._connect_timeout, ) except (FileNotFoundError, ConnectionRefusedError, asyncio.TimeoutError, OSError) as exc: self._writer = None self._reader = None raise MixerIpcError(f"Cannot connect to daemon: {exc}") @property def is_connected(self) -> bool: return self._connected # ------------------------------------------------------------------ # Send / receive (line-based protocol, one-shot connection) # ------------------------------------------------------------------ async def _send_request(self, method: str, params: dict = None) -> Any: """Open connection, send request, wait for response, close. Wire format: ``{"cmd": method, **params}`` (daemon's native format). The daemon expects ``cmd`` as the command field and all parameters at the top level (not nested under a ``params`` key). """ if not self._connected: raise MixerIpcError("Not connected to mixer daemon") # Build daemon-native wire message: {"cmd": method, ...params} msg: dict[str, Any] = {"cmd": method} if params: msg.update(params) # Open fresh connection await self._ensure_connection() try: # Send: JSON payload + newline payload = json.dumps(msg) + "\n" self._writer.write(payload.encode("utf-8")) await self._writer.drain() # Read: one response line response_line = await asyncio.wait_for( self._reader.readline(), timeout=self._request_timeout, ) if not response_line: raise MixerIpcError("Empty response from daemon") response = json.loads(response_line.decode("utf-8").strip()) # Check for error status if isinstance(response, dict): if response.get("status") == "error": raise MixerIpcError(response.get("message", "Unknown error")) return response except asyncio.TimeoutError: raise MixerIpcError(f"Request '{method}' timed out") except json.JSONDecodeError as e: raise MixerIpcError(f"Invalid JSON response: {e}") except MixerIpcError: raise except Exception as e: raise MixerIpcError(f"Request '{method}' failed: {e}") finally: await self._disconnect() # ------------------------------------------------------------------ # High-level API # ------------------------------------------------------------------ # --- State --- async def get_state(self) -> dict: """Get the full mixer state (channels, buses, routes, output routes).""" return await self._send_request("getState") # --- Channel Control --- async def set_channel_volume(self, channel_index: int, volume_db: float): """Set channel volume in dB (-inf to +12).""" await self._send_request("setChannelVolume", { "channel": channel_index, "volume": volume_db, }) async def set_channel_pan(self, channel_index: int, pan: float): """Set channel pan (-1.0 left to +1.0 right).""" await self._send_request("setChannelPan", { "channel": channel_index, "pan": pan, }) async def set_channel_mute(self, channel_index: int, mute: bool): """Set channel mute state.""" await self._send_request("setChannelMute", { "channel": channel_index, "mute": mute, }) async def set_channel_solo(self, channel_index: int, solo: bool): """Set channel solo state.""" await self._send_request("setChannelSolo", { "channel": channel_index, "solo": solo, }) async def set_channel_label(self, channel_index: int, label: str): """Set the label/name for a channel.""" await self._send_request("setChannelLabel", { "channel": channel_index, "label": label, }) # --- Channel Lifecycle --- async def add_channel(self, physical_input_index: int = 0) -> int: """Add a new channel. Returns the channel index.""" result = await self._send_request("addChannel", { "physicalInput": physical_input_index, }) return result.get("channelIndex", -1) async def remove_channel(self, channel_index: int): """Remove a channel by index.""" await self._send_request("removeChannel", { "channel": channel_index, }) # --- Bus Control --- async def set_bus_volume(self, bus_id: int, volume_db: float): """Set bus volume in dB (-inf to +12).""" await self._send_request("setBusVolume", { "busId": bus_id, "volume": volume_db, }) async def set_bus_mute(self, bus_id: int, mute: bool): """Set bus mute state.""" await self._send_request("setBusMute", { "busId": bus_id, "mute": mute, }) # --- Routing --- async def route_channel_to_bus(self, channel_index: int, bus_id: int, level_db: float = 0.0): """Route a channel to a bus with a given level in dB.""" await self._send_request("routeChannelToBus", { "channel": channel_index, "busId": bus_id, "level": level_db, }) async def route_bus_to_bus(self, source_bus_id: int, target_bus_id: int, level_db: float = 0.0): """Route one bus to another bus.""" await self._send_request("routeBusToBus", { "sourceBusId": source_bus_id, "targetBusId": target_bus_id, "level": level_db, }) async def remove_route(self, source_id: int, target_bus_id: int): """Remove a route.""" await self._send_request("removeRoute", { "sourceId": source_id, "targetBusId": target_bus_id, }) # --- Output Routing --- async def get_output_routes(self) -> list[dict]: """Get the output routing table.""" result = await self._send_request("getOutputRoutes") return result.get("routes", []) async def set_output_routes(self, routes: list[dict]): """Set the output routing table. Each route dict: {"sourceBusId": int, "sourceStartChannel": int, "targetStartChannel": int, "channels": int} """ await self._send_request("setOutputRoutes", { "routes": routes, }) # ------------------------------------------------------------------ # Convenience context manager # ------------------------------------------------------------------ class MixerIpcContext: """Async context manager for MixerIpcClient.""" def __init__(self, socket_path: str = "/tmp/oplabs-mixer.sock"): self.client = MixerIpcClient(socket_path=socket_path) async def __aenter__(self) -> MixerIpcClient: await self.client.connect() return self.client async def __aexit__(self, *args): await self.client.close()