from __future__ import annotations
import abc
from collections import defaultdict
from collections.abc import Iterable
from enum import Enum, auto
from qcodes.parameters import DelegateParameter, Parameter
from qcodes.instrument import InstrumentBase
import itertools
from warnings import warn
from typing import Any
[docs]
class BetterDelegateParameter(DelegateParameter):
# def get_ramp_values(self, value: float | Sized, step: float | None = None) -> Sequence[float | Sized]:
# if step is None:
# return [value]
# if isinstance(value, Sized) and len(value) > 1:
# raise RuntimeError("...")
# if self.get_latest() is None:
# if self.get() is not None:
# start_value = self.get()
# else:
# start_value = 0
# else:
# start_value = self.get_latest()
# if TYPE_CHECKING:
# assert isinstance(value, float)
# return permissive_range(start_value, value, step)[1:] + [value]
[docs]
def validate(self, value: Any | None) -> None:
if value is None:
value = 0
super().validate(value)
def __repr__(self) -> str:
def repr_(obj: Parameter | None) -> str:
return (
(f"{obj.__module__}.{type(obj).__name__}: {obj.full_name} at {id(obj)}")
if obj is not None else "None"
)
return f"<{repr_(self)} with source <{repr_(self.source)}>>"
[docs]
class NoneParameter(Parameter):
def __init__(self, **kwargs: Any):
super().__init__(name="NoneParameter", **kwargs)
[docs]
class Node(abc.ABC):
def __init__(
self,
name: str = "Node",
parameters: Iterable[Parameter] = (),
) -> None:
self._name = name
self._parameters = tuple(parameters)
@property
def name(self) -> str:
return self._name
@property
def parameters(self) -> Iterable[Parameter]:
return self._parameters
# def add_source(self, source: Node) -> None:
# _LOG.debug(f"Adding source {source.name} to Node {self.name}")
# def remove_source(self, source: Node) -> None:
# _LOG.debug(f"Removing source {source.name} to Node {self.name}")
# def activate(self) -> None:
# _LOG.debug(f"Activating Node {self.name}")
# def deactivate(self) -> None:
# _LOG.debug(f"Deactivating Node {self.name}")
[docs]
class Edge(Enum):
Active = auto()
Inactive = auto()
Disabled = auto()
[docs]
class SingleSourceNode(Node):
def __init__(
self,
name: str = "SingleSourceNode",
) -> None:
super().__init__(name=name)
self._source: Node | None = None
@property
def source(self) -> Node | None:
return self._source
# def add_source(self, source: Node) -> None:
# super().add_source(source=source)
# if self._source is None or self._source is source:
# self._source = source
# else:
# raise self.SourceError("...")
# def remove_source(self, source: Node) -> None:
# super().remove_source(source=source)
# if self._source == source:
# self._source = None
# else:
# raise self.SourceError("...")
[docs]
class ForwardingNode(SingleSourceNode):
@property
def parameters(self) -> Iterable[Parameter]:
if self._source is None:
return []
return self._source.parameters
[docs]
class MultiSourceForwardingNode(Node):
def __init__(
self,
name: str = "MultiSourceForwardingNode",
) -> None:
super().__init__(name=name)
self._sources: set[Node] = set()
@property
def parameters(self) -> Iterable[Parameter]:
return itertools.chain.from_iterable(source.parameters for source in self._sources)
# def add_source(self, source: Node) -> None:
# super().add_source(source=source)
# self._sources.add(source)
# def remove_source(self, source: Node) -> None:
# super().remove_source(source=source)
# self._sources.remove(source)
[docs]
class CompositeParameter(Parameter):
def __init__(self, name: str, **kwargs: Any):
super().__init__(name, **kwargs)
self.__params: set[Parameter] = set()
[docs]
def get_raw(self) -> Any:
raise NotImplementedError
@property
def source(self) -> set[Parameter]:
return self.__params
@source.setter
def source(self, parameter: Parameter) -> None:
if isinstance(parameter, NoneParameter):
self.__params.clear()
self.__params.add(parameter)
[docs]
class SeriesResistanceParameter(CompositeParameter):
[docs]
def get_raw(self) -> float:
return sum(
param() for param in self.source if param() is not None
)
[docs]
class DelegatingNode(SingleSourceNode):
def __init__(self, *parameters: DelegateParameter, name: str = "DelegatingNode"):
super().__init__(name=name)
self._delegates: dict[str, DelegateParameter] = {parameter.unit: parameter for parameter in parameters}
self._candidate_delegates: dict[str, list[Parameter]] = defaultdict(list)
assert len(parameters) == len(self._delegates)
self._active_units: set[str] = set()
@property
def parameters(self) -> Iterable[Parameter]:
if self._source is None:
return []
return self._source.parameters
# def add_source(self, source: Node) -> None:
# super().add_source(source)
# for parameter in source.parameters:
# self._candidate_delegates[parameter.unit].append(parameter)
# def remove_source(self, source: Node):
# super().remove_source(source)
# units_to_remove = {parameter.unit for parameter in source.parameters}
# if self._active_units.intersection(units_to_remove):
# raise self.SourceError(f"Attempt to remove an active source {source}")
def _activate_from_candidates(self, delegate_unit: str) -> None:
candidates = list(self._candidate_delegates[delegate_unit])
if isinstance(self._delegates[delegate_unit], CompositeParameter):
for candidate in candidates:
self._activate_parameter(candidate)
return
num_candidates = len(set(candidates))
if num_candidates == 0:
return
last_candidate = candidates[-1]
if num_candidates > 1:
excluded_candidates = set(candidates).difference({last_candidate})
warn(f"Multiple parameters found for unit {delegate_unit} on node {self.name} \n"
f"using {last_candidate} and excluding {excluded_candidates}")
self._activate_parameter(last_candidate)
def _activate_parameter(self, parameter: Parameter) -> None:
unit = parameter.unit
self._delegates[unit].source = parameter
if parameter.instrument is not None:
_apply_limits_to(parameter.instrument, self._delegates[unit].instrument)
# self._activate_units.add(unit)
def _apply_limits_to(
source_instrument: InstrumentBase,
delegate_instrument: InstrumentBase | None,
) -> None:
if delegate_instrument is None:
return
# if (getattr(delegate_instrument, "limit_rate", None) is not None
# and getattr(source_instrument, "limit_rate", None) is not None):
# source_instrument.limit_rate = delegate_instrument.limit_rate
if delegate_instrument.set_hw_limits() and (delegate_instrument.vals is not None):
valid_values = delegate_instrument.vals.valid_values
if len(valid_values) == 2:
sw_min, sw_max = valid_values
source_instrument.limit_min(sw_min)
source_instrument.limit_max(sw_max)
else:
raise ValueError("...")