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

@@ -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:
@@ -1279,4 +1280,4 @@ def reversed(self) -> 'Waveform':
         return self._inner
 
     def __repr__(self):
-        return f"ReversedWaveform(inner={self._inner!r})"
+        return f"ReversedWaveform(inner={self._inner!r})"

qupulse/pulses/pulse_template.py

@@ -67,7 +67,7 @@ def __init__(self, *,
         self._metadata = metadata
 
         self.__cached_hash_value = None
-
+        
     @property
     def metadata(self) -> TemplateMetadata:
         """The metadata is intended for information which does not concern the pulse itself but rather its usage.
@@ -157,7 +157,15 @@ 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.
+        """
+        return getattr(self.metadata,'pow_2_divisor',0)
+
     def create_program(self, *,
                        parameters: Optional[Mapping[str, Union[Expression, str, Number]]]=None,
                        measurement_mapping: Optional[Mapping[str, Optional[str]]]=None,
@@ -708,10 +716,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

@@ -1118,4 +1118,4 @@ def test_reversed_sample(self):
         np.testing.assert_equal(output, sample_output[::-1])
         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])])
+            ('A', list(1.5 - time_array[::-1]), mem[::-1])])

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),