docs: update AccessPointDriver MRO placement comment

The comment said AccessPointDriver sits BEFORE mixins, which was wrong after
the mixin refactor. NetworkDriver (parent) raises NotImplementedError for
standard NAPALM methods, so AccessPointDriver must come LAST in the MRO to
avoid shadowing mixin implementations.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

README.md

@@ -38,6 +38,7 @@ Requires Python ≥ 3.9 and NAPALM ≥ 4.0.
 | `SwitchDriver` | Ethernet switches | Cisco IOS, Arista EOS, Juniper EX |
 | `FirewallDriver` | Firewalls & UTM appliances | pfSense, Fortinet FortiOS, Cisco ASA |
 | `HypervisorDriver` | Hypervisors & virtualisation platforms | Proxmox VE, VMware ESXi, KVM/libvirt |
+| `OSDriver` | General-purpose operating systems | Linux, BSD, macOS |
 | `StorageDriver` | Storage appliances & NAS/SAN | TrueNAS, Synology DSM, QNAP QTS |
 
 ## Usage
@@ -105,6 +106,34 @@ class ProxmoxDriver(HypervisorDriver):
         ...
 ```
 
+### OS / Linux
+
+```python
+from napalm_device_types import OSDriver
+
+class LinuxDriver(OSDriver):
+
+    def get_packages(self):
+        # return List[PackageDict]
+        ...
+
+    def get_services(self):
+        # return List[ServiceDict]
+        ...
+
+    def get_users(self):
+        # return List[UserDict]
+        ...
+
+    def get_processes(self):
+        # return List[ProcessDict]
+        ...
+
+    def get_cron_jobs(self):
+        # return List[CronJobDict]
+        ...
+```
+
 ### Storage / NAS
 
 ```python

napalm_device_types/__init__.py

@@ -19,19 +19,37 @@ Available base classes:
 * :class:`~napalm_device_types.switch.SwitchDriver`
 * :class:`~napalm_device_types.firewall.FirewallDriver`
 * :class:`~napalm_device_types.hypervisor.HypervisorDriver`
+* :class:`~napalm_device_types.os.OSDriver`
 * :class:`~napalm_device_types.storage.StorageDriver`
+* :class:`~napalm_device_types.residential_gateway.ResidentialGatewayDriver`
+
+Also provided:
+
+* :class:`~napalm_device_types.config_lifecycle.ConfigLifecycleMixin` --
+  stand-alone mixin to reduce duplication of config lifecycle methods across
+  drivers.
 """
 
+from napalm_device_types.base import DeviceTypeDriver, FingerprintRule, PortSpec
 from napalm_device_types.access_point import AccessPointDriver
+from napalm_device_types.config_lifecycle import ConfigLifecycleMixin
 from napalm_device_types.firewall import FirewallDriver
 from napalm_device_types.hypervisor import HypervisorDriver
+from napalm_device_types.os import OSDriver
+from napalm_device_types.residential_gateway import ResidentialGatewayDriver
 from napalm_device_types.storage import StorageDriver
 from napalm_device_types.switch import SwitchDriver
 
 __all__ = [
     "AccessPointDriver",
+    "ConfigLifecycleMixin",
+    "DeviceTypeDriver",
+    "FingerprintRule",
     "FirewallDriver",
     "HypervisorDriver",
+    "OSDriver",
+    "PortSpec",
+    "ResidentialGatewayDriver",
     "StorageDriver",
     "SwitchDriver",
 ]

napalm_device_types/_ucd_metrics.py

@@ -0,0 +1,165 @@
+"""Shared SNMP metric collection helpers for UCD-MIB and IF-MIB."""
+from __future__ import annotations
+
+import asyncio
+import re
+from typing import Awaitable, Callable, Dict, Optional
+
+# ── UCD-MIB OIDs ──────────────────────────────────────────────────────────────
+OID_SYS_UPTIME  = "1.3.6.1.2.1.25.1.1.0"      # hrSystemUptime (centiseconds)
+OID_CPU_IDLE    = "1.3.6.1.4.1.2021.11.11.0"   # UCD ssCpuIdle (%)
+OID_MEM_TOTAL   = "1.3.6.1.4.1.2021.4.5.0"     # memTotalReal (kB)
+OID_MEM_FREE    = "1.3.6.1.4.1.2021.4.6.0"     # memAvailReal (kB)
+OID_MEM_BUFFER  = "1.3.6.1.4.1.2021.4.14.0"    # memBuffer (kB)
+OID_MEM_CACHED  = "1.3.6.1.4.1.2021.4.15.0"    # memCached (kB)
+OID_SWAP_TOTAL  = "1.3.6.1.4.1.2021.4.3.0"     # memTotalSwap (kB)
+OID_SWAP_AVAIL  = "1.3.6.1.4.1.2021.4.4.0"     # memAvailSwap (kB)
+OID_LOAD_1      = "1.3.6.1.4.1.2021.10.1.3.1"  # laLoad 1-min
+OID_LOAD_5      = "1.3.6.1.4.1.2021.10.1.3.2"  # laLoad 5-min
+OID_LOAD_15     = "1.3.6.1.4.1.2021.10.1.3.3"  # laLoad 15-min
+
+# ── IF-MIB OIDs ───────────────────────────────────────────────────────────────
+OID_IF_DESCR   = "1.3.6.1.2.1.2.2.1.2"
+OID_IF_SPEED   = "1.3.6.1.2.1.2.2.1.5"
+OID_IF_IN_OCT  = "1.3.6.1.2.1.2.2.1.10"
+OID_IF_OUT_OCT = "1.3.6.1.2.1.2.2.1.16"
+OID_IF_IN_ERR  = "1.3.6.1.2.1.2.2.1.14"
+OID_IF_OUT_ERR = "1.3.6.1.2.1.2.2.1.20"
+
+IF_SKIP_DEFAULT = re.compile(r'^(lo|sit\d|tun\d|docker|veth|br-|virbr)', re.IGNORECASE)
+
+SnmpGetFn  = Callable[[str], Awaitable[Optional[str]]]
+SnmpWalkFn = Callable[[str], Awaitable[Dict[str, str]]]
+
+
+def ticks_to_seconds(raw: str | None) -> int | None:
+    if raw is None:
+        return None
+    try:
+        return int(raw) // 100
+    except (ValueError, TypeError):
+        return None
+
+
+def build_if_metrics(
+    metrics: dict,
+    descr: Dict[str, str],
+    speed: Dict[str, str],
+    in_oct: Dict[str, str],
+    out_oct: Dict[str, str],
+    in_err: Dict[str, str],
+    out_err: Dict[str, str],
+    *,
+    if_skip: re.Pattern | None = None,
+    tx_err_is_drop: bool = False,
+) -> None:
+    if if_skip is None:
+        if_skip = IF_SKIP_DEFAULT
+    interfaces: dict = {}
+    for idx, name in descr.items():
+        if if_skip.match(name):
+            continue
+        iface: dict = {"name": name}
+        try:
+            iface["rx_bytes"]  = int(in_oct.get(idx, 0)  or 0)
+            iface["tx_bytes"]  = int(out_oct.get(idx, 0) or 0)
+            iface["rx_errors"] = int(in_err.get(idx, 0)  or 0)
+            tx_val = int(out_err.get(idx, 0) or 0)
+            if tx_err_is_drop:
+                iface["tx_errors"] = 0
+                iface["tx_drops"]  = tx_val
+            else:
+                iface["tx_errors"] = tx_val
+            spd = speed.get(idx)
+            iface["speed_mbps"] = int(spd) // 1_000_000 if spd and int(spd) > 0 else 0
+        except (ValueError, TypeError):
+            pass
+        interfaces[name] = iface
+    if interfaces:
+        metrics["interfaces"] = interfaces
+
+
+async def collect_ucd_metrics(
+    snmp_get: SnmpGetFn,
+    snmp_walk: SnmpWalkFn,
+    *,
+    tx_err_is_drop: bool = False,
+    if_skip: re.Pattern | None = None,
+) -> dict:
+    """Collect UCD-MIB system metrics + IF-MIB interface counters in parallel."""
+    (
+        uptime_raw, cpu_idle_raw,
+        mem_total_raw, mem_free_raw, mem_buf_raw, mem_cache_raw,
+        swap_total_raw, swap_avail_raw,
+        load1_raw, load5_raw, load15_raw,
+    ), (descr, speed, in_oct, out_oct, in_err, out_err) = await asyncio.gather(
+        asyncio.gather(
+            snmp_get(OID_SYS_UPTIME),
+            snmp_get(OID_CPU_IDLE),
+            snmp_get(OID_MEM_TOTAL),
+            snmp_get(OID_MEM_FREE),
+            snmp_get(OID_MEM_BUFFER),
+            snmp_get(OID_MEM_CACHED),
+            snmp_get(OID_SWAP_TOTAL),
+            snmp_get(OID_SWAP_AVAIL),
+            snmp_get(OID_LOAD_1),
+            snmp_get(OID_LOAD_5),
+            snmp_get(OID_LOAD_15),
+        ),
+        asyncio.gather(
+            snmp_walk(OID_IF_DESCR),
+            snmp_walk(OID_IF_SPEED),
+            snmp_walk(OID_IF_IN_OCT),
+            snmp_walk(OID_IF_OUT_OCT),
+            snmp_walk(OID_IF_IN_ERR),
+            snmp_walk(OID_IF_OUT_ERR),
+        ),
+    )
+
+    metrics: dict = {}
+
+    secs = ticks_to_seconds(uptime_raw)
+    if secs is not None:
+        metrics["uptime_seconds"] = secs
+
+    if cpu_idle_raw is not None:
+        try:
+            metrics["cpu_percent"] = round(100.0 - float(cpu_idle_raw), 1)
+        except ValueError:
+            pass
+
+    if mem_total_raw and mem_free_raw:
+        try:
+            total_kb = int(mem_total_raw)
+            free_kb  = int(mem_free_raw)
+            buf_kb   = int(mem_buf_raw)   if mem_buf_raw   else 0
+            cache_kb = int(mem_cache_raw) if mem_cache_raw else 0
+            used_kb  = max(0, total_kb - free_kb - buf_kb - cache_kb)
+            metrics["memory_total_bytes"] = total_kb * 1024
+            metrics["memory_used_bytes"]  = used_kb  * 1024
+            metrics["memory_percent"]     = round(used_kb / total_kb * 100, 1) if total_kb else 0.0
+            metrics["memory_free_bytes"]  = free_kb  * 1024
+        except ValueError:
+            pass
+
+    if swap_total_raw and swap_avail_raw:
+        try:
+            stotal = int(swap_total_raw)
+            savail = int(swap_avail_raw)
+            sused  = stotal - savail
+            metrics["swap_total_bytes"] = stotal * 1024
+            metrics["swap_used_bytes"]  = sused  * 1024
+            metrics["swap_percent"]     = round(sused / stotal * 100, 1) if stotal else 0.0
+        except ValueError:
+            pass
+
+    for key, raw in [("load_1", load1_raw), ("load_5", load5_raw), ("load_15", load15_raw)]:
+        if raw is not None:
+            try:
+                metrics[key] = float(raw)
+            except ValueError:
+                pass
+
+    build_if_metrics(metrics, descr, speed, in_oct, out_oct, in_err, out_err,
+                     if_skip=if_skip, tx_err_is_drop=tx_err_is_drop)
+    return metrics

napalm_device_types/access_point.py

@@ -11,23 +11,29 @@ Usage::
 """
 
 from typing import Any, Dict, List
