NAPALM/napalm-sonos
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Fingerprint-Attribute für Discovery-Scoring

Browse Source 
Ergänzt DRIVER_NAME, HTTP_FINGERPRINT, SNMP_FINGERPRINT, SSH_FINGERPRINT,
PORT_SPECS und SNMP_OBJECT_ID_PREFIX gemäß docs/DISCOVERY_FINGERPRINTING.md.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Christian Manivong
2026-06-24 14:47:08 +02:00
co-authored by Claude Sonnet 4.6
parent 6d8f01fe27
commit f6bd38a9fe
1 changed files with 48 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
napalm_sonos/sonos_driver.py
+48 -39
View File
@@ -2,11 +2,11 @@ from __future__ import annotations
import re
import socket
from typing import Any, Dict, List, Optional
from typing import Any
from napalm.base import NetworkDriver
from napalm.base.exceptions import ConnectionException
from napalm_device_types import AccessPointDriver, FingerprintRule, PortSpec
from napalm_device_types.models import DeviceActionResultDict
"""NAPALM driver for Sonos speakers.
@@ -19,7 +19,7 @@ media-playback controls.
_VOLUME_RE = re.compile(r"^\d+$")
class SonosDriver(NetworkDriver):
class SonosDriver(AccessPointDriver):
"""NAPALM driver for controlling Sonos speakers.
Implements the standard NAPALM interface plus media-playback methods
@@ -27,6 +27,15 @@ class SonosDriver(NetworkDriver):
"""
VENDOR = "Sonos"
DRIVER_NAME = "sonos"
PORT_SPECS = [
PortSpec("http", 1400, ("/xml/device_description.xml",), weight=9.0, mandatory=True),
PortSpec("https", 1443, weight=4.0),
]
HTTP_FINGERPRINT = [
FingerprintRule("sonos", weight=8.0, mandatory=True),
FingerprintRule("rincon", weight=5.0),
]
def __init__(
self,
@@ -34,7 +43,7 @@ class SonosDriver(NetworkDriver):
username: str = "",
password: str = "",
timeout: int = 60,
optional_args: Optional[Dict[str, Any]] = None,
optional_args: dict[str, Any] | None = None,
) -> None:
"""Initialise the driver with connection details."""
self.hostname = hostname
@@ -44,7 +53,7 @@ class SonosDriver(NetworkDriver):
self.optional_args = optional_args or {}
self._device: Any = None
self._speaker_info: Optional[Dict[str, Any]] = None
self._speaker_info: dict[str, Any] | None = None
def open(self) -> None:
"""Open a connection to the Sonos speaker."""
@@ -62,7 +71,7 @@ class SonosDriver(NetworkDriver):
self._device = None
self._speaker_info = None
def is_alive(self) -> Dict[str, bool]:
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}
@@ -75,7 +84,7 @@ class SonosDriver(NetworkDriver):
except OSError:
return {"is_alive": False}
def get_facts(self) -> Dict[str, Any]:
def get_facts(self) -> dict[str, Any]:
"""Return device facts gathered during connection."""
if not self._speaker_info:
return {}
@@ -91,8 +100,8 @@ class SonosDriver(NetworkDriver):
"interface_list": self._detect_interfaces(info),
}
def _detect_interfaces(self, info: Dict[str, Any]) -> List[str]:
ifaces: List[str] = []
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")
@@ -112,7 +121,7 @@ class SonosDriver(NetworkDriver):
pass
return False
def get_interfaces(self) -> Dict[str, Dict[str, Any]]:
def get_interfaces(self) -> dict[str, dict[str, Any]]:
"""Return interface information for this speaker."""
if not self._speaker_info:
return {}
@@ -121,7 +130,7 @@ class SonosDriver(NetworkDriver):
is_up = self.is_alive().get("is_alive", False)
has_eth = self._check_wired_connection()
interfaces: Dict[str, Dict[str, Any]] = {}
interfaces: dict[str, dict[str, Any]] = {}
if has_eth:
interfaces["eth0"] = {
"is_up": is_up,
@@ -143,7 +152,7 @@ class SonosDriver(NetworkDriver):
}
return interfaces
def get_interfaces_ip(self) -> Dict[str, Dict[str, Any]]:
def get_interfaces_ip(self) -> dict[str, dict[str, Any]]:
"""Return IP address information for the speaker interfaces."""
if not self._speaker_info:
return {}
@@ -156,45 +165,45 @@ class SonosDriver(NetworkDriver):
"prefix_length": 24,
}
}
interfaces: Dict[str, Any] = {}
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]]:
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]:
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]:
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]:
def get_config(self) -> dict[str, Any]:
"""Return the device config (not supported on Sonos)."""
return {}
def get_environment(self) -> Dict[str, Any]:
def get_environment(self) -> dict[str, Any]:
"""Return environment/power info (not supported on Sonos)."""
return {}
def get_optics(self) -> Dict[str, Any]:
def get_optics(self) -> dict[str, Any]:
"""Return optics info (not supported on Sonos)."""
return {}
def get_ntp_peers(self) -> Dict[str, Optional[str]]:
def get_ntp_peers(self) -> dict[str, str | None]:
"""Return NTP peers (not supported on Sonos)."""
return {}
def get_ntp_servers(self) -> Dict[str, Optional[str]]:
def get_ntp_servers(self) -> dict[str, str | None]:
"""Return NTP servers (not supported on Sonos)."""
return {}
def get_ntp_stats(self) -> List[Dict[str, Any]]:
def get_ntp_stats(self) -> list[dict[str, Any]]:
"""Return NTP stats (not supported on Sonos)."""
return []
@@ -207,7 +216,7 @@ class SonosDriver(NetworkDriver):
size: int = 64,
count: int = 5,
vrf: str = "",
) -> Dict[str, Any]:
) -> dict[str, Any]:
"""Ping a destination (not supported on Sonos)."""
return {
"success": {},
@@ -221,58 +230,58 @@ class SonosDriver(NetworkDriver):
ttl: int = 255,
timeout: int = 2,
vrf: str = "",
) -> Dict[str, Any]:
) -> 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]]:
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]:
def get_snmp_information(self) -> dict[str, Any]:
"""Return SNMP info (not supported on Sonos)."""
return {}
def get_users(self) -> Dict[str, Any]:
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]:
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]:
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]:
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]:
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]:
def get_probes_config(self) -> dict[str, Any]:
"""Return probe config (not supported on Sonos)."""
return {}
def get_probes_results(self) -> Dict[str, Any]:
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]]:
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]:
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]]:
def cli(self, commands: list[str]) -> list[dict[str, Any]]:
"""Run CLI commands (not supported on Sonos)."""
return []
@@ -310,7 +319,7 @@ class SonosDriver(NetworkDriver):
# Broad catch: SoCo transport errors are unpredictable
return "STOPPED"
def get_current_track(self) -> Dict[str, Any]:
def get_current_track(self) -> dict[str, Any]:
"""Return information about the currently playing track."""
if self._device is None:
return {}
@@ -383,7 +392,7 @@ class SonosDriver(NetworkDriver):
return
self._device.previous()
def get_group_info(self) -> Dict[str, Any]:
def get_group_info(self) -> dict[str, Any]:
"""Return information about the Sonos group this speaker belongs to."""
if self._device is None:
return {}
@@ -407,7 +416,7 @@ class SonosDriver(NetworkDriver):
# Broad catch: SoCo group/network operations can fail in many ways
return {}
def get_zone_info(self) -> Dict[str, Any]:
def get_zone_info(self) -> dict[str, Any]:
"""Return detailed zone/speaker information."""
if self._device is None:
return {}