Source code for semi_cr.core.lab.station.graphing.resolver

from collections.abc import Callable
from dataclasses import dataclass
from typing import Any




@dataclass(frozen=True)
class GraphNodeResolver:
    """
    Resolve dependency-graph nodes to canonical station-graph node IDs.

    Priority:
      1. Explicit QCoDeS object
      2. Static instrument terminal reference
      3. Path-based parameter reference
      4. Logical name
      5. Existing string node ID
      6. Fallback logical ID
    """

    qcodes_id: Callable[[Any], str]

    parameter_prefix: str = "qcodes://param/"
    logical_node_prefix: str = "lab://node/"

    def __call__(
        self,
        graph: Any,
        node_id: Any,
    ) -> str:

        attrs, value = self._node_data(
            graph,
            node_id,
        )

        # -----------------------------------------------------
        # 1. Explicit QCoDeS object
        # -----------------------------------------------------

        qc_obj = getattr(value, "qcodes_obj", None)

        if qc_obj is not None:
            return self.qcodes_id(qc_obj)

        # -----------------------------------------------------
        # 2. Existing string IDs
        # -----------------------------------------------------

        if self._is_canonical_id(node_id):
            return node_id


        # -----------------------------------------------------
        # 3. Path
        # -----------------------------------------------------

        path = getattr(value, "path", None)

        if isinstance(path, str) and path:
            return f"{self.parameter_prefix}{path}"

        # -----------------------------------------------------
        # 5. Logical name
        # -----------------------------------------------------

        name = getattr(value, "name", None)

        if isinstance(name, str) and name:
            return (
                f"{self.logical_node_prefix}{name}"
            )

        # -----------------------------------------------------
        # 6. Fallback
        # -----------------------------------------------------

        return (
            f"{self.logical_node_prefix}{node_id}"
        )

    @staticmethod
    def _node_data(
        graph: Any,
        node_id: Any,
    ) -> tuple[dict[str, Any], Any | None]:
        if hasattr(graph, "nodes"):
            attrs = graph.nodes[node_id]

            value = attrs.get("value")

            if value is None:
                value = attrs.get("obj")

            return attrs, value

        return {}, graph[node_id]

    @staticmethod
    def _is_canonical_id(node_id: Any) -> bool:
        return (
            isinstance(node_id, str)
            and node_id.startswith(
                (
                    "lab://",
                    "qcodes://",
                )
            )
        )