from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from typing import Any, TYPE_CHECKING
import networkx as nx
if TYPE_CHECKING:
from semi_cr.core.lab.pins.pins import RFInput
from semi_cr.core.lab.pins.base import BasePinChannel
from qualang_tools.units import unit
from quam.components.pulses import SquareReadoutPulse
from semi_cr.core.lab.utils.quam import update_rf_device_amplitude, update_rf_device_frequency
from semi_cr.core.lab.station.routing_v1.resolution import (
find_backend_object_for_pin,
find_backend_parameter_for_pin,
)
u = unit(coerce_to_integer=True)
# class RoutedElectricalQuantity:
# quantity: Literal["voltage", "current"]
# @property
# def source(self): ...
# @property
# def meter(self): ...
# @property
# def readback_parameter(self): ...
# class RoutedVoltage(RoutedElectricalQuantity):
# quantity = "voltage"
# class RoutedCurrent(RoutedElectricalQuantity):
# quantity = "current"
[docs]
class RoutedVoltage:
def __init__(
self,
graph: nx.MultiDiGraph,
pin: BasePinChannel,
modality: str, # "dc", "ac", "rf", "pulse", ...
):
self.graph = graph
self.pin = pin
self.modality = modality
self._source = None
self._readback = None
self._meter = None
@property
def source(self):
if self._source is None:
self._source = find_backend_parameter_for_pin(
graph=self.graph,
device=self.pin.parent,
pin_name=self.pin.short_name,
role="source",
quantity="voltage",
modality=self.modality,
parameter_name="voltage",
)
return self._source
@property
def readback_parameter(self):
if self._readback is None:
self._readback = find_backend_parameter_for_pin(
graph=self.graph,
device=self.pin.parent,
pin_name=self.pin.short_name,
role="source",
quantity="voltage",
modality=self.modality,
parameter_name="voltage_readback",
)
return self._readback
@property
def meter(self):
if self._meter is None:
self._meter = find_backend_parameter_for_pin(
graph=self.graph,
device=self.pin.parent,
pin_name=self.pin.short_name,
role="meter",
quantity="voltage",
modality=self.modality,
parameter_name="voltage",
)
return self._meter
[docs]
def set(self, value: float) -> None:
self.source.set(value)
[docs]
def get(self):
return self.source.get()
[docs]
def readback(self) -> float:
return float(self.readback_parameter.get())
[docs]
def measure(self) -> float:
return float(self.meter.get())
[docs]
def invalidate(self) -> None:
"""
Discard cached backend parameters.
The next operation resolves the route again. This is needed after
station reconnection, graph reconstruction, or configurable-routing
changes.
"""
self._source = None
self._readback = None
self._meter = None
[docs]
class RoutedCurrent:
def __init__(
self,
graph: nx.MultiDiGraph,
pin: BasePinChannel,
modality: str, # "dc", "ac", "rf", "pulse", ...
):
self.graph = graph
self.pin = pin
self.modality = modality
self._source = None
self._readback = None
self._meter = None
@property
def source(self):
if self._source is None:
self._source = find_backend_parameter_for_pin(
graph=self.graph,
device=self.pin.parent,
pin_name=self.pin.short_name,
role="source",
quantity="current",
modality=self.modality,
parameter_name="current",
)
return self._source
@property
def readback_parameter(self):
if self._readback is None:
self._readback = find_backend_parameter_for_pin(
graph=self.graph,
device=self.pin.parent,
pin_name=self.pin.short_name,
role="source",
quantity="current",
modality=self.modality,
parameter_name="current_readback",
)
return self._readback
@property
def meter(self):
if self._meter is None:
self._meter = find_backend_parameter_for_pin(
graph=self.graph,
device=self.pin.parent,
pin_name=self.pin.short_name,
role="meter",
quantity="current",
modality=self.modality,
parameter_name="current",
)
return self._meter
[docs]
def set(self, value: float) -> None:
self.source.set(value)
[docs]
def get(self):
return self.source.get()
[docs]
def readback(self) -> float:
return float(self.readback_parameter.get())
[docs]
def measure(self) -> float:
return float(self.meter.get())
[docs]
def invalidate(self) -> None:
"""
Discard cached backend parameters.
The next operation resolves the route again. This is needed after
station reconnection, graph reconstruction, or configurable-routing
changes.
"""
self._source = None
self._readback = None
self._meter = None
[docs]
@dataclass
class CurrentFromTransimpedanceVoltage:
voltage: Any
transimpedance_gain: Callable[[], float]
[docs]
def measure(self) -> float:
voltage = self.voltage.measure()
gain = float(self.transimpedance_gain())
return voltage / gain
[docs]
class RoutedRFDrive:
def __init__(
self,
graph: nx.MultiDiGraph,
pin: RFInput,
modality: str = "rf",
):
self.graph = graph
self.pin = pin
self.modality = modality
self._source = None
self.operations = {
"readout": SquareReadoutPulse(
length=1000,
amplitude=0.1,
)
}
@property
def source(self):
if self._source is None:
self._source = find_backend_object_for_pin(
graph=self.graph,
device=self.pin.parent,
pin_name=self.pin.short_name,
role="source",
modality=self.modality,
)
return self._source
@property
def port(self):
return self.source.port
@property
def intermediate_frequency(self):
return self.source.intermediate_frequency
@property
def frequency(self):
return self.source.frequency
@property
def amplitude(self):
return self.source.amplitude
@property
def readout_pulse(self):
return self.operations["readout"]
[docs]
def update_frequency(self, frequency):
self.pin.quam.channel.update_frequency(frequency)
[docs]
class RoutedIQReceiver:
def __init__(
self,
graph: nx.MultiDiGraph,
pin: RFInput,
modality: str = "rf",
):
self.graph = graph
self.pin = pin
self.modality = modality
self._receiver = None
@property
def receiver(self):
if self._receiver is None:
self._receiver = find_backend_object_for_pin(
graph=self.graph,
device=self.pin.parent,
pin_name=self.pin.short_name,
role="receiver",
modality=self.modality,
)
return self._receiver
@property
def port(self):
return self.receiver.port
@property
def time_of_flight(self):
return self.receiver.time_of_flight
[docs]
class RoutedRFFrequency:
def __init__(
self,
graph,
pin,
modality="rf",
):
self.graph = graph
self.pin = pin
self.modality = modality
[docs]
def set(self, frequency):
update_rf_device_frequency(frequency, self.pin)
[docs]
class RoutedPulseAmplitude:
def __init__(
self,
graph: nx.MultiDiGraph,
pin: BasePinChannel,
modality="rf",
):
self.graph = graph
self.pin = pin
self.modality = modality
self._amplitude_scale: Any = 1.0
[docs]
def set(self, amplitude):
update_rf_device_amplitude(amplitude, self.pin)
# Only update local state if the delegated operation succeeded.
self._amplitude_scale = amplitude
[docs]
def get(self):
"""
Return the current QUA amplitude scale.
This is protocol-construction state, not a physical hardware
readback.
"""
return self._amplitude_scale
[docs]
def reset(self) -> None:
self._amplitude_scale = 1.0