Play custom radio URL on Sonos. Uses old-school UPnP AVTransport API which may go away.

play_radio.py

#!/usr/bin/env python3
"""
sonos_set_radio.py

Set a custom radio stream on a Sonos coordinator via UPnP AVTransport SetAVTransportURI.

Examples:
  python3 play_radio.py --name "NTS 1" --url "http://stream-relay-geo.ntslive.net/stream"
  python3 play_radio.py --ip 192.168.1.42 --name "My station" --url "http://example.com/stream.mp3"

Inspired by [a comment](https://en.community.sonos.com/controllers-and-music-services-229131/tunein-no-longer-supported-custom-urls-no-more-6878679/index3.html)
from the Phonos app dev.

Vibe-coded with Codex.
"""

from __future__ import annotations

import argparse
import html
import socket
import sys
import urllib.request
import xml.etree.ElementTree as ET


SSDP_ADDR = ("239.255.255.250", 1900)
SSDP_MX = 2
SSDP_ST = "urn:schemas-upnp-org:device:ZonePlayer:1"


def ssdp_discover(timeout_s: float = 2.5) -> list[str]:
    msg = "\r\n".join(
        [
            "M-SEARCH * HTTP/1.1",
            f"HOST: {SSDP_ADDR[0]}:{SSDP_ADDR[1]}",
            'MAN: "ssdp:discover"',
            f"MX: {SSDP_MX}",
            f"ST: {SSDP_ST}",
            "",
            "",
        ]
    ).encode("utf-8")

    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
    sock.settimeout(timeout_s)
    # allow multicast TTL 2
    sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 2)

    found_locations: set[str] = set()

    sock.sendto(msg, SSDP_ADDR)
    while True:
        try:
            data, _addr = sock.recvfrom(65535)
        except socket.timeout:
            break

        text = data.decode("utf-8", errors="ignore")
        for line in text.split("\r\n"):
            if line.lower().startswith("location:"):
                loc = line.split(":", 1)[1].strip()
                found_locations.add(loc)

    sock.close()
    return sorted(found_locations)


def fetch_device_info(location_url: str) -> tuple[str, str]:
    """
    Returns (friendlyName, baseURL host/ip).
    """
    with urllib.request.urlopen(location_url, timeout=3) as resp:
        xml_bytes = resp.read()

    root = ET.fromstring(xml_bytes)
    ns = {"d": "urn:schemas-upnp-org:device-1-0"}
    name_el = root.find(".//d:device/d:friendlyName", ns)
    urlbase_el = root.find(".//d:URLBase", ns)

    friendly = name_el.text.strip() if name_el is not None and name_el.text else "Unknown Sonos"
    if urlbase_el is not None and urlbase_el.text:
        base = urlbase_el.text.strip()
    else:
        # fall back: derive from location (http://ip:port/...)
        base = location_url.split("/", 3)[:3]
        base = "/".join(base) + "/"

    return friendly, base


def build_didl(name: str, url: str) -> str:
    # forum snippet uses SecurityElement.Escape; html.escape is equivalent for XML escaping
    esc_name = html.escape(name, quote=True)
    esc_url = html.escape(url, quote=True)

    return (
        '<DIDL-Lite xmlns:dc="http://purl.org/dc/elements/1.1/" '
        'xmlns:upnp="urn:schemas-upnp-org:metadata-1-0/upnp/" '
        'xmlns="urn:schemas-upnp-org:metadata-1-0/DIDL-Lite/">'
        '<item id="" restricted="false">'
        f"<dc:title>{esc_name}</dc:title>"
        f"<res>{esc_url}</res>"
        "</item>"
        "</DIDL-Lite>"
    )


def soap_call(
    endpoint_url: str,
    service_urn: str,
    action: str,
    body_inner_xml: str,
    timeout_s: float = 5.0,
) -> str:
    soap = (
        '<?xml version="1.0" encoding="utf-8"?>'
        '<s:Envelope xmlns:s="http://schemas.xmlsoap.org/soap/envelope/" '
        's:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/">'
        "<s:Body>"
        f'<u:{action} xmlns:u="{service_urn}">'
        f"{body_inner_xml}"
        f"</u:{action}>"
        "</s:Body>"
        "</s:Envelope>"
    ).encode("utf-8")

    req = urllib.request.Request(endpoint_url, data=soap, method="POST")
    req.add_header("Content-Type", 'text/xml; charset="utf-8"')
    req.add_header("SOAPAction", f'"{service_urn}#{action}"')
    req.add_header("Connection", "close")

    with urllib.request.urlopen(req, timeout=timeout_s) as resp:
        return resp.read().decode("utf-8", errors="ignore")


def set_radio(coordinator_ip: str, name: str, url: str) -> None:
    av_transport_control = f"http://{coordinator_ip}:1400/MediaRenderer/AVTransport/Control"
    service = "urn:schemas-upnp-org:service:AVTransport:1"

    didl = build_didl(name, url)
    # Sonos expects x-rincon-mp3radio:// + raw stream URL
    transport_uri = f"x-rincon-mp3radio://{url}"

    body = (
        "<InstanceID>0</InstanceID>"
        f"<CurrentURI>{html.escape(transport_uri, quote=True)}</CurrentURI>"
        f"<CurrentURIMetaData>{html.escape(didl, quote=True)}</CurrentURIMetaData>"
    )

    print(f"Setting on {coordinator_ip} …")
    soap_call(av_transport_control, service, "SetAVTransportURI", body)

    play_body = "<InstanceID>0</InstanceID><Speed>1</Speed>"
    result = soap_call(av_transport_control, service, "Play", play_body)
    print(result)

    print("Done.")


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--ip", help="Coordinator speaker IP (skip discovery)")
    ap.add_argument("--name", required=True, help="Station name")
    ap.add_argument("--url", required=True, help="Stream URL (http(s) to mp3/aac stream)")
    args = ap.parse_args()

    if args.ip:
        set_radio(args.ip, args.name, args.url)
        return 0

    locations = ssdp_discover()
    if not locations:
        print("No Sonos devices found via SSDP. Try passing --ip.", file=sys.stderr)
        return 2

    devices: list[tuple[str, str]] = []
    for loc in locations:
        try:
            friendly, base = fetch_device_info(loc)
            # base like http://IP:1400/
            hostport = base.split("//", 1)[1].split("/", 1)[0]
            host = hostport.split(":", 1)[0]
            devices.append((friendly, host))
        except Exception:
            continue

    if not devices:
        print("Found Sonos responses but couldn’t parse device descriptions. Try --ip.", file=sys.stderr)
        return 2

    print("Select the coordinator (group leader) IP to control:")
    for i, (friendly, host) in enumerate(devices, 1):
        print(f"  {i}. {friendly} [{host}]")

    choice = input(f"Enter 1-{len(devices)}: ").strip()
    try:
        idx = int(choice) - 1
        coordinator_ip = devices[idx][1]
    except Exception:
        print("Invalid selection.", file=sys.stderr)
        return 2

    set_radio(coordinator_ip, args.name, args.url)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())