Source code for semi_cr.core.station.instruments.dmm

from __future__ import annotations

from dataclasses import dataclass
from typing import TYPE_CHECKING, Any

from semi_cr.core.lab.station.graphing.ids import terminal_id
from semi_cr.core.station.graph import GraphEdge, GraphNode
from semi_cr.core.station.instruments.qcodes_proxy import QCoDeSInstrumentProxy, QCoDeSInstrumentSpec

if TYPE_CHECKING:
    from collections.abc import Iterable, Mapping




@dataclass
class DMMVoltageTerminal:
    """A station-level, read-only voltage input exposed by a QCoDeS DMM."""

    parent_ref: DMMBlueprint
    name: str
    role: str | None = None

    @property
    def dmm(self) -> DMMBlueprint:
        return self.parent_ref

    @property
    def node_id(self) -> str:
        return terminal_id(self.dmm.name, self.name)



    def graph_nodes(self) -> Iterable[GraphNode]:
        yield GraphNode(
            id=self.node_id,
            obj=self,
            kind="terminal",
            attrs={
                "instrument": self.dmm.name,
                "terminal": self.name,
                "role": self.role,
                "quantity": "voltage",
                "modality": "dc",
            },
        )




    def graph_edges(self) -> Iterable[GraphEdge]:
        return ()




    def read_voltage(self) -> float:
        """Read the selected terminal through the connected live QCoDeS DMM."""
        instrument = self.dmm._qcodes_instrument
        if instrument is None:
            raise RuntimeError(f"{self.dmm.name} is not connected; call connect() first")
        terminal = getattr(instrument, self.name, None)
        if terminal is None:
            return self.dmm.read_voltage()
        for parameter_name in ("voltage", "volt"):
            parameter = getattr(terminal, parameter_name, None)
            if callable(parameter):
                return float(parameter())
        raise RuntimeError(
            f"{self.dmm.name} terminal {self.name!r} exposes neither a voltage nor volt parameter"
        )






@dataclass
class DMMBlueprint(QCoDeSInstrumentProxy):
    """Static station blueprint for a routable voltage-measuring DMM."""

    name: str
    inputs: list[DMMVoltageTerminal]
    connections: dict[str, list[str]]

    @property
    def address(self) -> str:
        return str(self.qcodes_config[1].get("init", {}).get("address", ""))



    def read_voltage(self) -> float:
        """Read the selected physical DMM through its live QCoDeS instrument."""
        if self._qcodes_instrument is None:
            raise RuntimeError(f"{self.name} is not connected; call connect() first")
        return float(self._qcodes_instrument.volt())




    @classmethod
    def from_instrument_spec(cls, spec: QCoDeSInstrumentSpec) -> DMMBlueprint:
        name, config = spec
        init: Mapping[str, Any] = config.get("init", {})
        blueprint = cls(name=name, qcodes_config=spec, inputs=[], connections={})
        connections = init.get("connections", [])
        blueprint.connections = {
            connection["name"]: list(connection.get("endpoints", []))
            for connection in connections
        }
        blueprint.inputs = [
            DMMVoltageTerminal(
                parent_ref=blueprint,
                name=connection["name"],
                role=connection.get("role", "meter"),
            )
            for connection in connections
        ]
        return blueprint




    def graph_nodes(self) -> Iterable[GraphNode]:
        for input_terminal in self.inputs:
            yield from input_terminal.graph_nodes()




    def graph_edges(self) -> Iterable[GraphEdge]:
        for input_terminal in self.inputs:
            yield from input_terminal.graph_edges()
            local = input_terminal.node_id
            for endpoint in self.connections.get(input_terminal.name, []):
                yield GraphEdge(
                    id=f"{endpoint}->{local}",
                    source=endpoint,
                    target=local,
                    kind="physical",
                )