feat: report unshaped devices with reasons

List each unshaped device from LibreQoS refresh output with circuit and parent context, and explain why the device was not attached to the shaping tree.

FIXES #856

src/LibreQoS.py

@@ -26,6 +26,12 @@ from virtual_tree_nodes import (
     build_physical_network,
     is_virtual_node,
 )
+from shaping_skip_report import (
+    build_unshaped_device_report,
+    collect_parent_node_names,
+    device_shaping_key,
+    format_unshaped_device_line,
+)
 
 from liblqos_python import is_lqosd_alive, clear_ip_mappings, delete_ip_mapping, validate_shaped_devices, \
     is_libre_already_running, create_lock_file, free_lock_file, add_ip_mapping, BatchedCommands, \
@@ -1554,7 +1560,7 @@ def refreshShapers():
         bakery = Bakery()
         bakery.start_batch() # Initializes the bakery transaction
         linuxTCcommands = []
-        devicesShaped = []
+        shapedDeviceKeys = set()
         # Root HTB Setup
         # Create MQ qdisc for each CPU core / rx-tx queue. Generate commands to create corresponding HTB and leaf classes. Prepare commands for execution later
         thisInterface = interface_a()
@@ -1809,8 +1815,7 @@ def refreshShapers():
                                             device.get('deviceID', ''),
                                         )
                                         #xdpCPUmapCommands.append('./bin/xdp_iphash_to_cpu_cmdline add --ip ' + str(ipv6) + ' --cpu ' + data[node]['up_cpuNum'] + ' --classid ' + circuit['up_classid'] + ' --upload 1')
-                            if device['deviceName'] not in devicesShaped:
-                                devicesShaped.append(device['deviceName'])
+                            shapedDeviceKeys.add(device_shaping_key(circuit, device))
                 # Recursive call this function for children nodes attached to this node
                 if 'children' in data[node]:
                     # Sort children to ensure consistent traversal order
@@ -1896,17 +1901,21 @@ def refreshShapers():
 
 
         # Recap - warn operator if devices were skipped
-        devicesSkipped = []
-        for circuit in subscriberCircuits:
-            for device in circuit['devices']:
-                if device['deviceName'] not in devicesShaped:
-                    devicesSkipped.append((device['deviceName'],device['deviceID']))
+        validParentNodes = collect_parent_node_names(network)
+        devicesSkipped = build_unshaped_device_report(
+            subscriberCircuits,
+            shapedDeviceKeys,
+            validParentNodes,
+            flat_network,
+        )
         if len(devicesSkipped) > 0:
-            warnings.warn('Some devices were not shaped. Please check to ensure they have a valid ParentNode listed in ShapedDevices.csv:', stacklevel=2)
+            warnings.warn(
+                str(len(devicesSkipped)) + " device(s) were not shaped. Detailed reasons are listed below.",
+                stacklevel=2,
+            )
             print("Devices not shaped:")
             for entry in devicesSkipped:
-                name, idNum = entry
-                print('DeviceID: ' + idNum + '\t DeviceName: ' + name)
+                print(format_unshaped_device_line(entry))
 
         # Save ShapedDevices.csv as ShapedDevices.lastLoaded.csv
         shutil.copyfile('ShapedDevices.csv', 'ShapedDevices.lastLoaded.csv')

src/shaping_skip_report.py

