Linspace followup waveforms

qupulse/hardware/awgs/base.py

@@ -319,7 +319,7 @@ def _sample_waveforms(self, waveforms: Sequence[Waveform]) -> List[Tuple[Tuple[n
             segment_length = int(segment_length)
             segment_end = segment_begin + segment_length
 
-            wf_time = time_array[:segment_length]
+            wf_time = time_array[:segment_length] * 2**waveform._pow_2_divisor
             wf_sample_memory = sample_memory[:segment_length]
 
             sampled_channels = []

qupulse/hardware/util.py

@@ -132,7 +132,7 @@ def get_waveform_length(waveform: Waveform,
     Returns:
         Number of samples for the waveform
     """
-    segment_length = waveform.duration * sample_rate_in_GHz
+    segment_length = waveform.duration * sample_rate_in_GHz / 2**waveform._pow_2_divisor
 
     # __round__ is implemented for Fraction and gmpy2.mpq
     rounded_segment_length = round(segment_length)

qupulse/program/waveforms.py

@@ -37,7 +37,7 @@ class ConstantFunctionPulseTemplateWarning(UserWarning):
 
 __all__ = ["Waveform", "TableWaveform", "TableWaveformEntry", "FunctionWaveform", "SequenceWaveform",
            "MultiChannelWaveform", "RepetitionWaveform", "TransformingWaveform", "ArithmeticWaveform",
-           "ConstantFunctionPulseTemplateWarning", "ConstantWaveform"]
+           "ConstantFunctionPulseTemplateWarning", "ConstantWaveform", "WaveformCollection"]
 
 PULSE_TO_WAVEFORM_ERROR = None  # error margin in pulse template to waveform conversion
 
@@ -59,10 +59,11 @@ class Waveform(metaclass=ABCMeta):
 
     __sampled_cache = WeakValueDictionary()
 
-    __slots__ = ('_duration',)
+    __slots__ = ('_duration','_pow_2_divisor')
 
-    def __init__(self, duration: TimeType):
+    def __init__(self, duration: TimeType, _pow_2_divisor: int = 0):
         self._duration = duration
+        self._pow_2_divisor = _pow_2_divisor
 
     @property
     def duration(self) -> TimeType:
@@ -1280,3 +1281,71 @@ def reversed(self) -> 'Waveform':
 
     def __repr__(self):
         return f"ReversedWaveform(inner={self._inner!r})"
+
+
+class WaveformCollection():
+    """ This class is intended to be a nested collection of equal-length waveforms
+    that may be used in a command-based architecture where a program structure
+    only defines one loop structure per waveform playback but
+    can cycle through multiple waveforms, i.e. in pseudocode
+
+    int i=0;
+    while i<10:
+        play(waveforms[i]);
+        i+=1;
+
+    or with higher nesting levels.
+
+    This was mainly conceived for the HDAWG and its CommandTable.
+
+    """
+    def __init__(self, waveform_collection: Tuple[Union[Waveform,"WaveformCollection"]]):
+        assert isinstance(waveform_collection,Sequence)
+        assert all(isinstance(wf,Waveform) for wf in waveform_collection) or\
+               all(isinstance(wf,type(self)) for wf in waveform_collection)
+        self._waveform_collection = tuple(waveform_collection)
+        try:
+            d=self.duration
+        except AssertionError as e:
+            raise e
+
+    @property
+    def duration(self) -> TimeType:
+        lens = [wf.duration for wf in self.waveform_collection]
+        assert np.all(np.isclose([float(l) for l in lens], float(lens[0]))), 'non-equal durations'
+        return lens[0]
+
+    @property
+    def waveform_collection(self) -> Tuple[Union[Waveform,"WaveformCollection"]]:
+        return self._waveform_collection
+
+    @property
+    def nesting_level(self) -> int:
+        #assume it is balanced for now.
+        if isinstance(self.waveform_collection[0],type(self)):
+            return self.waveform_collection[0].nesting_level+1
+        return 0
+
+    def flatten(self) -> Tuple[Waveform]:
+        #depth first
+        def flatten_tuple(nested_tuple):
+            for item in nested_tuple:
+                if isinstance(item, type(self)):
+                    yield from flatten_tuple(item.waveform_collection)
+                else:
+                    yield item
+        return tuple(flatten_tuple(self.waveform_collection))
+
+    def reversed(self) -> 'WaveformCollection':
+        """Returns a reversed version of this WaveformCollection.
+        order and waveform/sub-collection order are reversed.
+        """
+        rev = tuple(w.reversed() for w in self._waveform_collection[::-1])
+        return WaveformCollection(rev)
+
+    @property
+    def _pow_2_divisor(self) -> int:
+        #!!! the implementation of this feature has to be discussed, but it needs to be included somehow.
+        divs = set(wf._pow_2_divisor for wf in self.flatten())
+        assert len(divs)==1
+        return divs.pop()

qupulse/pulses/pulse_template.py

@@ -67,7 +67,8 @@ def __init__(self, *,
         self._metadata = metadata
 
         self.__cached_hash_value = None
-
+        self.__pow_2_divisor = 0
+
     @property
     def metadata(self) -> TemplateMetadata:
         """The metadata is intended for information which does not concern the pulse itself but rather its usage.
@@ -157,7 +158,18 @@ def initial_values(self) -> Dict[ChannelID, ExpressionScalar]:
     def final_values(self) -> Dict[ChannelID, ExpressionScalar]:
         """Values of defined channels at t == self.duration"""
         raise NotImplementedError(f"The pulse template of type {type(self)} does not implement `final_values`")
-
+
+    @property
+    def _pow_2_divisor(self) -> int:
+        """A hacky implementation of telling waveforms to be sampled at reduced rate.
+        The hardware implementation will be responsible for correctly handling this,
+        so do not use unless support is ascertained.
+        """
+
+    @_pow_2_divisor.setter
+    def _pow_2_divisor(self, val: int):
+        self.__pow_2_divisor = val
+
     def create_program(self, *,
                        parameters: Optional[Mapping[str, Union[Expression, str, Number]]]=None,
                        measurement_mapping: Optional[Mapping[str, Optional[str]]]=None,
@@ -708,10 +720,12 @@ def _internal_create_program(self, *,
             measurements = self.get_measurement_windows(parameters=scope,
                                                         measurement_mapping=measurement_mapping)
             program_builder.measure(measurements)
-
+            
             if global_transformation:
                 waveform = TransformingWaveform.from_transformation(waveform, global_transformation)
-
+
+            waveform._pow_2_divisor = self._pow_2_divisor
+
             constant_values = waveform.constant_value_dict()
             if constant_values is None:
                 program_builder.play_arbitrary_waveform(waveform)

tests/_program/waveforms_tests.py

@@ -9,7 +9,7 @@
     JumpInterpolationStrategy
 from qupulse.program.waveforms import MultiChannelWaveform, RepetitionWaveform, SequenceWaveform,\
     TableWaveformEntry, TableWaveform, TransformingWaveform, SubsetWaveform, ArithmeticWaveform, ConstantWaveform,\
-    Waveform, FunctorWaveform, FunctionWaveform, ReversedWaveform
+    Waveform, FunctorWaveform, FunctionWaveform, ReversedWaveform, WaveformCollection
 from qupulse.program.transformation import LinearTransformation
 from qupulse.expressions import ExpressionScalar, Expression
 
@@ -1119,3 +1119,56 @@ def test_reversed_sample(self):
         np.testing.assert_equal(dummy_wf.sample_calls, [
             ('A', list(1.5 - time_array[::-1]), None),
             ('A', list(1.5 - time_array[::-1]), mem[::-1])])
+
+
+class WaveformCollectionTests(unittest.TestCase):
+    def setUp(self):
+        self._dummy_waveforms = \
+            tuple(DummyWaveform(duration=2.,sample_output=np.array([i,2]),defined_channels={'A','B'}) for i in range(3))
+        self._dummy_waveforms2 = \
+            tuple(DummyWaveform(duration=2.,sample_output=np.array([i,3]),defined_channels={'C','D'}) for i in range(3))
+        self._flat_coll = WaveformCollection(self._dummy_waveforms)
+        self._flat_coll2 = WaveformCollection(self._dummy_waveforms2)
+        self._nested_coll_1 = WaveformCollection((self._flat_coll,self._flat_coll2))
+        self._nested_coll_2 = WaveformCollection(tuple(self._nested_coll_1 for i in range(4)))
+
+    def test_unequal(self):
+        self.assertRaises(AssertionError,lambda: WaveformCollection((self._flat_coll,self._dummy_waveforms[0])))
+
+    def test_duration(self):
+        self.assertAlmostEqual(self._flat_coll.duration, 2., places=12)
+        self.assertAlmostEqual(self._nested_coll_2.duration, 2., places=12)
+
+    def test_nesting(self):
+        self.assertEqual(self._nested_coll_1.waveform_collection,(self._flat_coll,self._flat_coll2))
+        self.assertEqual(self._nested_coll_2.nesting_level,2)
+        self.assertEqual(self._flat_coll.nesting_level,0)
+
+    def test_flatten(self):
+        self.assertEqual(self._nested_coll_1.flatten(),(*self._dummy_waveforms,*self._dummy_waveforms2))
+        self.assertEqual(len(self._nested_coll_2.flatten()),24)
+
+    def test_reversed(self):
+        rev_manual = WaveformCollection(
+            (WaveformCollection(tuple(wf.reversed() for wf in self._dummy_waveforms2[::-1])),
+             WaveformCollection(tuple(wf.reversed() for wf in self._dummy_waveforms[::-1]))))
+        rev = self._nested_coll_1.reversed()
+        self.assertEqual(rev.flatten(),rev_manual.flatten())
+        self.assertEqual(type(rev.flatten()[0]), ReversedWaveform)
+
+    def test_pow_2_divisor(self):
+        self.assertEqual(self._nested_coll_2._pow_2_divisor, 0)
+
+        wf_div, wf_div_inc = [], []
+        for i in range(3):
+            wf = DummyWaveform(duration=2.,sample_output=np.array([i,2]),defined_channels={'A','B'})
+            wf_div.append(wf)
+            wf._pow_2_divisor = 5
+            wf2 = DummyWaveform(duration=2.,sample_output=np.array([i,2]),defined_channels={'A','B'})
+            wf2._pow_2_divisor = i
+            wf_div_inc.append(wf2)
+
+        wfcoll, wfcoll2 = WaveformCollection(wf_div), WaveformCollection(wf_div_inc)
+
+        self.assertEqual(wfcoll._pow_2_divisor, 5)
+        self.assertRaises(AssertionError,lambda:wfcoll2._pow_2_divisor)

tests/hardware/base_tests.py

@@ -6,6 +6,8 @@
 
 from qupulse.utils.types import TimeType
 from qupulse.program.loop import Loop
+from qupulse.program.waveforms import FunctionWaveform
+from qupulse.expressions import Expression, ExpressionScalar
 from qupulse.hardware.awgs.base import ProgramEntry
 
 from tests.pulses.sequencing_dummies import DummyWaveform
@@ -102,3 +104,54 @@ def test_sample_waveforms(self):
             with mock.patch.object(entry, '_sample_empty_marker', return_value=empty_m):
                 sampled = entry._sample_waveforms(self.waveforms)
                 np.testing.assert_equal(expected_sampled, sampled)
+
+
+class ProgramEntryDivisorTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.channels = ('A',)
+        self.marker = tuple()
+        self.amplitudes = (1.,)
+        self.offset = (0.,)
+        self.voltage_transformations = (
+            mock.Mock(wraps=lambda x: x),
+        )
+        self.sample_rate = TimeType.from_float(2.4)
+
+        t = np.arange(0,400/12,1/2.4)
+
+        self.sampled = [
+            dict(A=np.sin(t)),
+            dict(A=np.sin(t[::8])),
+        ]
+
+        wf = FunctionWaveform(ExpressionScalar('sin(t)'), 400/12, 'A')
+        wf2 = FunctionWaveform(ExpressionScalar('sin(t)'), 400/12, 'A')
+        wf2._pow_2_divisor = 3
+        self.waveforms = [
+            wf,wf2
+        ]
+        self.loop = Loop(children=[Loop(waveform=wf) for wf in self.waveforms])
+
+    def test_sample_waveforms_with_divisor(self):
+        empty_ch = np.array([1,])
+        empty_m = np.array([])
+        # channels ==  (A,)
+
+        expected_sampled = [
+            ((self.sampled[0]['A'],), tuple()),
+            ((self.sampled[1]['A'],), tuple()),
+        ]
+
+        entry = ProgramEntry(program=self.loop,
+                             channels=self.channels,
+                             markers=self.marker,
+                             amplitudes=self.amplitudes,
+                             offsets=self.offset,
+                             voltage_transformations=self.voltage_transformations,
+                             sample_rate=self.sample_rate,
+                             waveforms=[])
+
+        with mock.patch.object(entry, '_sample_empty_channel', return_value=empty_ch):
+            with mock.patch.object(entry, '_sample_empty_marker', return_value=empty_m):
+                sampled = entry._sample_waveforms(self.waveforms)
+                np.testing.assert_equal(expected_sampled, sampled)

tests/hardware/util_tests.py

@@ -78,7 +78,25 @@ def test_get_sample_times_single_wf(self):
         np.testing.assert_equal(times, expected_times)
         np.testing.assert_equal(n_samples, np.asarray(4))
 
-
+    def test_pow_2_divisor(self):
+        sample_rate = TimeType.from_fraction(12, 5)
+        wf = DummyWaveform(duration=TimeType.from_fraction(400, 12))
+
+        wf._pow_2_divisor = 3
+        times, n_samples = get_sample_times(wf, sample_rate_in_GHz=sample_rate)
+
+        # the expected times are still at original sample rate, just with less
+        # max values, as the logic of having one time-array
+        # for all waveforms (which assumes a fixed sample rate)
+        # would not allow intercepting those here.
+        expected_times = np.arange(10) / float(sample_rate)
+        np.testing.assert_almost_equal(times, expected_times, decimal=10)
+
+        #the segment length however comes back reduced, 10 instead of 80
+        expected_len = np.asarray(10)
+        np.testing.assert_equal(n_samples, expected_len)
+
+
 class NotNoneIndexTest(unittest.TestCase):
     def test_not_none_indices(self):
         self.assertEqual(([None, 0, 1, None, None, 2], 3),