-from napalm.base import NetworkDriver
+from napalm_device_types.base import DeviceTypeDriver
+from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
 from napalm_device_types.models import (
+    ChannelScanEntryDict,
     Dot1XConfigDict,
     FastTransitionConfigDict,
+    HealthMetricsDict,
     MACACLDict,
     MeshConfigDict,
     MeshPeerDict,
     PackageDict,
     RadioStatusDict,
+    ServiceDict,
     SSIDBridgeDict,
     SSIDDict,
+    UpdateDict,
     WirelessClientDict,
     WirelessConfigDict,
 )
 
 
-class AccessPointDriver(NetworkDriver):
+class AccessPointDriver(DeviceTypeDriver):
+    TYPE_LABEL: str = "Access Point"
     """
     Abstract intermediate driver for wireless access points.
 
@@ -35,111 +41,41 @@ class AccessPointDriver(NetworkDriver):
     access-point-specific operations that concrete drivers must implement.
     """
 
-    def get_wireless_clients(self) -> List[WirelessClientDict]:
-        """
-        Returns a list of wireless clients currently associated with this
-        access point.
+    # Interfaces that carry no operational meaning on an access point and
+    # should be excluded from get_interfaces() / get_facts() interface_list.
+    _EXCLUDED_INTERFACES: frozenset = frozenset({"lo"})
 
-        Each entry contains:
+    # Interface name *prefixes* to exclude (e.g. Linux phy* are raw radio
+    # devices and have no IP/Ethernet significance at the AP level).
+    _EXCLUDED_INTERFACE_PREFIXES: tuple = ("phy",)
 
-        * mac (string) - client MAC address
-        * ssid (string) - SSID the client is connected to
-        * radio (string) - radio identifier (e.g. ``"radio0"``, ``"5GHz"``)
-        * signal (int) - received signal strength in dBm
-        * noise (int) - noise floor in dBm
-        * tx_rate (float) - TX bitrate in Mbit/s
-        * rx_rate (float) - RX bitrate in Mbit/s
-        * uptime (int) - association duration in seconds
+    _SNMP_SKIP_IF = IF_SKIP_DEFAULT
+    _SNMP_TX_ERR_IS_DROP: bool = False
 
-        Example::
+    @classmethod
+    async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict:
+        return await collect_ucd_metrics(
+            snmp_get, snmp_walk,
+            tx_err_is_drop=cls._SNMP_TX_ERR_IS_DROP,
+            if_skip=cls._SNMP_SKIP_IF,
+        )
 
-            [
-                {
-                    "mac": "AA:BB:CC:DD:EE:FF",
-                    "ssid": "MyNetwork",
-                    "radio": "radio1",
-                    "signal": -65,
-                    "noise": -95,
-                    "tx_rate": 300.0,
-                    "rx_rate": 144.0,
-                    "uptime": 3600,
-                }
-            ]
-        """
-        raise NotImplementedError
+    def _filter_interfaces(self, interfaces: Dict[str, Any]) -> Dict[str, Any]:
+        """Remove loopback and radio-device (phy*) interfaces from an interface dict."""
+        return {
+            name: data
+            for name, data in interfaces.items()
+            if name not in self._EXCLUDED_INTERFACES
+            and not name.startswith(self._EXCLUDED_INTERFACE_PREFIXES)
+        }
 
-    def get_ssids(self) -> Dict[str, SSIDDict]:
-        """
-        Returns the configured SSIDs (VAPs) on this access point.
-
-        Keys are SSID names.  Each value contains:
-
-        * enabled (bool) - whether the SSID is currently broadcasting
-        * radio (string) - radio the SSID is bound to
-        * bssid (string) - BSSID (MAC) of the VAP
-        * encryption (string) - e.g. ``"WPA2-PSK"``, ``"WPA3-SAE"``, ``"open"``
-        * hidden (bool) - whether the SSID is hidden
-        * clients (int) - number of currently associated clients
-
-        Example::
-
-            {
-                "MyNetwork": {
-                    "enabled": True,
-                    "radio": "radio1",
-                    "bssid": "AA:BB:CC:DD:EE:F0",
-                    "encryption": "WPA2-PSK",
-                    "hidden": False,
-                    "clients": 3,
-                },
-                "GuestNet": {
-                    "enabled": True,
-                    "radio": "radio0",
-                    "bssid": "AA:BB:CC:DD:EE:F1",
-                    "encryption": "WPA2-PSK",
-                    "hidden": False,
-                    "clients": 1,
-                },
-            }
-        """
-        raise NotImplementedError
-
-    def get_radio_status(self) -> Dict[str, RadioStatusDict]:
-        """
-        Returns the status of each radio interface.
-
-        Keys are radio identifiers (e.g. ``"radio0"``, ``"radio1"``).
-        Each value contains:
-
-        * enabled (bool) - whether the radio is active
-        * band (string) - frequency band, e.g. ``"2.4GHz"``, ``"5GHz"``, ``"6GHz"``
-        * channel (int) - operating channel number
-        * channel_width (int) - channel width in MHz (e.g. 20, 40, 80, 160)
-        * tx_power (int) - transmit power in dBm
-        * frequency (float) - center frequency in MHz
-
-        Example::
-
-            {
-                "radio0": {
-                    "enabled": True,
-                    "band": "2.4GHz",
-                    "channel": 6,
-                    "channel_width": 20,
-                    "tx_power": 20,
-                    "frequency": 2437.0,
-                },
-                "radio1": {
-                    "enabled": True,
-                    "band": "5GHz",
-                    "channel": 36,
-                    "channel_width": 80,
-                    "tx_power": 23,
-                    "frequency": 5180.0,
-                },
-            }
-        """
-        raise NotImplementedError
+    # AP-specific methods (get_wireless_clients, get_ssids, get_radio_status,
+    # get_interfaces, get_vlans, …) are intentionally NOT defined here.
+    # AccessPointDriver sits AFTER the mixin classes in the MRO of concrete
+    # drivers (e.g. OpenWrtDriver(InterfaceMixin, VlanMixin, …, AccessPointDriver)).
+    # NetworkDriver (a parent of AccessPointDriver) raises NotImplementedError for
+    # all standard NAPALM methods, so AccessPointDriver must come LAST to avoid
+    # shadowing the mixin implementations.
 
     def get_wireless_config(self) -> WirelessConfigDict:
         """
@@ -498,6 +434,67 @@ class AccessPointDriver(NetworkDriver):
         """
         raise NotImplementedError
 
