init

.gitignore

@@ -0,0 +1,4 @@
+.venv/
+__pycache__/
+*.egg-info/
+.pytest_cache/

napalm_sonos/__init__.py

@@ -0,0 +1,3 @@
+from napalm_sonos.sonos_driver import SonosDriver
+
+__all__ = ["SonosDriver"]

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

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"]

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"]