Driver abstraction (rebased)

.gitignore

@@ -9,3 +9,4 @@ dist/*
 doc/source/examples/.ipynb_checkpoints/*
 **.asv
 *.orig
+.idea/

qupulse/_program/seqc.py

@@ -29,7 +29,7 @@
 from qupulse.utils import replace_multiple
 from qupulse._program.waveforms import Waveform
 from qupulse._program._loop import Loop
-from qupulse.hardware.awgs.base import ProgramEntry
+from qupulse.hardware.awgs.old_base import ProgramEntry
 
 try:
     import zhinst.utils

qupulse/hardware/awgs/base.py

@@ -1,218 +1,176 @@
-"""This module defines the common interface for arbitrary waveform generators.
+from abc import ABC, abstractmethod
+from typing import Optional
 
-Classes:
-    - AWG: Common AWG interface.
-    - DummyAWG: A software stub implementation of the AWG interface.
-    - ProgramOverwriteException
-    - OutOfWaveformMemoryException
-"""
+from qupulse.hardware.awgs.base_features import Feature, FeatureAble
+from qupulse.utils.types import Collection
 
-from abc import abstractmethod
-from typing import Set, Tuple, Callable, Optional
 
-from qupulse.hardware.util import get_sample_times
-from qupulse.utils.types import ChannelID
-from qupulse._program._loop import Loop
-from qupulse._program.waveforms import Waveform
-from qupulse.comparable import Comparable
-from qupulse._program.instructions import InstructionSequence
-from qupulse.utils.types import TimeType
+__all__ = ["AWGDevice", "AWGChannelTuple", "AWGChannel", "AWGMarkerChannel", "AWGDeviceFeature", "AWGChannelFeature",
+           "AWGChannelTupleFeature"]
 
-__all__ = ["AWG", "Program", "ProgramOverwriteException",
-           "OutOfWaveformMemoryException", "AWGAmplitudeOffsetHandling"]
 
-Program = InstructionSequence
+class AWGDeviceFeature(Feature, ABC):
+    """Base class for features that are used for `AWGDevice`s"""
+    def __init__(self):
+        super().__init__(AWGDevice)
 
 
-class AWGAmplitudeOffsetHandling:
-    IGNORE_OFFSET = 'ignore_offset'   # Offset is ignored.
-    CONSIDER_OFFSET = 'consider_offset' # Offset is discounted from the waveforms.
-    # TODO OPTIMIZED = 'optimized' # Offset and amplitude are set depending on the waveforms to maximize the waveforms resolution
+class AWGChannelFeature(Feature, ABC):
+    """Base class for features that are used for `AWGChannel`s"""
+    def __init__(self):
+        super().__init__(AWGChannel)
 
-    _valid = [IGNORE_OFFSET, CONSIDER_OFFSET]
 
+class AWGChannelTupleFeature(Feature, ABC):
+    """Base class for features that are used for `AWGChannelTuple`s"""
+    def __init__(self):
+        super().__init__(AWGChannelTuple)
 
-class AWG(Comparable):
-    """An arbitrary waveform generator abstraction class.
 
-    It represents a set of channels that have to have(hardware enforced) the same:
-        -control flow
-        -sample rate
+class AWGDevice(FeatureAble[AWGDeviceFeature], ABC):
+    """Base class for all drivers of all arbitrary waveform generators"""
 
-    It keeps track of the AWG state and manages waveforms and programs on the hardware.
-    """
-
-    def __init__(self, identifier: str):
-        self._identifier = identifier
-        self._amplitude_offset_handling = AWGAmplitudeOffsetHandling.IGNORE_OFFSET
+    def __init__(self, name: str):
+        """
+        Args:
+            name: The name of the device as a String
+        """
+        super().__init__()
+        self._name = name
 
-    @property
-    def identifier(self) -> str:
-        return self._identifier
+    def __del__(self):
+        self.cleanup()
 
     @property
-    def amplitude_offset_handling(self) -> str:
-        return self._amplitude_offset_handling
-
-    @amplitude_offset_handling.setter
-    def amplitude_offset_handling(self, value):
-        """
-        value (str): See possible values at `AWGAmplitudeOffsetHandling`
-        """
-        if value not in AWGAmplitudeOffsetHandling._valid:
-            raise ValueError('"{}" is invalid as AWGAmplitudeOffsetHandling'.format(value))
+    def name(self) -> str:
+        """Returns the name of a Device as a String"""
+        return self._name
 
-        self._amplitude_offset_handling = value
+    @abstractmethod
+    def cleanup(self) -> None:
+        """Function for cleaning up the dependencies of the device"""
+        raise NotImplementedError()
 
     @property
     @abstractmethod
-    def num_channels(self):
-        """Number of channels"""
+    def channels(self) -> Collection["AWGChannel"]:
+        """Returns a list of all channels of a Device"""
+        raise NotImplementedError()
 
     @property
     @abstractmethod
-    def num_markers(self):
-        """Number of marker channels"""
+    def marker_channels(self) -> Collection["AWGMarkerChannel"]:
+        """Returns a list of all marker channels of a device. The collection may be empty"""
+        raise NotImplementedError()
 
+    @property
     @abstractmethod
-    def upload(self, name: str,
-               program: Loop,
-               channels: Tuple[Optional[ChannelID], ...],
-               markers: Tuple[Optional[ChannelID], ...],
-               voltage_transformation: Tuple[Optional[Callable], ...],
-               force: bool=False) -> None:
-        """Upload a program to the AWG.
-
-        Physically uploads all waveforms required by the program - excluding those already present -
-        to the device and sets up playback sequences accordingly.
-        This method should be cheap for program already on the device and can therefore be used
-        for syncing. Programs that are uploaded should be fast(~1 sec) to arm.
+    def channel_tuples(self) -> Collection["AWGChannelTuple"]:
+        """Returns a list of all channel tuples of a list"""
+        raise NotImplementedError()
+
 
+class AWGChannelTuple(FeatureAble[AWGChannelTupleFeature], ABC):
+    """Base class for all groups of synchronized channels of an AWG"""
+
+    def __init__(self, idn: int):
+        """
         Args:
-            name: A name for the program on the AWG.
-            program: The program (a sequence of instructions) to upload.
-            channels: Tuple of length num_channels that ChannelIDs of  in the program to use. Position in the list corresponds to the AWG channel
-            markers: List of channels in the program to use. Position in the List in the list corresponds to the AWG channel
-            voltage_transformation: transformations applied to the waveforms extracted rom the program. Position
-            in the list corresponds to the AWG channel
-            force: If a different sequence is already present with the same name, it is
-                overwritten if force is set to True. (default = False)
+             idn: The identification number of a channel tuple
         """
+        super().__init__()
 
-    @abstractmethod
-    def remove(self, name: str) -> None:
-        """Remove a program from the AWG.
+        self._idn = idn
 
-        Also discards all waveforms referenced only by the program identified by name.
+    @property
+    def idn(self) -> int:
+        """Returns the identification number of a channel tuple"""
+        return self._idn
 
-        Args:
-            name: The name of the program to remove.
-        """
+    @property
+    def name(self) -> str:
+        """Returns the name of a channel tuple"""
+        return "{dev}_CT{idn}".format(dev=self.device.name, idn=self.idn)
 
+    @property
     @abstractmethod
-    def clear(self) -> None:
-        """Removes all programs and waveforms from the AWG.
+    def sample_rate(self) -> float:
+        """Returns the sample rate of a channel tuple as a float"""
+        raise NotImplementedError()
 
-        Caution: This affects all programs and waveforms on the AWG, not only those uploaded using qupulse!
-        """
+    # Optional sample_rate-setter
+    # @sample_rate.setter
+    # def sample_rate(self, sample_rate: float) -> None:
 
+    @property
     @abstractmethod
-    def arm(self, name: Optional[str]) -> None:
-        """Load the program 'name' and arm the device for running it. If name is None the awg will "dearm" its current
-        program."""
+    def device(self) -> AWGDevice:
+        """Returns the device which the channel tuple belong to"""
+        raise NotImplementedError()
 
     @property
     @abstractmethod
-    def programs(self) -> Set[str]:
-        """The set of program names that can currently be executed on the hardware AWG."""
+    def channels(self) -> Collection["AWGChannel"]:
+        """Returns a list of all channels of the channel tuple"""
+        raise NotImplementedError()
 
     @property
     @abstractmethod
-    def sample_rate(self) -> float:
-        """The sample rate of the AWG."""
+    def marker_channels(self) -> Collection["AWGMarkerChannel"]:
+        """Returns a list of all marker channels of the channel tuple. The collection may be empty"""
+        raise NotImplementedError()
+
+
+class _BaseAWGChannel(ABC):
+    """Base class for a single channel of an AWG"""
+
+    def __init__(self, idn: int):
+        """
+        Args:
+            idn: The identification number of a channel
+        """
+        super().__init__()
+        self._idn = idn
 
     @property
-    def compare_key(self) -> int:
-        """Comparison and hashing is based on the id of the AWG so different devices with the same properties
-        are ot equal"""
-        return id(self)
+    def idn(self) -> int:
+        """Returns the identification number of a channel"""
+        return self._idn
 
-    def __copy__(self) -> None:
+    @property
+    @abstractmethod
+    def device(self) -> AWGDevice:
+        """Returns the device which the channel belongs to"""
         raise NotImplementedError()
 
-    def __deepcopy__(self, memodict={}) -> None:
+    @property
+    @abstractmethod
+    def channel_tuple(self) -> Optional[AWGChannelTuple]:
+        """Returns the channel tuple which a channel belongs to"""
         raise NotImplementedError()
 
+    @abstractmethod
+    def _set_channel_tuple(self, channel_tuple) -> None:
+        """
+        Sets the channel tuple which a channel belongs to
 
-class ProgramOverwriteException(Exception):
+        Args:
+            channel_tuple: reference to the channel tuple
+        """
+        raise NotImplementedError()
 
-    def __init__(self, name) -> None:
-        super().__init__()
-        self.name = name
-
-    def __str__(self) -> str:
-        return "A program with the given name '{}' is already present on the device." \
-               " Use force to overwrite.".format(self.name)
-
-
-class ProgramEntry:
-    """This is a helper class for implementing awgs drivers. A driver can subclass it to help organizing sampled
-    waveforms"""
-    def __init__(self, loop: Loop,
-                 channels: Tuple[Optional[ChannelID], ...],
-                 markers: Tuple[Optional[ChannelID], ...],
-                 amplitudes: Tuple[float, ...],
-                 offsets: Tuple[float, ...],
-                 voltage_transformations: Tuple[Optional[Callable], ...],
-                 sample_rate: TimeType):
-        assert len(channels) == len(amplitudes) == len(offsets) == len(voltage_transformations)
-
-        self._channels = tuple(channels)
-        self._markers = tuple(markers)
-        self._amplitudes = tuple(amplitudes)
-        self._offsets = tuple(offsets)
-        self._voltage_transformations = tuple(voltage_transformations)
-
-        self._sample_rate = sample_rate
-
-        self._loop = loop
-        self._waveforms = {node.waveform: None for node in loop.get_depth_first_iterator() if node.is_leaf()}
-
-        time_array, segment_lengths = get_sample_times(self._waveforms.keys(), sample_rate)
-
-        for waveform, segment_length in zip(self._waveforms.keys(), segment_lengths):
-            wf_time = time_array[:segment_length]
-
-            sampled_channels = []
-            for channel, trafo, amplitude, offset in zip(channels, voltage_transformations, amplitudes, offsets):
-                if channel is None:
-                    sampled_channels.append(None)
-                else:
-                    sampled = waveform.get_sampled(channel, wf_time)
-                    if trafo:
-                        sampled = trafo(sampled)
-                    sampled = sampled - offset
-                    sampled /= amplitude
-                    sampled_channels.append(waveform.get_sampled(channel, wf_time))
-
-            sampled_markers = []
-            for marker in markers:
-                if marker is None:
-                    sampled_markers.append(None)
-                else:
-                    sampled_markers.append(waveform.get_sampled(marker, wf_time) != 0)
-            self._waveforms[waveform] = (tuple(sampled_channels), tuple(sampled_markers))
-
-
-class OutOfWaveformMemoryException(Exception):
-
-    def __str__(self) -> str:
-        return "Out of memory error adding waveform to waveform memory."
-
-
-class ChannelNotFoundException(Exception):
-    def __init__(self, channel):
-        self.channel = channel
-
-    def __str__(self) -> str:
-        return 'Marker or channel not found: {}'.format(self.channel)
+
+class AWGChannel(_BaseAWGChannel, FeatureAble[AWGChannelFeature], ABC):
+    """Base class for a single channel of an AWG"""
+    @property
+    def name(self) -> str:
+        """Returns the name of a channel"""
+        return "{dev}_C{idn}".format(dev=self.device.name, idn=self.idn)
+
+
+class AWGMarkerChannel(_BaseAWGChannel, ABC):
+    """Base class for a single marker channel of an AWG"""
+    @property
+    def name(self) -> str:
+        """Returns the name of a marker channel"""
+        return "{dev}_M{idn}".format(dev=self.device.name, idn=self.idn)