@@ -0,0 +1,113 @@
+def _string_value(value):
+    if value is None:
+        return ""
+    return str(value)
+
+
+def _circuit_parent_values(circuit):
+    logical_parent = _string_value(circuit.get("logicalParentNode", circuit.get("ParentNode")))
+    effective_parent = _string_value(circuit.get("ParentNode"))
+    return logical_parent, effective_parent
+
+
+def device_shaping_key(circuit, device):
+    return (
+        _string_value(circuit.get("circuitID", "")),
+        _string_value(device.get("deviceID", "")),
+        _string_value(device.get("deviceName", "")),
+    )
+
+
+def collect_parent_node_names(network):
+    names = set()
+
+    def walk(nodes):
+        if not isinstance(nodes, dict):
+            return
+        for name, details in nodes.items():
+            if not isinstance(name, str) or not isinstance(details, dict):
+                continue
+            names.add(name)
+            children = details.get("children")
+            if isinstance(children, dict):
+                walk(children)
+
+    walk(network)
+    return names
+
+
+def _classify_unshaped_circuit(circuit, valid_parent_nodes, flat_network):
+    logical_parent_str, effective_parent_str = _circuit_parent_values(circuit)
+
+    if flat_network:
+        return (
+            "unattached_flat_network",
+            "Circuit was not attached during shaping in flat-network mode.",
+        )
+
+    if logical_parent_str in ("", "none"):
+        return (
+            "missing_parent",
+            "No ParentNode was configured for this circuit, so it could not be attached to the shaping tree.",
+        )
+
+    if effective_parent_str not in valid_parent_nodes:
+        return (
+            "unknown_parent",
+            f"ParentNode '{logical_parent_str}' was not found in the shaping tree.",
+        )
+
+    return (
+        "unattached_circuit",
+        f"Circuit was not attached during shaping even though parent node '{effective_parent_str}' exists.",
+    )
+
+
+def format_unshaped_device_line(entry):
+    return (
+        f"DeviceID: {entry['deviceID']}\t DeviceName: {entry['deviceName']}"
+        f"\t CircuitID: {entry['circuitID']}\t CircuitName: {entry['circuitName']}"
+        f"\t LogicalParent: {entry['logicalParentNode']}"
+        f"\t EffectiveParent: {entry['effectiveParentNode']}"
+        f"\t Reason: {entry['reasonText']}"
+    )
+
+
+def build_unshaped_device_report(subscriber_circuits, shaped_device_keys, valid_parent_nodes, flat_network):
+    skipped_devices = []
+    shaped_key_set = set(shaped_device_keys)
+
+    for circuit in subscriber_circuits:
+        logical_parent, effective_parent = _circuit_parent_values(circuit)
+        reason_code, reason_text = _classify_unshaped_circuit(
+            circuit,
+            valid_parent_nodes,
+            flat_network,
+        )
+        for device in circuit.get("devices", []):
+            if device_shaping_key(circuit, device) in shaped_key_set:
+                continue
+            skipped_devices.append(
+                {
+                    "deviceID": _string_value(device.get("deviceID", "")),
+                    "deviceName": _string_value(device.get("deviceName", "")),
+                    "circuitID": _string_value(circuit.get("circuitID", "")),
+                    "circuitName": _string_value(circuit.get("circuitName", "")),
+                    "logicalParentNode": logical_parent,
+                    "effectiveParentNode": effective_parent,
+                    "reasonCode": reason_code,
+                    "reasonText": reason_text,
+                }
+            )
+
+    skipped_devices.sort(
+        key=lambda entry: (
+            entry["reasonCode"],
+            entry["logicalParentNode"],
+            entry["effectiveParentNode"],
+            entry["circuitID"],
+            entry["deviceID"],
+            entry["deviceName"],
+        )
+    )
+    return skipped_devices

src/test_shaping_skip_report.py