+    def get_services(self) -> List[ServiceDict]:
+        """
+        Returns the list of system services known to the device's init system.
+
+        Each entry contains:
+
+        * name (string)    - service name as registered with the init system
+        * running (bool)   - ``True`` if the service process is currently running
+        * enabled (bool)   - ``True`` if the service starts automatically at boot
+        * pid (int)        - process ID of the main service process; 0 if not running
+
+        Example::
+
+            [
+                {"name": "lldpd",   "running": True,  "enabled": True,  "pid": 2341},
+                {"name": "sshd",    "running": True,  "enabled": True,  "pid": 1198},
+                {"name": "cron",    "running": False, "enabled": False, "pid": 0},
+            ]
+        """
+        raise NotImplementedError
+
+    def manage_service(self, name: str, action: str) -> Dict[str, Any]:
+        """
+        Execute a lifecycle action on a named service.
+
+        :param name: Service name as returned by :meth:`get_services`.
+        :param action: One of ``start``, ``stop``, ``restart``, ``enable``, ``disable``.
+        :returns: ``{"success": bool, "output": str}``
+        :raises ValueError: If ``name`` or ``action`` is invalid.
+        :raises NotImplementedError: If the driver does not support service management.
+        """
+        raise NotImplementedError
+
+    def get_available_updates(self) -> List[UpdateDict]:
+        """
+        Returns the list of installed packages that have a newer version available.
+
+        Uses the local package manager cache — does not run ``opkg update`` / ``apk update``.
+
+        :returns: List of :class:`~napalm_device_types.models.UpdateDict`.
+        :raises NotImplementedError: If the driver does not support update listing.
+
+        Example::
+
+            [
+                {"name": "busybox",  "current_version": "1.36.1-1", "new_version": "1.37.0-1"},
+                {"name": "dropbear", "current_version": "2022.83-2", "new_version": "2024.86-1"},
+            ]
+        """
+        raise NotImplementedError
+
+    def apply_updates(self, packages: List[str]) -> Dict[str, Any]:
+        """
+        Upgrade one or more packages to their newest available version.
+
+        :param packages: List of package names to upgrade.
+        :returns: ``{"success": bool, "output": str}``
+        :raises NotImplementedError: If the driver does not support package upgrades.
+        """
+        raise NotImplementedError
+
     def set_package_config(self, name: str, config: Dict[str, Any]) -> None:
         """
         Writes a new configuration for an installed package.

napalm_device_types/base.py

@@ -0,0 +1,80 @@
+"""
+DeviceTypeDriver — common base for all netOrk device-type drivers.
+
+Provides the fingerprinting contract used by the discovery subsystem to
+identify devices from SNMP / SSH / HTTP probe data without requiring a
+live connection.
+"""
+
+from __future__ import annotations
+
+from typing import NamedTuple
+
+from napalm.base import NetworkDriver
+
+
+class FingerprintRule(NamedTuple):
+    """Single pattern-matching rule for device fingerprinting.
+
+    pattern   — substring matched against lowercased probe data
+    weight    — evidence added to the driver's score on match
+    mandatory — pattern absent → driver immediately disqualified
+    negative  — pattern present → weight subtracted instead of added
+    """
+
+    pattern: str
+    weight: float = 1.0
+    mandatory: bool = False
+    negative: bool = False
+
+
+class PortSpec(NamedTuple):
+    """Port to probe during discovery and its fingerprinting contribution.
+
+    scheme    — "http", "https", or "ssh"
+    port      — TCP port number
+    paths     — URL paths to try (http/https only; ignored for ssh)
+    weight    — evidence added to score when this port responds
+    mandatory — port silent → driver immediately disqualified
+    """
+
+    scheme: str
+    port: int
+    paths: tuple[str, ...] = ("/",)
+    weight: float = 5.0
+    mandatory: bool = False
+
+
+class DeviceTypeDriver(NetworkDriver):
+    """Common base for all netOrk device-type drivers.
+
+    Sits between napalm.base.NetworkDriver and the type-specific abstract
+    classes (FirewallDriver, SwitchDriver, …). Adds the fingerprinting
+    interface consumed by the discovery subsystem; does not implement any
+    NAPALM abstract methods.
+
+    Override these class attributes in each concrete driver:
+
+    VENDOR                Human-readable vendor name ("AVM", "OPNsense", …).
+    DRIVER_NAME           NAPALM entry-point key ("fritzbox", "opnsense", …).
+    PORT_SPECS            Non-standard ports to probe beyond 80/443/22.
+                          None = no driver-specific ports expected.
+    SNMP_OBJECT_ID_PREFIX OID prefix for sysObjectID (1.3.6.1.2.1.1.2.0).
+                          Match → +15.0 score; set but no match → disqualified.
+    SNMP_FINGERPRINT      Rules matched against sysDescr (lowercased).
+    SSH_FINGERPRINT       Rules matched against SSH version string +
+                          pre-auth banner (lowercased, space-joined).
+    HTTP_FINGERPRINT      Rules matched against page title + Server header +
+                          body snippet (lowercased, whitespace-normalised).
+    """
+
+    VENDOR: str = ""
+    DRIVER_NAME: str = ""
+    PORT_SPECS: list[PortSpec] | None = None
+    SNMP_OBJECT_ID_PREFIX: str | None = None
+    SNMP_FINGERPRINT: list[FingerprintRule] = []
+    SSH_FINGERPRINT: list[FingerprintRule] = []
+    HTTP_FINGERPRINT: list[FingerprintRule] = []
+    OUI_PREFIXES: list[str] = []
+    # Format: "AA:BB:CC" — first 3 octets of MAC, uppercase, colon-separated.
+    # A match contributes fixed weight 6.0 to the fingerprint score.

napalm_device_types/config_lifecycle.py

@@ -0,0 +1,113 @@
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+import difflib
+from typing import ClassVar
+
+from napalm.base.exceptions import (
+    CommandErrorException,
+    MergeConfigException,
+    ReplaceConfigException,
+)
+
+
+class ConfigLifecycleMixin:
+    """Mixin providing the standard NAPALM config lifecycle for CLI-driven drivers.
+
+    Provides a complete config lifecycle: ``load_merge_candidate``,
+    ``load_replace_candidate``, ``compare_config``, ``discard_config``,
+    ``has_pending_commit``, and a basic ``rollback`` that re-stages the
+    last backup and calls ``commit_config``.
+
+    Concrete drivers **must** provide:
+
+    * ``_get_running_config()`` -- return the current running config as a
+      string (e.g. ``show running-config`` or ``uci export``)
+    * ``commit_config()`` -- vendor-specific apply-and-persist logic
+
+    Override ``rollback()`` if the device needs a smarter revert (e.g.
+    block-diff rollback or config-import-based recovery).
+
+    Config-state attributes (``_candidate_config``, ``_candidate_mode``,
+    ``_backup_config``) are annotated here for type checkers; they default
+    to ``None`` and are written by the lifecycle methods.
+    """
+
+    _candidate_config: str | None = None
+    _candidate_mode: str | None = None
+    _backup_config: str | None = None
+
+    _comment_chars: ClassVar[tuple[str, ...]] = ("!", "#")
+
+    def _get_running_config(self) -> str:
+        raise NotImplementedError
+
+    def commit_config(self, message: str = "", revert_in: int | None = None) -> None:
+        raise NotImplementedError
+
+    def load_merge_candidate(
+        self, filename: str | None = None, config: str | None = None
+    ) -> None:
+        if filename is not None:
+            try:
+                with open(filename) as fh:
+                    config = fh.read()
+            except OSError as exc:
+                raise MergeConfigException(str(exc)) from exc
+        if config is None:
+            raise MergeConfigException("Either 'filename' or 'config' must be provided.")
+        self._candidate_config = config
+        self._candidate_mode = "merge"
+
+    def load_replace_candidate(
+        self, filename: str | None = None, config: str | None = None
+    ) -> None:
+        if filename is not None:
+            try:
+                with open(filename) as fh:
+                    config = fh.read()
+            except OSError as exc:
+                raise ReplaceConfigException(str(exc)) from exc
+        if config is None:
+            raise ReplaceConfigException("Either 'filename' or 'config' must be provided.")
+        self._candidate_config = config
+        self._candidate_mode = "replace"
+
+    def compare_config(self) -> str:
+        if self._candidate_config is None:
+            return ""
+
+        if self._candidate_mode == "merge":
+            lines = []
+            for line in self._candidate_config.splitlines():
+                if line.strip() and not line.strip().startswith(self._comment_chars):
+                    lines.append(f"+{line}")
+            return "\n".join(lines)
+
+        running = self._get_running_config()
+        diff = difflib.unified_diff(
+            running.splitlines(),
+            self._candidate_config.splitlines(),
+            fromfile="running-config",
+            tofile="candidate-config",
+            lineterm="",
+        )
+        return "\n".join(diff)
+
+    def discard_config(self) -> None:
+        self._candidate_config = None
+        self._candidate_mode = None
+
+    def has_pending_commit(self) -> bool:
+        return self._candidate_config is not None
+
+    def rollback(self) -> None:
+        if self._backup_config is None:
+            raise CommandErrorException(
+                "No backup configuration available – "
+                "commit_config has not been called in this session."
+            )
+        self._candidate_config = self._backup_config
+        self._candidate_mode = "merge"
+        self.commit_config()
+        self._backup_config = None

napalm_device_types/firewall.py

@@ -11,8 +11,10 @@ Usage::
 """
 
 from typing import Any, Dict, List
