Fix inspect factory labels via nGraphFromDrivers

src/videoipath_automation_tool/apps/inspect/api/inspect_api.py

@@ -8,9 +8,8 @@
 from __future__ import annotations
 
 import logging
-from typing import Any, Optional
+from typing import Any
 
-from . import queries
 from videoipath_automation_tool.apps.inspect.model.actions import (
     InspectApiAddDevicesItem,
     InspectApiAddDevicesRequest,
@@ -58,9 +57,11 @@
 from videoipath_automation_tool.connector.vip_connector import VideoIPathConnector
 from videoipath_automation_tool.utils.cross_app_utils import create_fallback_logger
 
+from . import queries
+
 
 class InspectAPI:
-    def __init__(self, vip_connector: VideoIPathConnector, logger: Optional[logging.Logger] = None) -> None:
+    def __init__(self, vip_connector: VideoIPathConnector, logger: logging.Logger | None = None) -> None:
         self._logger = logger or create_fallback_logger("videoipath_automation_tool_inspect_api")
         self.vip_connector = vip_connector
         self._logger.debug("Inspect API initialized.")
@@ -73,7 +74,7 @@ def get_device_skeleton(self) -> list[InspectApiNodeStatusItem]:
         items = _extract_items(response.data, "status", "collector", "inspect", "nodeStatus")
         return [InspectApiNodeStatusItem.model_validate(item) for item in items]
 
-    def get_device_detail(self, device_id: str) -> Optional[InspectApiNodeStatusItem]:
+    def get_device_detail(self, device_id: str) -> InspectApiNodeStatusItem | None:
         """One device's full nodeStatus sub-tree (lazy hydration)."""
         response = self.vip_connector.rest.get(queries.device_detail(device_id), allow_projection=True)
         items = _extract_items(response.data, "status", "collector", "inspect", "nodeStatus")
@@ -87,7 +88,7 @@ def get_edge_skeleton(self) -> list[InspectApiExternalEdgesByDeviceKeyItem]:
         items = _extract_items(response.data, "status", "collector", "externalEdgesByDeviceKey")
         return [InspectApiExternalEdgesByDeviceKeyItem.model_validate(item) for item in items]
 
-    def get_edge_pair(self, pair_id: str) -> Optional[InspectApiExternalEdgesByDeviceKeyItem]:
+    def get_edge_pair(self, pair_id: str) -> InspectApiExternalEdgesByDeviceKeyItem | None:
         """A single external-edge device pair (targeted refresh)."""
         response = self.vip_connector.rest.get(queries.edge_pair(pair_id), allow_projection=True)
         items = _extract_items(response.data, "status", "collector", "externalEdgesByDeviceKey")
@@ -126,6 +127,26 @@ def get_virtual_devices(self) -> list[InspectApiVirtualDeviceInstance]:
         items = _extract_items(response.data, "status", "network", "virtualDevices")
         return [InspectApiVirtualDeviceInstance.model_validate(item) for item in items]
 
+    def get_ngraph_factory_labels(self, *element_ids: str) -> dict[str, str]:
+        """Unchangeable ``fDescriptor.label`` values for the given device/vertex ids.
+
+        Collector ``nodeStatus`` does not populate factory labels on 2025.4.9. The primary source
+        is ``status/network/nGraphFromDrivers`` (driver-reported graph). When that is empty (e.g.
+        topology virtual devices), falls back to ``config/network/nGraphElements``.
+        """
+        if not element_ids:
+            return {}
+        result: dict[str, str] = {}
+        for device_id in _factory_label_device_roots(*element_ids):
+            response = self.vip_connector.rest.get(queries.driver_factory_labels(device_id), allow_projection=True)
+            items = _extract_items(response.data, "status", "network", "nGraphFromDrivers")
+            result.update(_flatten_driver_factory_labels(items))
+        if not result:
+            response = self.vip_connector.rest.get(queries.config_factory_labels(*element_ids), allow_projection=True)
+            items = _extract_items(response.data, "config", "network", "nGraphElements")
+            result.update(_flatten_config_factory_labels(items))
+        return result
+
     # --- Lookups (baselines for compare-and-commit) ---
 
     def lookup_inspect_device(self, device_id: str) -> InspectApiLookupInspectDeviceResponse:
@@ -205,6 +226,65 @@ def add_virtual_topology(self, data: InspectApiAddVirtualTopologyData) -> Inspec
 # --- Internal ---
 
 
+def _factory_label_device_roots(*element_ids: str) -> list[str]:
+    """Distinct device ids to query on ``nGraphFromDrivers`` (one GET per device)."""
+    roots: list[str] = []
+    for element_id in element_ids:
+        if "::" in element_id:
+            continue
+        root = _factory_label_device_root(element_id)
+        if root is not None and root not in roots:
+            roots.append(root)
+    return roots
+
+
+def _factory_label_device_root(element_id: str) -> str | None:
+    if "::" in element_id:
+        return None
+    parts = element_id.split(".")
+    if len(parts) == 1:
+        return element_id
+    if parts[0] == "virtual" and len(parts) >= 2 and parts[1].isdigit():
+        return f"{parts[0]}.{parts[1]}" if len(parts) > 2 else element_id
+    if parts[0].startswith("device") and len(parts) > 1 and parts[1].isdigit():
+        return parts[0]
+    return element_id
+
+
+def _flatten_driver_factory_labels(items: list[dict[str, Any]]) -> dict[str, str]:
+    """Map element id → ``fDescriptor.label`` from one ``nGraphFromDrivers`` item."""
+    result: dict[str, str] = {}
+    for item in items:
+        if not isinstance(item, dict):
+            continue
+        for key, value in item.items():
+            if key in ("_id", "_vid") or not isinstance(value, dict):
+                continue
+            label = _fdescriptor_label(value.get("fDescriptor"))
+            if label:
+                result[key] = label
+    return result
+
+
+def _flatten_config_factory_labels(items: list[dict[str, Any]]) -> dict[str, str]:
+    result: dict[str, str] = {}
+    for item in items:
+        if not isinstance(item, dict):
+            continue
+        element_id = item.get("_id")
+        label = _fdescriptor_label(item.get("fDescriptor"))
+        if isinstance(element_id, str) and label:
+            result[element_id] = label
+    return result
+
+
+def _fdescriptor_label(descriptor: Any) -> str | None:
+    if not isinstance(descriptor, dict):
+        return None
+    label = descriptor.get("label")
+    return label if isinstance(label, str) and label else None
+
+
 def _extract_items(data: dict[str, Any], *path: str) -> list[dict[str, Any]]:
     """Walk ``data`` down ``path`` and return the ``_items`` list (empty if any node is absent)."""
     node: Any = data

src/videoipath_automation_tool/apps/inspect/api/queries.py

@@ -80,6 +80,28 @@ def virtual_devices() -> str:
     return _build(_VIRTUAL_DEVICES)
 
 
+def driver_factory_labels(device_id: str) -> str:
+    """GET path for driver-reported ``fDescriptor.label`` values of one device graph.
+
+    Collector ``nodeStatus`` does not populate factory labels on 2025.4.9. The Inspect UI reads
+    the driver graph from ``status/network/nGraphFromDrivers`` (not collector, not config nGraph).
+    One device id returns the baseDevice plus all vertices/edges keyed by element id.
+    """
+    return _build(f"/status/network/nGraphFromDrivers/{device_id}/*/fDescriptor/**")
+
+
+def config_factory_labels(*element_ids: str) -> str:
+    """GET path for persisted ``fDescriptor.label`` values (Topology config nGraph fallback).
+
+    Used when ``nGraphFromDrivers`` has no entry (e.g. topology virtual devices). Each id is
+    matched as ``_id`` (baseDevice / vertex) or ``deviceId`` (all vertices of a device).
+    """
+    if not element_ids:
+        raise ValueError("element_ids must not be empty.")
+    clauses = " or ".join(f"_id='{element_id}' or deviceId='{element_id}'" for element_id in element_ids)
+    return _build(f"/config/network/nGraphElements/* where {clauses} /fDescriptor/**")
+
+
 # --- Internal ---
 
 # Characters that are meaningful in the projection grammar and must survive encoding.
@@ -147,14 +169,16 @@ def _build(path: str) -> str:
 
 __all__ = [
     "MAX_QUERY_LENGTH",
-    "encode",
-    "device_skeleton",
+    "alarms_section",
+    "collector_full",
+    "config_factory_labels",
     "device_detail",
-    "edge_skeleton",
+    "device_skeleton",
+    "driver_factory_labels",
     "edge_pair",
+    "edge_skeleton",
+    "encode",
     "paths_section",
-    "alarms_section",
-    "collector_full",
-    "virtual_templates",
     "virtual_devices",
+    "virtual_templates",
 ]

src/videoipath_automation_tool/apps/inspect/domain/device.py

@@ -72,9 +72,13 @@ def description(self, value: str) -> None:
 
     @property
     def factory_label(self) -> str | None:
-        """Device-reported factory label (``fDescriptor.label`` / collector ``label``)."""
+        """Unchangeable device-reported factory label (driver ``nGraphFromDrivers`` ``fDescriptor.label``).
+
+        Never the user override (``descriptor.label``). Collector ``nodeStatus`` does not populate
+        this on 2025.4.9, so the snapshot resolves it from fromDrivers (config nGraph fallback).
+        """
         node = self._record().node
-        return node.label
+        return node.factory_label or self.snapshot.get_factory_label(self.id)
 
     @property
     def pid(self) -> str | None:
@@ -159,7 +163,7 @@ def tags(self, value: list[str] | tuple[str, ...]) -> None:
     @property
     def local_assigned_tags(self) -> list[str]:
         """Device ``localAssignedTags`` (distinct from collector ``tags`` when both are present)."""
-        return self._staged_or("localAssignedTags", lambda: [], adapt=list)
+        return self._staged_or("localAssignedTags", list, adapt=list)
 
     @local_assigned_tags.setter
     def local_assigned_tags(self, value: list[str]) -> None:
@@ -331,7 +335,7 @@ def __repr__(self) -> str:
 
     __str__ = __repr__
 
-    def _record(self) -> "_DeviceRecord":
+    def _record(self) -> _DeviceRecord:
         record = self.snapshot.get_device_record(self.id)
         if record is None:
             raise KeyError(f"Device '{self.id}' is no longer present in the snapshot.")

src/videoipath_automation_tool/apps/inspect/domain/port.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any, Mapping, Self
+from collections.abc import Mapping
+from typing import TYPE_CHECKING, Any, Self
 
 from pydantic import Field, model_validator
 
@@ -116,8 +117,20 @@ def label(self) -> str | None:
 
     @property
     def factory_label(self) -> str | None:
-        """The device-reported (factory) port label, even when a manual override is set."""
-        return self.indexed.port.label
+        """The unchangeable device-reported factory port label, even when a manual override is set.
+
+        Collector ``nodeStatus`` exposes this as the top-level ``label`` when present; on 2025.4.9
+        that field is null, so the snapshot resolves ``fDescriptor.label`` from
+        ``nGraphFromDrivers`` (config nGraph fallback) via a vertex on this port.
+        """
+        collector = self.indexed.port.factory_label
+        if collector:
+            return collector
+        for vertex_id, _ in self._vertex_sides():
+            label = self.snapshot.get_factory_label(vertex_id, device_id=self.indexed.device_id)
+            if label:
+                return label
+        return None
 
     @property
     def description(self) -> str | None:

src/videoipath_automation_tool/apps/inspect/domain/vertex.py

@@ -25,13 +25,13 @@
 
 from videoipath_automation_tool.apps.inspect.model.collector import InspectApiSingleVertexInfo
 from videoipath_automation_tool.apps.inspect.model.common import (
+    _STAGED_MISSING,
     InspectCodecFormat,
     InspectControl,
     InspectEditableModel,
     InspectSipsMode,
     InspectVertexKind,
     InspectVertexType,
-    _STAGED_MISSING,
     format_repr,
 )
 from videoipath_automation_tool.apps.inspect.snapshot import InspectSnapshot
@@ -77,8 +77,12 @@ def vertex_type(self) -> InspectVertexType | str | None:
 
     @property
     def factory_label(self) -> str | None:
-        """Device-reported factory label of the owning port (set when built via a port)."""
-        return self.port_factory_label
+        """Unchangeable factory label of this vertex (driver ``nGraphFromDrivers`` ``fDescriptor.label``).
+
+        When built via a port, the owning port's collector factory label is used if fromDrivers is
+        unavailable (offline tests). Never the editable form ``label``.
+        """
+        return self.snapshot.get_factory_label(self.id) or self.port_factory_label
 
     @property
     def is_active(self) -> bool | None:
@@ -648,10 +652,10 @@ def build_vertex(
 
 
 __all__ = [
-    "InspectVertex",
+    "InspectCodecVertex",
     "InspectGenericVertex",
     "InspectIpVertex",
-    "InspectCodecVertex",
     "InspectResourceTransformVertex",
+    "InspectVertex",
     "build_vertex",
 ]

src/videoipath_automation_tool/apps/inspect/model/collector.py

@@ -11,11 +11,11 @@
     InspectApiEndpointStatus,
     InspectApiRestV2Header,
     InspectApiStatusContext,
+    InspectApiStatusSummary,
     InspectIconSize,
     InspectIconType,
     InspectSdpStrategy,
     InspectServiceStatus,
-    InspectApiStatusSummary,
     InspectSeverity,
     InspectVertexType,
     map_severity,
@@ -138,6 +138,12 @@ def assigned_tags(self) -> list[str]:
             return list(assigned["all"])
         return []
 
+    @property
+    def factory_label(self) -> str | None:
+        """Device-reported factory port label (top-level ``label``). Distinct from the user
+        override in ``descriptor.label``. Null on 2025.4.9 collector payloads."""
+        return self.label or None
+
     @property
     def effective_label(self) -> str | None:
         if self.descriptor is not None and self.descriptor.label:
@@ -243,6 +249,15 @@ class InspectApiNodeStatusItem(InspectApiBaseModel):
     def _map_sync_severity(cls, value: Any) -> Any:
         return map_severity(value)
 
+    @property
+    def factory_label(self) -> str | None:
+        """Unchangeable device-reported label: ``fDescriptor.label``, then the legacy top-level
+        ``label``. Never ``descriptor.label`` (the user override). Collector ``nodeStatus`` leaves
+        both empty on 2025.4.9 — domain objects then resolve from nGraph ``fDescriptor``."""
+        if self.fDescriptor is not None and self.fDescriptor.label:
+            return self.fDescriptor.label
+        return self.label or None
+
     @property
     def effective_label(self) -> str | None:
         """The label the UI shows: user ``descriptor.label``, falling back to the device-reported
@@ -394,9 +409,9 @@ class InspectApiCollectorResponse(InspectApiBaseModel):
     "InspectApiPathSegment",
     "InspectApiPathServiceFields",
     "InspectApiPathStructure",
-    "InspectPortStatus",
     "InspectApiSingleVertexInfo",
     "InspectApiSuperProfileItem",
     "InspectApiTagInfoItem",
     "InspectApiVertexInfoFields",
+    "InspectPortStatus",
 ]