@@ -0,0 +1,165 @@
+import unittest
+
+from shaping_skip_report import (
+    build_unshaped_device_report,
+    collect_parent_node_names,
+    device_shaping_key,
+    format_unshaped_device_line,
+)
+
+
+class TestShapingSkipReport(unittest.TestCase):
+    def test_collect_parent_node_names_walks_nested_tree(self):
+        network = {
+            "Root": {
+                "children": {
+                    "AP_A": {
+                        "children": {
+                            "Sector_1": {}
+                        }
+                    }
+                }
+            }
+        }
+
+        self.assertEqual(
+            collect_parent_node_names(network),
+            {"Root", "AP_A", "Sector_1"},
+        )
+
+    def test_reports_unknown_parent_for_unattached_device(self):
+        circuit = {
+            "circuitID": "100",
+            "circuitName": "Subscriber 100",
+            "logicalParentNode": "Missing_AP",
+            "ParentNode": "Missing_AP",
+            "devices": [
+                {"deviceID": "dev-1", "deviceName": "Radio A"},
+            ],
+        }
+
+        skipped = build_unshaped_device_report(
+            [circuit],
+            shaped_device_keys=set(),
+            valid_parent_nodes={"Root", "AP_A"},
+            flat_network=False,
+        )
+
+        self.assertEqual(len(skipped), 1)
+        self.assertEqual(skipped[0]["reasonCode"], "unknown_parent")
+        self.assertIn("Missing_AP", skipped[0]["reasonText"])
+
+    def test_reports_missing_parent_in_non_flat_network(self):
+        circuit = {
+            "circuitID": "101",
+            "circuitName": "Subscriber 101",
+            "logicalParentNode": "none",
+            "ParentNode": "none",
+            "devices": [
+                {"deviceID": "dev-2", "deviceName": "Radio B"},
+            ],
+        }
+
+        skipped = build_unshaped_device_report(
+            [circuit],
+            shaped_device_keys=set(),
+            valid_parent_nodes={"Root", "Generated_PN_1"},
+            flat_network=False,
+        )
+
+        self.assertEqual(len(skipped), 1)
+        self.assertEqual(skipped[0]["reasonCode"], "missing_parent")
+
+    def test_shaped_device_is_not_reported(self):
+        circuit = {
+            "circuitID": "102",
+            "circuitName": "Subscriber 102",
+            "logicalParentNode": "AP_A",
+            "ParentNode": "AP_A",
+            "devices": [
+                {"deviceID": "dev-3", "deviceName": "Radio C"},
+            ],
+        }
+        shaped_keys = {device_shaping_key(circuit, circuit["devices"][0])}
+
+        skipped = build_unshaped_device_report(
+            [circuit],
+            shaped_device_keys=shaped_keys,
+            valid_parent_nodes={"AP_A"},
+            flat_network=False,
+        )
+
+        self.assertEqual(skipped, [])
+
+    def test_duplicate_device_names_do_not_collide(self):
+        shaped_circuit = {
+            "circuitID": "200",
+            "circuitName": "Subscriber 200",
+            "logicalParentNode": "AP_A",
+            "ParentNode": "AP_A",
+            "devices": [
+                {"deviceID": "dev-4", "deviceName": "Shared Name"},
+            ],
+        }
+        skipped_circuit = {
+            "circuitID": "201",
+            "circuitName": "Subscriber 201",
+            "logicalParentNode": "Ghost_AP",
+            "ParentNode": "Ghost_AP",
+            "devices": [
+                {"deviceID": "dev-5", "deviceName": "Shared Name"},
+            ],
+        }
+
+        skipped = build_unshaped_device_report(
+            [shaped_circuit, skipped_circuit],
+            shaped_device_keys={device_shaping_key(shaped_circuit, shaped_circuit["devices"][0])},
+            valid_parent_nodes={"AP_A"},
+            flat_network=False,
+        )
+
+        self.assertEqual(len(skipped), 1)
+        self.assertEqual(skipped[0]["deviceID"], "dev-5")
+        self.assertEqual(skipped[0]["reasonCode"], "unknown_parent")
+
+    def test_flat_network_uses_flat_reason_when_device_is_unattached(self):
+        circuit = {
+            "circuitID": "300",
+            "circuitName": "Subscriber 300",
+            "logicalParentNode": "none",
+            "ParentNode": "none",
+            "devices": [
+                {"deviceID": "dev-6", "deviceName": "Radio D"},
+            ],
+        }
+
+        skipped = build_unshaped_device_report(
+            [circuit],
+            shaped_device_keys=set(),
+            valid_parent_nodes={"Generated_PN_1"},
+            flat_network=True,
+        )
+
+        self.assertEqual(len(skipped), 1)
+        self.assertEqual(skipped[0]["reasonCode"], "unattached_flat_network")
+
+    def test_format_unshaped_device_line_contains_context(self):
+        line = format_unshaped_device_line(
+            {
+                "deviceID": "dev-7",
+                "deviceName": "Radio E",
+                "circuitID": "301",
+                "circuitName": "Subscriber 301",
+                "logicalParentNode": "Ghost_AP",
+                "effectiveParentNode": "Ghost_AP",
+                "reasonText": "ParentNode 'Ghost_AP' was not found in the shaping tree.",
+            }
+        )
+
+        self.assertIn("DeviceID: dev-7", line)
+        self.assertIn("CircuitName: Subscriber 301", line)
+        self.assertIn("Reason: ParentNode 'Ghost_AP' was not found in the shaping tree.", line)
+
+
+if __name__ == "__main__":
+    unittest.main()