-from napalm.base import NetworkDriver
+from napalm_device_types.base import DeviceTypeDriver
+from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
 from napalm_device_types.models import (
+    HealthMetricsDict,
     NATTranslationDict,
     PackageDict,
     SecurityZoneDict,
@@ -21,7 +23,8 @@ from napalm_device_types.models import (
 )
 
 
-class FirewallDriver(NetworkDriver):
+class FirewallDriver(DeviceTypeDriver):
+    TYPE_LABEL: str = "Firewall"
     """
     Abstract intermediate driver for firewall/security devices.
 
@@ -30,6 +33,18 @@ class FirewallDriver(NetworkDriver):
     that concrete drivers must implement.
     """
 
+    _SNMP_SKIP_IF = IF_SKIP_DEFAULT
+    # OPNsense reports drops in the out-error counter.
+    _SNMP_TX_ERR_IS_DROP: bool = True
+
+    @classmethod
+    async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict:
+        return await collect_ucd_metrics(
+            snmp_get, snmp_walk,
+            tx_err_is_drop=cls._SNMP_TX_ERR_IS_DROP,
+            if_skip=cls._SNMP_SKIP_IF,
+        )
+
     def get_nat_translations(self) -> List[NATTranslationDict]:
         """
         Returns a list of active NAT translation entries.

napalm_device_types/hypervisor.py

@@ -11,8 +11,10 @@ Usage::
 """
 
 from typing import Any, Dict, List
-from napalm.base import NetworkDriver
+from napalm_device_types.base import DeviceTypeDriver
+from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
 from napalm_device_types.models import (
+    HealthMetricsDict,
     PackageDict,
     SnapshotDict,
     StorageVolumeDict,
@@ -22,7 +24,8 @@ from napalm_device_types.models import (
 )
 
 
-class HypervisorDriver(NetworkDriver):
+class HypervisorDriver(DeviceTypeDriver):
+    TYPE_LABEL: str = "Hypervisor"
     """
     Abstract intermediate driver for hypervisors and virtualisation platforms
     (e.g. Proxmox VE, VMware ESXi, KVM/libvirt, Hyper-V).
@@ -31,6 +34,17 @@ class HypervisorDriver(NetworkDriver):
     hypervisor-specific operations that concrete drivers must implement.
     """
 
+    _SNMP_SKIP_IF = IF_SKIP_DEFAULT
+    _SNMP_TX_ERR_IS_DROP: bool = False
+
+    @classmethod
+    async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict:
+        return await collect_ucd_metrics(
+            snmp_get, snmp_walk,
+            tx_err_is_drop=cls._SNMP_TX_ERR_IS_DROP,
+            if_skip=cls._SNMP_SKIP_IF,
+        )
+
     # ------------------------------------------------------------------
     # Virtual machines – read
     # ------------------------------------------------------------------

napalm_device_types/models.py

@@ -6,7 +6,7 @@ abstract device-type driver classes in this package.
 """
 
 from typing import Dict, List, Optional
-from typing_extensions import TypedDict
+from typing_extensions import NotRequired, TypedDict
 
 
 # ---------------------------------------------------------------------------
@@ -50,6 +50,23 @@ class PackageDict(TypedDict):
     source: str
 
 
+class ServiceDict(TypedDict):
+    """A system service managed by the device's init system (e.g. procd on OpenWrt)."""
+
+    name: str
+    running: bool
+    enabled: bool
+    pid: int  # 0 if not running
+
+
+class UpdateDict(TypedDict):
+    """A software package that has a newer version available in the package repository."""
+
+    name: str
+    current_version: str
+    new_version: str
+
+
 # ---------------------------------------------------------------------------
 # Access Point
 # ---------------------------------------------------------------------------
@@ -64,6 +81,10 @@ class WirelessClientDict(TypedDict):
     tx_rate: float
     rx_rate: float
     uptime: int
+    # Optional fields populated by DHCP cross-reference (e.g. from firewall)
+    ip: NotRequired[str]
+    hostname: NotRequired[str]
+    lease_end: NotRequired[int]  # Unix timestamp when DHCP lease expires
 
 
 class SSIDDict(TypedDict):
@@ -84,6 +105,18 @@ class RadioStatusDict(TypedDict):
     frequency: float
 
 
+class ChannelScanEntryDict(TypedDict):
+    """One neighboring AP entry as returned by get_channel_scan()."""
+
+    bssid: str
+    ssid: str
+    frequency: int      # centre frequency in MHz, e.g. 2437
+    channel: int        # channel number, e.g. 6
+    signal_dbm: int     # RSSI in dBm, e.g. -72
+    channel_width: int  # MHz: 20 / 40 / 80 / 160; 0 if unknown
+    band: str           # "2.4GHz" | "5GHz" | "6GHz"
+
+
 class WirelessConfigDict(TypedDict):
     country_code: str
     regulatory_domain: str
@@ -289,6 +322,44 @@ class VPNTunnelDict(TypedDict):
     uptime: int
     bytes_in: int
     bytes_out: int
+    description: NotRequired[str]  # human-readable tunnel description / name
+
+
+# ---------------------------------------------------------------------------
+# Residential Gateway
+# ---------------------------------------------------------------------------
+
+
+class WANStatusDict(TypedDict):
+    connection_type: str  # e.g. "DSL", "Cable", "PPPoE", "DHCP"
+    is_connected: bool
+    external_ip: str
+    uptime: int
+    bytes_sent: int
+    bytes_received: int
+    max_bitrate_up: int  # kbit/s
+    max_bitrate_down: int  # kbit/s
+    external_ipv6: NotRequired[str]
+    link_status: NotRequired[str]  # physical line state, e.g. "Up" / "Down"
+
+
+class PortForwardDict(TypedDict):
+    name: str
+    protocol: str  # "TCP" or "UDP"
+    external_port: int
+    internal_ip: str
+    internal_port: int
+    enabled: bool
+    remote_host: NotRequired[str]  # restrict forward to a specific remote source
+
+
+class HostDict(TypedDict):
+    mac: str
+    ip: str
+    hostname: str
+    interface_type: str  # "LAN", "WLAN", ...
+    is_active: bool
+    lease_time_remaining: NotRequired[int]  # seconds
 
 
 # ---------------------------------------------------------------------------
@@ -475,3 +546,161 @@ class ReplicationJobDict(TypedDict):
     last_run: float    # Unix epoch; 0.0 if never run
     last_status: str   # "success", "failed", "running", "pending"
     bytes_sent: int    # bytes transferred in the last run
+
+
+# ---------------------------------------------------------------------------
+# OS / General-purpose Linux
+# ---------------------------------------------------------------------------
+
+
+class UserDict(TypedDict):
+    """A local OS user account."""
+
+    username: str
+    uid: int
+    gid: int
+    home: str
+    shell: str
+    groups: List[str]   # all supplementary group names
+
+
+class ProcessDict(TypedDict):
+    """A running OS process."""
+
+    pid: int
+    ppid: int
+    user: str
+    cpu: float          # CPU utilisation 0.0–100.0
+    memory: float       # RSS percentage of total RAM 0.0–100.0
+    vsz: int            # virtual memory size in KiB
+    rss: int            # resident set size in KiB
+    tty: str            # controlling terminal; empty string if none
+    state: str          # "R" running, "S" sleeping, "D" uninterruptible, "Z" zombie, etc.
+    started: str        # start time string as reported by ``ps`` (e.g. "12:34" or "May28")
+    command: str        # full command line
+
+
+class CronJobDict(TypedDict):
+    """A scheduled cron task."""
+
+    user: str           # owner of the crontab entry
+    schedule: str       # five-field cron expression, e.g. "0 * * * *"
+    command: str        # shell command to execute
+    description: NotRequired[str]  # optional inline comment
+
+
+class ApplyUpdatesResultDict(TypedDict):
+    """Result of an ``apply_updates()`` call."""
+
+    success: bool       # True if the package manager exited without error
+    output: str         # raw stdout/stderr captured from the package manager
+    error: NotRequired[str]  # error message if success is False
+
+
+class DockerContainerDict(TypedDict):
+    """A Docker container entry as returned by ``docker ps -a``."""
+
+    id: str             # short container ID
+    name: str           # container name(s)
+    image: str          # image reference
+    image_version: str  # OCI org.opencontainers.image.version label; empty if absent
+    command: str        # entrypoint / command
+    created: str        # creation timestamp string
+    status: str         # human-readable status (e.g. "Up 3 hours")
+    ports: str          # port mapping string
+    state: str          # "running", "exited", "paused", etc.
+
+
+class DockerImageDict(TypedDict):
+    """A local Docker image entry as returned by ``docker images``."""
+
+    id: str             # short image ID
+    repository: str     # image repository
+    tag: str            # image tag
+    size: str           # human-readable size string (e.g. "187MB")
+    created: str        # creation timestamp string
+    version: str        # OCI org.opencontainers.image.version label; empty if absent
+
+
+class DockerVolumeDict(TypedDict):
+    """A Docker volume entry as returned by ``docker volume ls``."""
+
+    name: str           # volume name
+    driver: str         # volume driver (e.g. "local")
+    mountpoint: str     # host filesystem path
+    scope: str          # "local" or "global"
+
+
+class DockerNetworkDict(TypedDict):
+    """A Docker network entry as returned by ``docker network ls``."""
+
+    id: str             # short network ID
+    name: str           # network name
+    driver: str         # network driver (e.g. "bridge", "host", "overlay")
+    scope: str          # network scope
+    ipv6: str           # "true" if IPv6 is enabled, "false" otherwise
+    internal: str       # "true" if the network is internal, "false" otherwise
+
+
+class DockerInfoDict(TypedDict):
+    """Return value of ``get_docker_info()``."""
+
+    available: bool                              # False if Docker is absent/inaccessible
+    permission_denied: NotRequired[bool]         # True when socket access is denied
+    version: NotRequired[str]                    # Docker Engine version string
+    containers: NotRequired[List[DockerContainerDict]]
+    images: NotRequired[List[DockerImageDict]]
+    volumes: NotRequired[List[DockerVolumeDict]]
+    networks: NotRequired[List[DockerNetworkDict]]
+    outdated_images: NotRequired[List[str]]      # image names with a newer remote digest
+
+
+class DeviceActionResultDict(TypedDict):
+    """Return value of ``run_device_action()``."""
+
+    success: bool   # True if the action completed without error
+    output: str     # human-readable output or status message
+
+
+class SNMPConfigDict(TypedDict):
+    """SNMP agent configuration detected on the device."""
+
+    running: bool           # True if the SNMP daemon is active
+    community: str          # read community string (e.g. "public")
+    port: int               # listening port (default 161)
+    version: str            # highest supported version: "1", "2c", or "3"
+
+
+# ---------------------------------------------------------------------------
+# Health Metrics  (returned by get_health_metrics())
+# ---------------------------------------------------------------------------
+
+
+class HealthMetricsIfaceDict(TypedDict):
+    """Per-interface counters inside a HealthMetricsDict."""
+
+    name: str
+    rx_bytes: NotRequired[int]
+    tx_bytes: NotRequired[int]
+    rx_errors: NotRequired[int]
+    tx_errors: NotRequired[int]
+    tx_drops: NotRequired[int]   # populated instead of tx_errors when the driver reports drops
+    speed_mbps: NotRequired[int]
+
+
+class HealthMetricsDict(TypedDict):
+    """Return value of ``get_health_metrics()``."""
+
+    uptime_seconds: NotRequired[int]
+    cpu_percent: NotRequired[float]
+    memory_total_bytes: NotRequired[int]
+    memory_used_bytes: NotRequired[int]
+    memory_free_bytes: NotRequired[int]
+    memory_percent: NotRequired[float]
+    swap_total_bytes: NotRequired[int]
+    swap_used_bytes: NotRequired[int]
+    swap_percent: NotRequired[float]
+    load_1: NotRequired[float]
+    load_5: NotRequired[float]
+    load_15: NotRequired[float]
+    interfaces: NotRequired[Dict[str, HealthMetricsIfaceDict]]

napalm_device_types/os.py

@@ -0,0 +1,429 @@
+"""
+Abstract base class for OS / general-purpose server drivers.
+
+Usage::
+
+    from napalm_device_types import OSDriver
+
+    class LinuxDriver(OSDriver):
+        def get_packages(self):
+            ...
+"""
+
+import re
+from typing import List, Optional
+from napalm_device_types.base import DeviceTypeDriver
+from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
+from napalm_device_types.models import (
+    ApplyUpdatesResultDict,
+    CronJobDict,
+    DeviceActionResultDict,
+    DockerInfoDict,
+    HealthMetricsDict,
+    PackageDict,
+    ProcessDict,
+    ServiceDict,
+    SNMPConfigDict,
+    UpdateDict,
+    UserDict,
+)
+
+
+class OSDriver(DeviceTypeDriver):
+    TYPE_LABEL: str = "OS"
+    """
+    Abstract intermediate driver for general-purpose operating systems
+    (e.g. Linux, BSD, macOS).
+
+    Inherits all standard NAPALM NetworkDriver methods and adds OS-specific
+    operations that concrete drivers must implement.
+    """
+
+    # Interfaces matching this pattern are excluded from health-metric collection.
+    _SNMP_SKIP_IF: re.Pattern = IF_SKIP_DEFAULT
+    # Set True for drivers where the out-error counter actually reports drops (e.g. OPNsense).
+    _SNMP_TX_ERR_IS_DROP: bool = False
+
+    @classmethod
+    async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict:
+        """Collect SNMP health metrics (CPU, memory, uptime, load, interfaces).
+
+        :param snmp_get:  async callable ``(oid: str) -> Optional[str]``
+        :param snmp_walk: async callable ``(oid: str) -> Dict[str, str]``
+        """
+        return await collect_ucd_metrics(
+            snmp_get, snmp_walk,
+            tx_err_is_drop=cls._SNMP_TX_ERR_IS_DROP,
+            if_skip=cls._SNMP_SKIP_IF,
+        )
+
+    # ------------------------------------------------------------------
+    # Package management
+    # ------------------------------------------------------------------
+
+    def get_packages(self) -> List[PackageDict]:
+        """
+        Returns a list of all installed software packages.
+
+        Each entry contains:
+
+        * name (string) - package name
+        * version (string) - installed version string
+        * installed (bool) - always ``True`` for this method
+        * description (string) - short package description
+        * size (int) - installed size in bytes; ``0`` if unavailable
+        * source (string) - package source / repository name; empty string if unavailable
+
+        Example::
+
+            [
+                {
+                    "name": "openssh-server",
+                    "version": "1:9.2p1-2+deb12u2",
+                    "installed": True,
+                    "description": "secure shell (SSH) server, for secure access from remote machines",
+                    "size": 524288,
+                    "source": "Debian",
+                },
+            ]
+        """
+        raise NotImplementedError
+
+    def get_pending_updates(self) -> List[UpdateDict]:
+        """
+        Returns a list of packages that have a newer version available.
+
+        Each entry contains:
+
+        * name (string) - package name
+        * current_version (string) - currently installed version
+        * new_version (string) - version available in the repository
+
+        Example::
+
+            [
+                {
+                    "name": "openssh-server",
+                    "current_version": "1:9.2p1-2+deb12u1",
+                    "new_version": "1:9.2p1-2+deb12u2",
+                },
+            ]
+        """
+        raise NotImplementedError
+
+    def apply_updates(self, packages: List[str]) -> ApplyUpdatesResultDict:
+        """
+        Upgrades the given packages to the newest available version.
+
+        Only packages that are already installed may be upgraded; this method
+        does **not** install new packages.  Pass an empty list to upgrade
+        **all** packages that have pending updates.
+
+        :param packages: List of package names to upgrade.  Each name must
+            match ``^[a-zA-Z0-9_\\-\\+\\.]+$``; a :exc:`ValueError` is raised
+            for any name that does not conform.
+        :returns: A dict with:
+
+            * success (bool) – ``True`` if the package manager exited without error
+            * output (string) – combined stdout / stderr from the package manager
+            * error (string, optional) – short error message when *success* is ``False``
+
+        :raises ValueError: If any package name fails the safety check.
+
+        Example::
+
+            result = driver.apply_updates(["openssh-server", "curl"])
+            # → {"success": True, "output": "Reading package lists...\\n..."}
+
+            # Upgrade everything:
+            result = driver.apply_updates([])
+        """
+        raise NotImplementedError
+
+    # ------------------------------------------------------------------
+    # Service management
+    # ------------------------------------------------------------------
+
+    def get_services(self) -> List[ServiceDict]:
+        """
+        Returns a list of system services and their current state.
+
+        Each entry contains:
+
+        * name (string) - service unit name (without ``.service`` suffix)
+        * running (bool) - ``True`` if the service is currently active
+        * enabled (bool) - ``True`` if the service starts automatically on boot
+        * pid (int) - main process ID; ``0`` if not running
+
+        Example::
+
+            [
+                {
+                    "name": "ssh",
+                    "running": True,
+                    "enabled": True,
+                    "pid": 1234,
+                },
+                {
+                    "name": "cron",
+                    "running": True,
+                    "enabled": True,
+                    "pid": 5678,
+                },
+            ]
+        """
+        raise NotImplementedError
+
+    # ------------------------------------------------------------------
+    # Users
+    # ------------------------------------------------------------------
+
+    def get_users(self) -> List[UserDict]:
+        """
+        Returns a list of local OS user accounts.
+
+        Each entry contains:
+
+        * username (string) - login name
+        * uid (int) - numeric user ID
+        * gid (int) - primary group ID
+        * home (string) - home directory path
+        * shell (string) - login shell path
+        * groups (list of strings) - all supplementary group names
+
+        Example::
+
+            [
+                {
+                    "username": "admin",
+                    "uid": 1000,
+                    "gid": 1000,
+                    "home": "/home/admin",
+                    "shell": "/bin/bash",
+                    "groups": ["sudo", "docker", "adm"],
+                },
+            ]
+        """
+        raise NotImplementedError
+
+    # ------------------------------------------------------------------
+    # Processes
+    # ------------------------------------------------------------------
+
+    def get_processes(self) -> List[ProcessDict]:
+        """
+        Returns a snapshot of currently running processes.
+
+        Each entry contains:
+
+        * pid (int) - process ID
+        * ppid (int) - parent process ID
+        * user (string) - effective user name
+        * cpu (float) - CPU utilisation percentage
+        * memory (float) - RSS as a percentage of total RAM
+        * vsz (int) - virtual memory size in KiB
+        * rss (int) - resident set size in KiB
+        * tty (string) - controlling terminal; empty string if none
+        * state (string) - process state: ``"R"`` running, ``"S"`` sleeping,
+          ``"D"`` uninterruptible, ``"Z"`` zombie, ``"T"`` stopped, etc.
+        * started (string) - start time as printed by ``ps`` (e.g. ``"12:34"`` or ``"May28"``)
+        * command (string) - full command line
+
+        Example::
+
+            [
+                {
+                    "pid": 1,
+                    "ppid": 0,
+                    "user": "root",
+                    "cpu": 0.0,
+                    "memory": 0.1,
+                    "vsz": 168576,
+                    "rss": 13312,
+                    "tty": "",
+                    "state": "S",
+                    "started": "May28",
+                    "command": "/sbin/init",
+                },
+            ]
+        """
+        raise NotImplementedError
+
+    # ------------------------------------------------------------------
+    # Cron jobs
+    # ------------------------------------------------------------------
+
+    def get_cron_jobs(self) -> List[CronJobDict]:
+        """
+        Returns scheduled cron tasks from all user crontabs and ``/etc/cron.d``.
+
+        Each entry contains:
+
+        * user (string) - owner of the crontab entry
+        * schedule (string) - five-field cron expression (e.g. ``"0 * * * *"``)
+        * command (string) - shell command to execute
+        * description (string, optional) - inline comment text if present
+
+        Example::
+
+            [
+                {
+                    "user": "root",
+                    "schedule": "0 4 * * *",
+                    "command": "/usr/local/bin/backup.sh",
+                    "description": "nightly backup",
+                },
+            ]
+        """
+        raise NotImplementedError
+
+    # ------------------------------------------------------------------
+    # SNMP
+    # ------------------------------------------------------------------
+
+    def get_snmp_config(self) -> Optional[SNMPConfigDict]:
+        """
+        Returns the SNMP agent configuration currently active on the device,
+        or ``None`` if no SNMP daemon is running or detectable.
+
+        The returned dictionary contains:
+
+        * running (bool) - whether the SNMP daemon is currently active
+        * community (string) - the read community string (e.g. ``"public"``)
+        * port (int) - the UDP port the agent listens on (default ``161``)
+        * version (string) - highest supported SNMP version: ``"1"``, ``"2c"``, or ``"3"``
+
+        Example::
+
+            # snmpd running with community "public":
+            {
+                "running": True,
+                "community": "public",
+                "port": 161,
+                "version": "2c",
+            }
+
+            # snmpd not installed / not running:
+            None
+        """
+        raise NotImplementedError
+
+    # ------------------------------------------------------------------
+    # Docker
+    # ------------------------------------------------------------------
+
+    def get_docker_info(self) -> DockerInfoDict:
+        """
+        Returns information about the local Docker environment.
+
+        If Docker is not installed or the current user lacks access to the
+        Docker socket, returns ``{"available": False}``.  When the user has
+        no socket permission, ``permission_denied`` is additionally set to
+        ``True``.
+
+        When Docker is available the dict contains:
+
+        * available (bool) - always ``True``
+        * version (string) - Docker Engine version string
+        * containers (list) - all containers (running and stopped), each with:
+
+          * id (string) - short container ID
+          * name (string) - container name(s)
+          * image (string) - image reference
+          * image_version (string) - OCI ``org.opencontainers.image.version`` label; empty if absent
+          * command (string) - entrypoint / command string
+          * created (string) - creation timestamp string
+          * status (string) - human-readable status (e.g. ``"Up 3 hours"``)
+          * ports (string) - port mapping string
+          * state (string) - ``"running"``, ``"exited"``, ``"paused"``, etc.
+
+        * images (list) - local images, each with:
+
+          * id (string) - short image ID
+          * repository (string) - image repository
+          * tag (string) - image tag
+          * size (string) - human-readable size string (e.g. ``"187MB"``)
+          * created (string) - creation timestamp string
+          * version (string) - OCI ``org.opencontainers.image.version`` label; empty if absent
+
+        * volumes (list) - Docker volumes, each with:
+
+          * name (string) - volume name
+          * driver (string) - volume driver
+          * mountpoint (string) - host filesystem path
+          * scope (string) - ``"local"`` or ``"global"``
+
+        * networks (list) - Docker networks, each with:
+
+          * id (string) - short network ID
+          * name (string) - network name
+          * driver (string) - network driver (e.g. ``"bridge"``, ``"host"``, ``"overlay"``)
+          * scope (string) - network scope
+          * ipv6 (string) - ``"true"`` if IPv6 is enabled
+          * internal (string) - ``"true"`` if the network is internal
+
+        * outdated_images (list of strings) - image names where the local digest
+          differs from the latest remote digest; empty list if all images are
+          current or update checks could not be performed.
+
+        Example::
+
+            # Docker not installed:
+            {"available": False}
+
+            # Docker installed, no socket permission:
+            {"available": False, "permission_denied": True}
+
+            # Docker available:
+            {
+                "available": True,
+                "version": "Docker version 27.3.1, build ce12230",
+                "containers": [
+                    {
+                        "id": "a1b2c3d4e5f6",
+                        "name": "my-app",
+                        "image": "nginx:latest",
+                        "image_version": "1.27.0",
+                        "command": "nginx -g 'daemon off;'",
+                        "created": "2026-05-28 10:00:00 +0000 UTC",
+                        "status": "Up 3 days",
+                        "ports": "0.0.0.0:80->80/tcp",
+                        "state": "running",
+                    },
+                ],
+                "images": [...],
+                "volumes": [],
+                "networks": [...],
+                "outdated_images": ["nginx:latest"],
+            }
+        """
+        raise NotImplementedError
+
+    # ------------------------------------------------------------------
+    # Generic device actions
+    # ------------------------------------------------------------------
+
+    def run_device_action(self, action: str) -> DeviceActionResultDict:
+        """
+        Executes a named administrative action on the device.
+
+        This method is an extensibility point for driver-specific one-off
+        operations that do not fit any other NAPALM API method.  Each driver
+        documents the action names it supports.
+
+        :param action: Action identifier string (e.g. ``"fix_docker_permissions"``).
+        :raises NotImplementedError: If the driver does not implement this method.
+        :raises ValueError: If ``action`` is not a recognised action name for
+            this driver.
+
+        :returns: A dict with:
+
+            * success (bool) – ``True`` if the action completed without error
+            * output (string) – human-readable output or status message
+
+        Example::
+
+            result = driver.run_device_action("fix_docker_permissions")
+            # → {"success": True, "output": "Added 'pi' to the docker group. Reconnect..."}
+        """
+        raise NotImplementedError

napalm_device_types/residential_gateway.py

@@ -0,0 +1,197 @@
+"""
+Abstract base class for residential gateway drivers.
+
+A "residential gateway" combines the roles of router, firewall and
+wireless access point in a single consumer device (e.g. AVM FritzBox,
+ISP-supplied DSL/cable routers). This base class merges the relevant
+subsets of :class:`~napalm_device_types.firewall.FirewallDriver` and
+:class:`~napalm_device_types.access_point.AccessPointDriver` plus
+gateway-specific operations (WAN status, port forwarding, connected
+hosts).
+
+Usage::
+
+    from napalm_device_types import ResidentialGatewayDriver
+
+    class FritzBoxDriver(ResidentialGatewayDriver):
+        def get_wan_status(self):
+            ...
+"""
+
+from typing import Dict, List
+from napalm_device_types.base import DeviceTypeDriver
+from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
+from napalm_device_types.models import (
+    HealthMetricsDict,
+    HostDict,
+    NATTranslationDict,
+    PortForwardDict,
+    RadioStatusDict,
+    SSIDDict,
+    VPNTunnelDict,
+    WANStatusDict,
+    WirelessClientDict,
+)
+
+
+class ResidentialGatewayDriver(DeviceTypeDriver):
+    TYPE_LABEL: str = "Gateway"
+    """
+    Abstract intermediate driver for residential gateways (router + firewall + AP).
+
+    Inherits all standard NAPALM NetworkDriver methods and adds the
+    gateway-, firewall- and wireless-specific operations that concrete
+    drivers must implement.
+    """
+
+    _SNMP_SKIP_IF = IF_SKIP_DEFAULT
+    _SNMP_TX_ERR_IS_DROP: bool = False
+
+    @classmethod
+    async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict:
+        return await collect_ucd_metrics(
+            snmp_get, snmp_walk,
+            tx_err_is_drop=cls._SNMP_TX_ERR_IS_DROP,
+            if_skip=cls._SNMP_SKIP_IF,
+        )
+
+    def get_wan_status(self) -> WANStatusDict:
+        """
+        Returns the status of the device's internet (WAN) uplink.
+
+        Contains:
+
+        * connection_type (string) - e.g. ``"DSL"``, ``"Cable"``, ``"PPPoE"``, ``"DHCP"``
+        * is_connected (bool) - whether the WAN connection is currently established
+        * external_ip (string) - the public IPv4 address assigned to the WAN interface
+        * uptime (int) - seconds since the WAN connection was last (re-)established
+        * bytes_sent (int) - total bytes transmitted on the WAN interface
+        * bytes_received (int) - total bytes received on the WAN interface
+        * max_bitrate_up (int) - upstream sync rate in kbit/s
+        * max_bitrate_down (int) - downstream sync rate in kbit/s
+        * external_ipv6 (string, optional) - the public IPv6 address, if any
+        * link_status (string, optional) - physical line state, e.g. ``"Up"`` / ``"Down"``
+
+        Example::
+
+            {
+                "connection_type": "DSL",
+                "is_connected": True,
+                "external_ip": "203.0.113.7",
+                "uptime": 345600,
+                "bytes_sent": 1234567890,
+                "bytes_received": 9876543210,
+                "max_bitrate_up": 40000,
+                "max_bitrate_down": 250000,
+                "link_status": "Up",
+            }
+        """
+        raise NotImplementedError
+
+    def get_port_forwards(self) -> List[PortForwardDict]:
+        """
+        Returns the configured port forwarding (port mapping) rules.
+
+        Each entry contains:
+
+        * name (string) - the rule's description/name
+        * protocol (string) - ``"TCP"`` or ``"UDP"``
+        * external_port (int) - the WAN-side port
+        * internal_ip (string) - the LAN host the traffic is forwarded to
+        * internal_port (int) - the LAN-side port
+        * enabled (bool) - whether the rule is currently active
+        * remote_host (string, optional) - restricts the forward to a specific
+          remote source address; empty/absent means "any"
+
+        Example::
+
+            [
+                {
+                    "name": "Webserver HTTPS",
+                    "protocol": "TCP",
+                    "external_port": 443,
+                    "internal_ip": "192.168.1.10",
+                    "internal_port": 443,
+                    "enabled": True,
+                }
+            ]
+        """
+        raise NotImplementedError
+
+    def get_hosts(self) -> List[HostDict]:
+        """
+        Returns the list of hosts known to the gateway (LAN clients).
+
+        Each entry contains:
+
+        * mac (string) - the host's MAC address
+        * ip (string) - the host's current IP address
+        * hostname (string) - the host's reported hostname (empty if unknown)
+        * interface_type (string) - how the host is connected, e.g. ``"LAN"``, ``"WLAN"``
+        * is_active (bool) - whether the host is currently online
+        * lease_time_remaining (int, optional) - remaining DHCP lease time in seconds
+
+        Example::
+
+            [
+                {
+                    "mac": "AA:BB:CC:DD:EE:FF",
+                    "ip": "192.168.1.42",
+                    "hostname": "laptop",
+                    "interface_type": "WLAN",
+                    "is_active": True,
+                    "lease_time_remaining": 3600,
+                }
+            ]
+        """
+        raise NotImplementedError
+
+    def get_nat_translations(self) -> List[NATTranslationDict]:
+        """
+        Returns a list of active NAT translation entries.
+
+        See :meth:`napalm_device_types.firewall.FirewallDriver.get_nat_translations`
+        for the entry format. Residential gateways typically derive this from
+        the active port-forwarding/NAT-PT table rather than a live connection
+        tracker; drivers that cannot provide this should return an empty list.
+        """
+        raise NotImplementedError
+
+    def get_vpn_tunnels(self) -> Dict[str, VPNTunnelDict]:
+        """
+        Returns the status of VPN tunnels (e.g. WireGuard road-warrior
+        access, IPsec site-to-site).
+
+        See :meth:`napalm_device_types.firewall.FirewallDriver.get_vpn_tunnels`
+        for the entry format. Drivers that cannot provide this should return
+        an empty dict.
+        """
+        raise NotImplementedError
+
+    def get_wireless_clients(self) -> List[WirelessClientDict]:
+        """
+        Returns the list of wireless clients currently associated with the
+        device's built-in access point(s).
+
+        See :meth:`napalm_device_types.access_point.AccessPointDriver.get_wireless_clients`
+        for the entry format.
+        """
+        raise NotImplementedError
+
+    def get_ssids(self) -> Dict[str, SSIDDict]:
+        """
+        Returns the configured wireless networks (SSIDs).
+
+        See :meth:`napalm_device_types.access_point.AccessPointDriver.get_ssids`
+        for the entry format.
+        """
+        raise NotImplementedError
+
+    def get_radio_status(self) -> Dict[str, RadioStatusDict]:
+        """
+        Returns the status of the device's wireless radios.
+
+        See :meth:`napalm_device_types.access_point.AccessPointDriver.get_radio_status`
+        for the entry format.
+        """
+        raise NotImplementedError

napalm_device_types/storage.py

@@ -11,7 +11,7 @@ Usage::
 """
 
 from typing import Any, Dict, List
-from napalm.base import NetworkDriver
+from napalm_device_types.base import DeviceTypeDriver
 from napalm_device_types.models import (
     DiskPoolDict,
     LogicalVolumeDict,
@@ -25,7 +25,8 @@ from napalm_device_types.models import (
 )
 
 
-class StorageDriver(NetworkDriver):
+class StorageDriver(DeviceTypeDriver):
+    TYPE_LABEL: str = "Storage"
     """
     Abstract intermediate driver for storage appliances and NAS/SAN devices
     (e.g. TrueNAS SCALE/CORE, Synology DSM, QNAP QTS, NetApp ONTAP,

napalm_device_types/switch.py

@@ -10,10 +10,11 @@ Usage::
             ...
 """
 
-from typing import Dict
-from napalm.base import NetworkDriver
+from typing import Dict, List
+from napalm_device_types.base import DeviceTypeDriver
 from napalm_device_types.models import (
     Dot1XPortDict,
+    HealthMetricsDict,
     InterfaceConfigDict,
     MACACLDict,
     PoESummaryDict,
@@ -23,7 +24,8 @@ from napalm_device_types.models import (
 )
 
 
-class SwitchDriver(NetworkDriver):
+class SwitchDriver(DeviceTypeDriver):
+    TYPE_LABEL: str = "Switch"
     """
     Abstract intermediate driver for Ethernet switches.
 
@@ -32,6 +34,15 @@ class SwitchDriver(NetworkDriver):
     operations that concrete drivers must implement.
     """
 
+    @classmethod
+    async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict:
+        """Collect SNMP health metrics for this switch type.
+
+        Switch vendors use proprietary OIDs — each concrete driver must
+        override this classmethod.
+        """
+        raise NotImplementedError
+
     def get_spanning_tree(self) -> Dict[str, SpanningTreeDict]:
         """
         Returns spanning tree status for each STP instance.
@@ -365,6 +376,27 @@ class SwitchDriver(NetworkDriver):
         """
         raise NotImplementedError
 
+    def set_lag_members(self, lag_name: str, members: List[str]) -> None:
+        """
+        Sets the full member-port list of a LAG/trunk interface.
+
+        Diffs *members* against the LAG's current members (as reported by
+        ``get_interfaces()``'s ``lag_members`` field) and adds/removes ports
+        on the device to match. The LAG itself must already exist on the
+        device; this method only manages its membership.
+
+        :param lag_name: Name of the LAG/trunk interface as returned by
+            ``get_interfaces()`` (e.g. ``"Trk1"``, ``"ch1"``, ``"Lag1"``).
+        :param members: Full desired list of member port names.
+        :raises NotImplementedError: If the driver does not implement this method.
+        :raises ValueError: If *lag_name* does not refer to a valid LAG.
+
+        Example::
+
+            driver.set_lag_members("Lag1", ["GigabitEthernet0/1", "GigabitEthernet0/2"])
+        """
+        raise NotImplementedError
+
     def set_poe_enabled(self, interface: str, enabled: bool) -> None:
         """
         Administratively enables or disables PoE on a single port.

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-device-types"
-version = "0.1.0"
+version = "0.5.0"
 description = "Abstract device-type base classes for NAPALM drivers"
 readme = "README.md"
 requires-python = ">=3.9"

tests/test_fingerprint_base.py

@@ -0,0 +1,150 @@
+"""Tests for DeviceTypeDriver, FingerprintRule, and PortSpec."""
+
+from __future__ import annotations
+
+import pytest
+from napalm_device_types import (
+    AccessPointDriver,
+    DeviceTypeDriver,
+    FingerprintRule,
+    FirewallDriver,
+    HypervisorDriver,
+    OSDriver,
+    PortSpec,
+    ResidentialGatewayDriver,
+    StorageDriver,
+    SwitchDriver,
+)
+from napalm.base import NetworkDriver
+
+
+# ── Hierarchy ─────────────────────────────────────────────────────────────────
+
+TYPE_DRIVERS = [
+    AccessPointDriver,
+    FirewallDriver,
+    HypervisorDriver,
+    OSDriver,
+    ResidentialGatewayDriver,
+    StorageDriver,
+    SwitchDriver,
+]
+
+
+@pytest.mark.parametrize("cls", TYPE_DRIVERS)
+def test_type_driver_inherits_device_type_driver(cls):
+    assert issubclass(cls, DeviceTypeDriver)
+
+
+@pytest.mark.parametrize("cls", TYPE_DRIVERS)
+def test_type_driver_inherits_network_driver(cls):
+    """Transitiv: DeviceTypeDriver → NetworkDriver muss erhalten bleiben."""
+    assert issubclass(cls, NetworkDriver)
+
+
+def test_device_type_driver_inherits_network_driver():
+    assert issubclass(DeviceTypeDriver, NetworkDriver)
+
+
+# ── Default attributes ────────────────────────────────────────────────────────
+
+def test_defaults_on_device_type_driver():
+    assert DeviceTypeDriver.VENDOR == ""
+    assert DeviceTypeDriver.DRIVER_NAME == ""
+    assert DeviceTypeDriver.PORT_SPECS is None
+    assert DeviceTypeDriver.SNMP_OBJECT_ID_PREFIX is None
+    assert DeviceTypeDriver.SNMP_FINGERPRINT == []
+    assert DeviceTypeDriver.SSH_FINGERPRINT == []
+    assert DeviceTypeDriver.HTTP_FINGERPRINT == []
+    assert DeviceTypeDriver.OUI_PREFIXES == []
+
+
+@pytest.mark.parametrize("cls", TYPE_DRIVERS)
+def test_fingerprint_attributes_inherited(cls):
+    assert hasattr(cls, "HTTP_FINGERPRINT")
+    assert hasattr(cls, "SNMP_FINGERPRINT")
+    assert hasattr(cls, "SSH_FINGERPRINT")
+    assert hasattr(cls, "PORT_SPECS")
+    assert hasattr(cls, "SNMP_OBJECT_ID_PREFIX")
+    assert hasattr(cls, "OUI_PREFIXES")
+
+
+# ── FingerprintRule ───────────────────────────────────────────────────────────
+
+def test_fingerprint_rule_defaults():
+    r = FingerprintRule("fritz!box")
+    assert r.pattern == "fritz!box"
+    assert r.weight == 1.0
+    assert r.mandatory is False
+    assert r.negative is False
+
+
+def test_fingerprint_rule_mandatory():
+    r = FingerprintRule("opnsense", weight=8.0, mandatory=True)
+    assert r.mandatory is True
+    assert r.weight == 8.0
+
+
+def test_fingerprint_rule_negative():
+    r = FingerprintRule("pfsense", weight=5.0, negative=True)
+    assert r.negative is True
+
+
+def test_fingerprint_rule_is_immutable():
+    r = FingerprintRule("test", weight=3.0)
+    with pytest.raises(AttributeError):
+        r.weight = 99.0  # type: ignore[misc]
+
+
+# ── PortSpec ──────────────────────────────────────────────────────────────────
+
+def test_port_spec_defaults():
+    p = PortSpec("https", 8006)
+    assert p.scheme == "https"
+    assert p.port == 8006
+    assert p.paths == ("/",)
+    assert p.weight == 5.0
+    assert p.mandatory is False
+
+
+def test_port_spec_mandatory():
+    p = PortSpec("http", 1400, ("/xml/device_description.xml",), weight=9.0, mandatory=True)
+    assert p.mandatory is True
+    assert p.paths == ("/xml/device_description.xml",)
+
+
+def test_port_spec_is_immutable():
+    p = PortSpec("http", 80)
+    with pytest.raises(AttributeError):
+        p.port = 8080  # type: ignore[misc]
+
+
+# ── Concrete driver subclass ──────────────────────────────────────────────────
+
+def test_concrete_driver_overrides_fingerprint():
+    class MyDriver(FirewallDriver):
+        VENDOR = "Acme"
+        DRIVER_NAME = "acme"
+        HTTP_FINGERPRINT = [FingerprintRule("acme portal", weight=9.0, mandatory=True)]
+        PORT_SPECS = [PortSpec("https", 9443, weight=7.0)]
+        OUI_PREFIXES = ["AA:BB:CC"]
+
+    assert MyDriver.VENDOR == "Acme"
+    assert MyDriver.HTTP_FINGERPRINT[0].pattern == "acme portal"
+    assert MyDriver.PORT_SPECS[0].port == 9443
+    assert MyDriver.OUI_PREFIXES == ["AA:BB:CC"]
+    # base class unaffected
+    assert FirewallDriver.HTTP_FINGERPRINT == []
+    assert FirewallDriver.PORT_SPECS is None
+    assert FirewallDriver.OUI_PREFIXES == []
+
+
+def test_type_label_unaffected_by_refactor():
+    """TYPE_LABEL muss nach dem Refactoring noch korrekt sein."""
+    assert FirewallDriver.TYPE_LABEL == "Firewall"
+    assert SwitchDriver.TYPE_LABEL == "Switch"
+    assert ResidentialGatewayDriver.TYPE_LABEL == "Gateway"
+    assert HypervisorDriver.TYPE_LABEL == "Hypervisor"
+    assert OSDriver.TYPE_LABEL == "OS"
+    assert StorageDriver.TYPE_LABEL == "Storage"
+    assert AccessPointDriver.TYPE_LABEL == "Access Point"

tests/test_type_label.py

@@ -0,0 +1,53 @@
+"""Tests for TYPE_LABEL class attribute on all base driver classes."""
+
+import pytest
+from napalm_device_types import (
+    AccessPointDriver,
+    FirewallDriver,
+    HypervisorDriver,
+    OSDriver,
+    ResidentialGatewayDriver,
+    StorageDriver,
+    SwitchDriver,
+)
+
+
+BASE_CLASSES = [
+    (AccessPointDriver, "Access Point"),
+    (FirewallDriver, "Firewall"),
+    (HypervisorDriver, "Hypervisor"),
+    (OSDriver, "OS"),
+    (ResidentialGatewayDriver, "Gateway"),
+    (StorageDriver, "Storage"),
+    (SwitchDriver, "Switch"),
+]
+
+
+@pytest.mark.parametrize("cls, expected_label", BASE_CLASSES)
+def test_type_label_present(cls, expected_label):
+    assert hasattr(cls, "TYPE_LABEL"), f"{cls.__name__} is missing TYPE_LABEL"
+
+
+@pytest.mark.parametrize("cls, expected_label", BASE_CLASSES)
+def test_type_label_value(cls, expected_label):
+    assert cls.TYPE_LABEL == expected_label
+
+
+@pytest.mark.parametrize("cls, _", BASE_CLASSES)
+def test_type_label_is_nonempty_string(cls, _):
+    assert isinstance(cls.TYPE_LABEL, str) and cls.TYPE_LABEL
+
+
+def test_subclass_inherits_type_label():
+    class MySwitch(SwitchDriver):
+        pass
+
+    assert MySwitch.TYPE_LABEL == "Switch"
+
+
+def test_subclass_can_override_type_label():
+    class MySpecialOS(OSDriver):
+        TYPE_LABEL = "Linux"
+
+    assert MySpecialOS.TYPE_LABEL == "Linux"
+    assert OSDriver.TYPE_LABEL == "OS"  # base class unaffected