The problem
Speakers on a local network can be controlled on that network. This project explores that route for Sonos, combining a Python API with a small web interface for discovering speakers and controlling playback.
The approach: FastAPI and asyncio handle the API and network requests. UPnP/SOAP talks to the speakers, and HTMX provides the browser controls.
Reported project measurements: Sub-25ms command response latency over local WiFi; 100% cloud-free local network execution; lightweight memory footprint under 35MB RAM on Raspberry Pi.
How it works
Layered Service-Oriented Architecture (SOA): Segregates SOAP transport primitives (BaseSonosClient) from UPnP service clients (AVTransportClient, RenderingControlClient) and high-level audio domain services.
Command / Registry Dispatch Pattern: Centralized dispatch via ACTION_REGISTRY mapping string commands directly to asynchronous lambdas, eliminating verbose endpoint routing trees.
Hypermedia-Driven Architecture (HDA): HTMX-powered frontend with server-rendered Jinja2 HTML fragments, achieving dynamic UI reactivity without large client-side JavaScript bundles.
How the pieces connect
flowchart TD
Client[Browser / HTMX Client] -->|HTTP / Form Data| Router[FastAPI Application Gateway]
subgraph routing["Routing & Middleware"]
Router --> ErrorDecorator["@api_error_handler Decorator"]
Router --> Registry[Action Registry Dispatcher]
end
subgraph serviceLayer["Service Layer"]
Registry --> AVService[AVTransport Client]
Registry --> RenderService[RenderingControl Client]
Router --> ZoneService[Zone & Topology Service]
Router --> RadioService[Radio Service / pyradios]
end
subgraph hardwareIntegration["Hardware Integration"]
AVService -->|SOAP / XML POST| SonosHW[Sonos Speaker - Port 1400]
RenderService -->|SOAP / XML POST| SonosHW
ZoneService -->|SOAP / XML POST| SonosHW
Router -->|SSDP Multicast / UDP 1900| SonosHW
end
Implementation notes
Dynamic SOAP Invocation & Robust XML Extraction (sonos/client.py)
async def _invoke_soap_request(
self, path: str, service_urn: str, action: str, body_content: str = ""
) -> str:
url = f"http://{self.ip}:{self.port}{path}"
soap_action = f"{service_urn}#{action}"
soap_body = (
''
''
""
f''
f"{body_content}"
f""
""
""
)
headers = {
"SOAPAction": f'"{soap_action}"',
"Content-Type": "text/xml; charset=utf-8",
"Accept-Encoding": "gzip",
}
async with aiohttp.ClientSession() as session:
async with session.post(url, headers=headers, data=soap_body.encode("utf-8")) as response:
content = await response.text()
if response.status >= 400:
response.raise_for_status()
return content
Declarative Dynamic Command Routing (sonos/router.py)
ACTION_REGISTRY = {
"setvolume": lambda ip, value: get_rendering_control_client(ip).set_volume(int(value)),
"getvolume": lambda ip, value=None: get_rendering_control_client(ip).get_volume(),
"play": lambda ip, value=None: get_av_transport_client(ip).play(),
"pause": lambda ip, value=None: get_av_transport_client(ip).pause(),
"seek": lambda ip, value=None: get_av_transport_client(ip).seek(value),
"settrack": lambda ip, value: get_av_transport_client(ip).set_av_transport_uri(value),
"status": lambda ip, value=None: get_av_transport_client(ip).get_transport_info(),
}
Tradeoffs and lessons
- Direct UPnP/SOAP Implementation vs. Heavy 3rd-Party SDKs: Implemented a bespoke, lightweight asynchronous client over aiohttp to ensure strict async event-loop compatibility and predictable error boundaries.
- Server-Driven HTMX Swaps vs. Client-Side SPA: Traded client-side JavaScript state machines for HTMX polling (
hx-trigger="every 2s") and partial DOM updates, lowering memory footprint for low-power edge hosting. - SSDP Multicast Discovery with Nmap Fallback: Leveraged UDP SSDP discovery (M-SEARCH) for standard zero-conf resolution, with optional socket port scanning on port 1400 for hardened local networks.