commit 6d8f01fe270b2f2e20862fd9a1aa60925a38c28e Author: Christian Manivong Date: Sat Jun 13 03:18:28 2026 +0200 init diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..64c75f7 --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +.venv/ +__pycache__/ +*.egg-info/ +.pytest_cache/ diff --git a/napalm_sonos/__init__.py b/napalm_sonos/__init__.py new file mode 100644 index 0000000..fd4514c --- /dev/null +++ b/napalm_sonos/__init__.py @@ -0,0 +1,3 @@ +from napalm_sonos.sonos_driver import SonosDriver + +__all__ = ["SonosDriver"] diff --git a/napalm_sonos/sonos_driver.py b/napalm_sonos/sonos_driver.py new file mode 100644 index 0000000..fc914eb --- /dev/null +++ b/napalm_sonos/sonos_driver.py @@ -0,0 +1,478 @@ +from __future__ import annotations + +import re +import socket +from typing import Any, Dict, List, Optional + +from napalm.base import NetworkDriver +from napalm.base.exceptions import ConnectionException + +from napalm_device_types.models import DeviceActionResultDict + +"""NAPALM driver for Sonos speakers. + +Provides network-automation-style access to Sonos speakers via the +SoCo library, exposing NAPALM standard methods alongside Sonos-specific +media-playback controls. +""" + +_VOLUME_RE = re.compile(r"^\d+$") + + +class SonosDriver(NetworkDriver): + """NAPALM driver for controlling Sonos speakers. + + Implements the standard NAPALM interface plus media-playback methods + (play, pause, volume, etc.) using the SoCo library. + """ + + VENDOR = "Sonos" + + def __init__( + self, + hostname: str, + username: str = "", + password: str = "", + timeout: int = 60, + optional_args: Optional[Dict[str, Any]] = None, + ) -> None: + """Initialise the driver with connection details.""" + self.hostname = hostname + self.username = username + self.password = password + self.timeout = timeout + self.optional_args = optional_args or {} + + self._device: Any = None + self._speaker_info: Optional[Dict[str, Any]] = None + + def open(self) -> None: + """Open a connection to the Sonos speaker.""" + try: + from soco import SoCo + + self._device = SoCo(self.hostname) + self._speaker_info = self._device.get_speaker_info() + except Exception as e: + # Broad catch: SoCo can raise a variety of transport/XML errors + raise ConnectionException(f"Cannot connect to Sonos at {self.hostname}: {e}") + + def close(self) -> None: + """Close the connection to the Sonos speaker.""" + self._device = None + self._speaker_info = None + + def is_alive(self) -> Dict[str, bool]: + """Check whether the speaker is reachable on the network.""" + if self._device is None: + return {"is_alive": False} + try: + sock = socket.create_connection( + (self.hostname, 1400), timeout=5 + ) + sock.close() + return {"is_alive": True} + except OSError: + return {"is_alive": False} + + def get_facts(self) -> Dict[str, Any]: + """Return device facts gathered during connection.""" + if not self._speaker_info: + return {} + info = self._speaker_info + return { + "vendor": self.VENDOR, + "model": info.get("model_name", "Unknown"), + "serial_number": info.get("serial_number", ""), + "os_version": info.get("software_version", ""), + "hostname": info.get("player_name", self.hostname), + "fqdn": info.get("ip_address", self.hostname), + "uptime": 0, + "interface_list": self._detect_interfaces(info), + } + + def _detect_interfaces(self, info: Dict[str, Any]) -> List[str]: + ifaces: List[str] = [] + has_eth = self._check_wired_connection() + if has_eth: + ifaces.append("eth0") + ifaces.append("wlan0") + return ifaces + + def _check_wired_connection(self) -> bool: + if self._device is None: + return False + try: + props = self._device.device_properties + if isinstance(props, dict): + wifimode = props.get("WifiMode", "") + return wifimode in ("0", "", None) + except Exception: + # Broad catch: SoCo device_properties may fail for many reasons + pass + return False + + def get_interfaces(self) -> Dict[str, Dict[str, Any]]: + """Return interface information for this speaker.""" + if not self._speaker_info: + return {} + info = self._speaker_info + mac = info.get("mac_address", "") + is_up = self.is_alive().get("is_alive", False) + has_eth = self._check_wired_connection() + + interfaces: Dict[str, Dict[str, Any]] = {} + if has_eth: + interfaces["eth0"] = { + "is_up": is_up, + "is_enabled": True, + "description": "Ethernet", + "last_flapped": -1.0, + "speed": 100, + "mtu": 1500, + "mac_address": mac, + } + interfaces["wlan0"] = { + "is_up": is_up, + "is_enabled": True, + "description": "WiFi", + "last_flapped": -1.0, + "speed": 0, + "mtu": 1500, + "mac_address": mac, + } + return interfaces + + def get_interfaces_ip(self) -> Dict[str, Dict[str, Any]]: + """Return IP address information for the speaker interfaces.""" + if not self._speaker_info: + return {} + info = self._speaker_info + ip = info.get("ip_address", self.hostname) + has_eth = self._check_wired_connection() + + ipv4_entry = { + ip: { + "prefix_length": 24, + } + } + interfaces: Dict[str, Any] = {} + if has_eth: + interfaces["eth0"] = {"ipv4": ipv4_entry, "ipv6": {}} + interfaces["wlan0"] = {"ipv4": ipv4_entry, "ipv6": {}} + return interfaces + + def get_arp_table(self) -> List[Dict[str, Any]]: + """Return the ARP table (not supported on Sonos).""" + return [] + + def get_lldp_neighbors(self) -> Dict[str, Any]: + """Return LLDP neighbors (not supported on Sonos).""" + return {} + + def get_lldp_neighbors_detail(self, interface: str = "") -> Dict[str, Any]: + """Return detailed LLDP neighbors (not supported on Sonos).""" + return {} + + def get_config(self) -> Dict[str, Any]: + """Return the device config (not supported on Sonos).""" + return {} + + def get_environment(self) -> Dict[str, Any]: + """Return environment/power info (not supported on Sonos).""" + return {} + + def get_optics(self) -> Dict[str, Any]: + """Return optics info (not supported on Sonos).""" + return {} + + def get_ntp_peers(self) -> Dict[str, Optional[str]]: + """Return NTP peers (not supported on Sonos).""" + return {} + + def get_ntp_servers(self) -> Dict[str, Optional[str]]: + """Return NTP servers (not supported on Sonos).""" + return {} + + def get_ntp_stats(self) -> List[Dict[str, Any]]: + """Return NTP stats (not supported on Sonos).""" + return [] + + def ping( + self, + destination: str, + source: str = "", + ttl: int = 255, + timeout: int = 2, + size: int = 64, + count: int = 5, + vrf: str = "", + ) -> Dict[str, Any]: + """Ping a destination (not supported on Sonos).""" + return { + "success": {}, + "error": "Not supported on Sonos", + } + + def traceroute( + self, + destination: str, + source: str = "", + ttl: int = 255, + timeout: int = 2, + vrf: str = "", + ) -> Dict[str, Any]: + """Run traceroute (not supported on Sonos).""" + return { + "success": {}, + "error": "Not supported on Sonos", + } + + def get_route_to(self, destination: str) -> List[Dict[str, Any]]: + """Return routing table (not supported on Sonos).""" + return [] + + def get_snmp_information(self) -> Dict[str, Any]: + """Return SNMP info (not supported on Sonos).""" + return {} + + def get_users(self) -> Dict[str, Any]: + """Return user accounts (not supported on Sonos).""" + return {} + + def get_network_instances(self, name: str = "") -> Dict[str, Any]: + """Return network instances / VRFs (not supported on Sonos).""" + return {} + + def get_bgp_config(self, group: str = "") -> Dict[str, Any]: + """Return BGP config (not supported on Sonos).""" + return {} + + def get_bgp_neighbors(self) -> Dict[str, Any]: + """Return BGP neighbors (not supported on Sonos).""" + return {} + + def get_bgp_neighbors_detail(self, neighbor_address: str = "") -> Dict[str, Any]: + """Return detailed BGP neighbor info (not supported on Sonos).""" + return {} + + def get_probes_config(self) -> Dict[str, Any]: + """Return probe config (not supported on Sonos).""" + return {} + + def get_probes_results(self) -> Dict[str, Any]: + """Return probe results (not supported on Sonos).""" + return {} + + def get_mac_address_table(self) -> List[Dict[str, Any]]: + """Return MAC address table (not supported on Sonos).""" + return [] + + def get_firewall_policies(self) -> Dict[str, Any]: + """Return firewall policies (not supported on Sonos).""" + return {} + + def cli(self, commands: List[str]) -> List[Dict[str, Any]]: + """Run CLI commands (not supported on Sonos).""" + return [] + + def commit_config(self) -> None: + """Commit the candidate config (no-op on Sonos).""" + pass + + def discard_config(self) -> None: + """Discard the candidate config (no-op on Sonos).""" + pass + + def load_replace_candidate(self, filename: str = "", config: str = "") -> None: + """Load a replacement config (not supported on Sonos).""" + raise NotImplementedError("Sonos does not support config replacement") + + def load_merge_candidate(self, filename: str = "", config: str = "") -> None: + """Load a merge candidate (not supported on Sonos).""" + raise NotImplementedError("Sonos does not support config merge") + + def compare_config(self) -> str: + """Compare candidate with active config (no-op on Sonos).""" + return "" + + def rollback(self) -> None: + """Roll back to the previous config (no-op on Sonos).""" + pass + + def get_playback_state(self) -> str: + """Return the current playback state (PLAYING, PAUSED_PLAYBACK, STOPPED).""" + if self._device is None: + return "STOPPED" + try: + return self._device.get_playback_state() + except Exception: + # Broad catch: SoCo transport errors are unpredictable + return "STOPPED" + + def get_current_track(self) -> Dict[str, Any]: + """Return information about the currently playing track.""" + if self._device is None: + return {} + try: + return self._device.get_current_track_info() + except Exception: + # Broad catch: SoCo transport errors are unpredictable + return {} + + def get_volume(self) -> int: + """Return the current volume (0-100).""" + if self._device is None: + return 0 + try: + return self._device.volume + except Exception: + # Broad catch: SoCo transport errors are unpredictable + return 0 + + def set_volume(self, volume: int) -> None: + """Set the volume (0-100).""" + if self._device is None: + return + volume = max(0, min(100, volume)) + self._device.volume = volume + + def get_mute(self) -> bool: + """Return whether the speaker is muted.""" + if self._device is None: + return False + try: + return bool(self._device.mute) + except Exception: + # Broad catch: SoCo transport errors are unpredictable + return False + + def set_mute(self, muted: bool) -> None: + """Mute or unmute the speaker.""" + if self._device is None: + return + self._device.mute = muted + + def play(self) -> None: + """Start playback.""" + if self._device is None: + return + self._device.play() + + def pause(self) -> None: + """Pause playback.""" + if self._device is None: + return + self._device.pause() + + def stop(self) -> None: + """Stop playback.""" + if self._device is None: + return + self._device.stop() + + def next_track(self) -> None: + """Skip to the next track.""" + if self._device is None: + return + self._device.next() + + def previous_track(self) -> None: + """Go back to the previous track.""" + if self._device is None: + return + self._device.previous() + + def get_group_info(self) -> Dict[str, Any]: + """Return information about the Sonos group this speaker belongs to.""" + if self._device is None: + return {} + try: + group = self._device.group + coordinator = group.coordinator + members = group.members + return { + "coordinator": coordinator.player_name if coordinator else "", + "coordinator_uid": coordinator.uid if coordinator else "", + "members": [ + { + "name": m.player_name, + "uid": m.uid, + "ip_address": m.ip_address, + } + for m in members + ], + } + except Exception: + # Broad catch: SoCo group/network operations can fail in many ways + return {} + + def get_zone_info(self) -> Dict[str, Any]: + """Return detailed zone/speaker information.""" + if self._device is None: + return {} + try: + info = self._speaker_info or {} + return { + "player_name": info.get("player_name", ""), + "zone_name": info.get("zone_name", ""), + "uid": info.get("uid", ""), + "model_name": info.get("model_name", ""), + "software_version": info.get("software_version", ""), + "hardware_version": info.get("hardware_version", ""), + "serial_number": info.get("serial_number", ""), + "mac_address": info.get("mac_address", ""), + "ip_address": info.get("ip_address", self.hostname), + } + except Exception: + # Broad catch: speaker_info dict access is safe, but keep for safety + return {} + + def run_device_action(self, action: str) -> DeviceActionResultDict: + """Run a device action (play, pause, stop, next_track, previous_track, set_volume:N, toggle_mute).""" + result: DeviceActionResultDict = { + "success": False, + "action": action, + "message": "", + } + try: + if action == "play": + self.play() + result["success"] = True + result["message"] = "Playback started" + elif action == "pause": + self.pause() + result["success"] = True + result["message"] = "Playback paused" + elif action == "stop": + self.stop() + result["success"] = True + result["message"] = "Playback stopped" + elif action == "next_track": + self.next_track() + result["success"] = True + result["message"] = "Skipped to next track" + elif action == "previous_track": + self.previous_track() + result["success"] = True + result["message"] = "Skipped to previous track" + elif action.startswith("set_volume:"): + parts = action.split(":", 1) + if len(parts) == 2 and _VOLUME_RE.match(parts[1]): + vol = int(parts[1]) + self.set_volume(vol) + result["success"] = True + result["message"] = f"Volume set to {vol}" + else: + result["message"] = f"Invalid volume value: {action}" + elif action == "toggle_mute": + current = self.get_mute() + self.set_mute(not current) + result["success"] = True + result["message"] = f"Mute toggled to {not current}" + else: + result["message"] = f"Unknown action: {action}" + except Exception as e: + # Broad catch: action methods wrap SoCo calls that can fail variously + result["message"] = f"Action '{action}' failed: {e}" + return result diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..59efe32 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,54 @@ +[build-system] +requires = ["setuptools>=68", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "napalm-sonos" +version = "0.1.0" +description = "NAPALM driver for Sonos speakers (SoCo)." +readme = "README.md" +license = { text = "Apache-2.0" } +requires-python = ">=3.9" +authors = [ + { name = "Christian Manivong" }, +] +classifiers = [ + "Topic :: Utilities", + "License :: OSI Approved :: Apache Software License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Operating System :: POSIX :: Linux", + "Operating System :: MacOS", +] +dependencies = [ + "napalm>=4.0.0", + "soco>=0.30.0", +] + +[project.optional-dependencies] +dev = [ + "pytest", + "pytest-cov", + "black", + "ruff", +] + +[project.entry-points."napalm.drivers"] +sonos = "napalm_sonos.sonos_driver:SonosDriver" + +[project.urls] +Repository = "https://github.com/napalm-automation-community/napalm-sonos" + +[tool.setuptools.packages.find] +where = ["."] +include = ["napalm_sonos*"] + +[tool.ruff] +line-length = 100 +target-version = "py39" + +[tool.pytest.ini_options] +testpaths = ["tests"] diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_sonos.py b/tests/test_sonos.py new file mode 100644 index 0000000..18ce1fb --- /dev/null +++ b/tests/test_sonos.py @@ -0,0 +1,207 @@ +from unittest.mock import MagicMock, patch + +from napalm.base.exceptions import ConnectionException + +from napalm_sonos import SonosDriver + + +def create_mock_speaker_info(): + return { + "player_name": "Living Room", + "zone_name": "Living Room", + "uid": "RINCON_000E58ABCDE01400", + "model_name": "Sonos One", + "software_version": "15.2", + "hardware_version": "1.20.1", + "serial_number": "000E58ABCDE", + "mac_address": "00:0E:58:AB:CD:EF", + "ip_address": "192.168.1.100", + } + + +class TestSonosDriver: + def setup_method(self): + self.driver = SonosDriver("192.168.1.100") + self.mock_device = MagicMock() + self.mock_device.get_speaker_info.return_value = create_mock_speaker_info() + self.mock_device.volume = 30 + self.mock_device.mute = False + self.mock_device.get_playback_state.return_value = "PLAYING" + self.mock_device.get_current_track_info.return_value = { + "title": "Test Song", + "artist": "Test Artist", + "album": "Test Album", + } + self.driver._device = self.mock_device + self.driver._speaker_info = create_mock_speaker_info() + + def test_open_success(self): + driver = SonosDriver("192.168.1.100") + with patch("soco.SoCo") as mock_soco: + mock_instance = MagicMock() + mock_instance.get_speaker_info.return_value = create_mock_speaker_info() + mock_soco.return_value = mock_instance + driver.open() + assert driver._device is not None + assert driver._speaker_info is not None + assert driver._speaker_info["player_name"] == "Living Room" + + def test_open_failure(self): + driver = SonosDriver("192.168.1.100") + with patch("soco.SoCo") as mock_soco: + mock_soco.side_effect = Exception("Connection refused") + try: + driver.open() + assert False, "Expected ConnectionException" + except ConnectionException: + pass + + def test_close(self): + self.driver.close() + assert self.driver._device is None + assert self.driver._speaker_info is None + + def test_is_alive(self): + with patch("socket.create_connection") as mock_conn: + mock_conn.return_value.__enter__.return_value = MagicMock() + result = self.driver.is_alive() + assert result["is_alive"] is True + + def test_is_alive_no_device(self): + self.driver._device = None + result = self.driver.is_alive() + assert result["is_alive"] is False + + def test_get_facts(self): + facts = self.driver.get_facts() + assert facts["vendor"] == "Sonos" + assert facts["model"] == "Sonos One" + assert facts["serial_number"] == "000E58ABCDE" + assert facts["os_version"] == "15.2" + assert facts["hostname"] == "Living Room" + assert "wlan0" in facts["interface_list"] + + def test_get_facts_empty(self): + self.driver._speaker_info = None + assert self.driver.get_facts() == {} + + def test_get_interfaces(self): + with patch("socket.create_connection") as mock_conn: + mock_conn.return_value.__enter__.return_value = MagicMock() + ifaces = self.driver.get_interfaces() + assert "wlan0" in ifaces + assert ifaces["wlan0"]["is_up"] is True + assert ifaces["wlan0"]["mac_address"] == "00:0E:58:AB:CD:EF" + + def test_get_interfaces_ip(self): + ifaces_ip = self.driver.get_interfaces_ip() + assert "wlan0" in ifaces_ip + assert "192.168.1.100" in ifaces_ip["wlan0"]["ipv4"] + + def test_get_playback_state(self): + state = self.driver.get_playback_state() + assert state == "PLAYING" + + def test_get_volume(self): + vol = self.driver.get_volume() + assert vol == 30 + + def test_set_volume(self): + self.driver.set_volume(50) + assert self.mock_device.volume == 50 + + def test_set_volume_clamped(self): + self.driver.set_volume(200) + assert self.mock_device.volume == 100 + + def test_get_mute(self): + assert self.driver.get_mute() is False + + def test_set_mute(self): + self.driver.set_mute(True) + assert self.mock_device.mute is True + + def test_play(self): + self.driver.play() + self.mock_device.play.assert_called_once() + + def test_pause(self): + self.driver.pause() + self.mock_device.pause.assert_called_once() + + def test_stop(self): + self.driver.stop() + self.mock_device.stop.assert_called_once() + + def test_next_track(self): + self.driver.next_track() + self.mock_device.next.assert_called_once() + + def test_previous_track(self): + self.driver.previous_track() + self.mock_device.previous.assert_called_once() + + def test_get_current_track(self): + track = self.driver.get_current_track() + assert track["title"] == "Test Song" + assert track["artist"] == "Test Artist" + + def test_get_group_info(self): + mock_coordinator = MagicMock() + mock_coordinator.player_name = "Living Room" + mock_coordinator.uid = "RINCON_000E58ABCDE01400" + + mock_member = MagicMock() + mock_member.player_name = "Kitchen" + mock_member.uid = "RINCON_000E58ABCDE01401" + mock_member.ip_address = "192.168.1.101" + + mock_group = MagicMock() + mock_group.coordinator = mock_coordinator + mock_group.members = [mock_coordinator, mock_member] + self.mock_device.group = mock_group + + group = self.driver.get_group_info() + assert group["coordinator"] == "Living Room" + assert len(group["members"]) == 2 + + def test_get_zone_info(self): + zone = self.driver.get_zone_info() + assert zone["player_name"] == "Living Room" + assert zone["model_name"] == "Sonos One" + assert zone["ip_address"] == "192.168.1.100" + + def test_run_device_action_play(self): + result = self.driver.run_device_action("play") + assert result["success"] is True + assert result["action"] == "play" + self.mock_device.play.assert_called_once() + + def test_run_device_action_pause(self): + result = self.driver.run_device_action("pause") + assert result["success"] is True + self.mock_device.pause.assert_called_once() + + def test_run_device_action_stop(self): + result = self.driver.run_device_action("stop") + assert result["success"] is True + self.mock_device.stop.assert_called_once() + + def test_run_device_action_set_volume(self): + result = self.driver.run_device_action("set_volume:42") + assert result["success"] is True + assert self.mock_device.volume == 42 + + def test_run_device_action_set_volume_invalid(self): + result = self.driver.run_device_action("set_volume:abc") + assert result["success"] is False + + def test_run_device_action_toggle_mute(self): + result = self.driver.run_device_action("toggle_mute") + assert result["success"] is True + assert self.mock_device.mute is True + + def test_run_device_action_unknown(self): + result = self.driver.run_device_action("foobar") + assert result["success"] is False + assert "Unknown" in result["message"]