From f11bbb749f76c647efa3ee5244221fac01fcde1f Mon Sep 17 00:00:00 2001 From: Nomos11 <82180697+Nomos11@users.noreply.github.com> Date: Tue, 29 Jul 2025 14:09:19 +0200 Subject: [PATCH 1/5] waveformcollection --- qupulse/program/waveforms.py | 59 ++++++++++++++++++++++++++++++++++++ 1 file changed, 59 insertions(+) diff --git a/qupulse/program/waveforms.py b/qupulse/program/waveforms.py index 2a41ab911..ac3dee50d 100644 --- a/qupulse/program/waveforms.py +++ b/qupulse/program/waveforms.py @@ -1280,3 +1280,62 @@ 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"]]): + self._waveform_collection = tuple(waveform_collection) + + @property + def duration(self) -> TimeType: + lens = [wf.duration for wf in self.flatten()] + assert np.all(np.isclose([float(l) for l in lens], float(lens[0]))) + return lens[0] + + @property + def waveform_collection(self): + return self._waveform_collection + + @property + def nesting_level(self): + #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]: + 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.""" + 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() \ No newline at end of file From eb61fba0e1d9874b0c9b49fd161c5d4f6b2a4648 Mon Sep 17 00:00:00 2001 From: Nomos11 <82180697+Nomos11@users.noreply.github.com> Date: Tue, 29 Jul 2025 15:15:48 +0200 Subject: [PATCH 2/5] tests, typehints --- qupulse/program/waveforms.py | 23 +++++++++++++------ tests/_program/waveforms_tests.py | 38 ++++++++++++++++++++++++++++++- 2 files changed, 53 insertions(+), 8 deletions(-) diff --git a/qupulse/program/waveforms.py b/qupulse/program/waveforms.py index ac3dee50d..5b7cbd0a0 100644 --- a/qupulse/program/waveforms.py +++ b/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 @@ -1282,7 +1282,6 @@ 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 @@ -1300,26 +1299,34 @@ class WaveformCollection(): """ 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.flatten()] - assert np.all(np.isclose([float(l) for l in lens], float(lens[0]))) + 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): + def waveform_collection(self) -> Tuple[Union[Waveform,"WaveformCollection"]]: return self._waveform_collection @property - def nesting_level(self): + 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)): @@ -1329,7 +1336,9 @@ def flatten_tuple(nested_tuple): return tuple(flatten_tuple(self.waveform_collection)) def reversed(self) -> 'WaveformCollection': - """Returns a reversed version of this 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) diff --git a/tests/_program/waveforms_tests.py b/tests/_program/waveforms_tests.py index ece22dce1..233d9fadd 100644 --- a/tests/_program/waveforms_tests.py +++ b/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,39 @@ 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) \ No newline at end of file From 4e8bf183581f7570801c23ba76be5a268a7ddf6e Mon Sep 17 00:00:00 2001 From: Nomos11 <82180697+Nomos11@users.noreply.github.com> Date: Tue, 29 Jul 2025 17:04:38 +0200 Subject: [PATCH 3/5] pow_2_divisor and tests --- qupulse/hardware/awgs/base.py | 2 +- qupulse/hardware/util.py | 2 +- qupulse/program/waveforms.py | 5 +-- qupulse/pulses/pulse_template.py | 22 ++++++++++--- tests/_program/waveforms_tests.py | 19 ++++++++++- tests/hardware/base_tests.py | 53 +++++++++++++++++++++++++++++++ tests/hardware/util_tests.py | 20 +++++++++++- 7 files changed, 113 insertions(+), 10 deletions(-) diff --git a/qupulse/hardware/awgs/base.py b/qupulse/hardware/awgs/base.py index fe84d1934..cadbd83b5 100644 --- a/qupulse/hardware/awgs/base.py +++ b/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 = [] diff --git a/qupulse/hardware/util.py b/qupulse/hardware/util.py index ec1a91eae..cd9a85334 100644 --- a/qupulse/hardware/util.py +++ b/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) diff --git a/qupulse/program/waveforms.py b/qupulse/program/waveforms.py index 5b7cbd0a0..a8d1fde9f 100644 --- a/qupulse/program/waveforms.py +++ b/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: diff --git a/qupulse/pulses/pulse_template.py b/qupulse/pulses/pulse_template.py index 719ba220f..7684e8659 100644 --- a/qupulse/pulses/pulse_template.py +++ b/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) diff --git a/tests/_program/waveforms_tests.py b/tests/_program/waveforms_tests.py index 233d9fadd..968dff894 100644 --- a/tests/_program/waveforms_tests.py +++ b/tests/_program/waveforms_tests.py @@ -1154,4 +1154,21 @@ def test_reversed(self): 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) \ No newline at end of file + 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) \ No newline at end of file diff --git a/tests/hardware/base_tests.py b/tests/hardware/base_tests.py index 137fd071c..ee40b87d0 100644 --- a/tests/hardware/base_tests.py +++ b/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) diff --git a/tests/hardware/util_tests.py b/tests/hardware/util_tests.py index 9b983c28f..48926653a 100644 --- a/tests/hardware/util_tests.py +++ b/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), From 1e639e9a81497e1fc551fc124906073d6e13f71f Mon Sep 17 00:00:00 2001 From: Nomos11 <82180697+Nomos11@users.noreply.github.com> Date: Fri, 1 Aug 2025 11:10:00 +0200 Subject: [PATCH 4/5] pow_2_divisor only in PT-metadata --- qupulse/pulses/pulse_template.py | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/qupulse/pulses/pulse_template.py b/qupulse/pulses/pulse_template.py index 7684e8659..83b0d34c0 100644 --- a/qupulse/pulses/pulse_template.py +++ b/qupulse/pulses/pulse_template.py @@ -67,7 +67,6 @@ def __init__(self, *, self._metadata = metadata self.__cached_hash_value = None - self.__pow_2_divisor = 0 @property def metadata(self) -> TemplateMetadata: @@ -165,10 +164,7 @@ def _pow_2_divisor(self) -> int: 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 + return getattr(self.metadata,'pow_2_divisor',0) def create_program(self, *, parameters: Optional[Mapping[str, Union[Expression, str, Number]]]=None, From 60363e7901e94c0618efeb12d16a1fbf8ff23726 Mon Sep 17 00:00:00 2001 From: Nomos11 <82180697+Nomos11@users.noreply.github.com> Date: Thu, 14 Aug 2025 17:09:02 +0200 Subject: [PATCH 5/5] remove WaveformCollection and tests --- qupulse/program/waveforms.py | 72 +------------------------------ tests/_program/waveforms_tests.py | 57 +----------------------- 2 files changed, 4 insertions(+), 125 deletions(-) diff --git a/qupulse/program/waveforms.py b/qupulse/program/waveforms.py index a8d1fde9f..6805af872 100644 --- a/qupulse/program/waveforms.py +++ b/qupulse/program/waveforms.py @@ -37,7 +37,7 @@ class ConstantFunctionPulseTemplateWarning(UserWarning): __all__ = ["Waveform", "TableWaveform", "TableWaveformEntry", "FunctionWaveform", "SequenceWaveform", "MultiChannelWaveform", "RepetitionWaveform", "TransformingWaveform", "ArithmeticWaveform", - "ConstantFunctionPulseTemplateWarning", "ConstantWaveform", "WaveformCollection"] + "ConstantFunctionPulseTemplateWarning", "ConstantWaveform"] PULSE_TO_WAVEFORM_ERROR = None # error margin in pulse template to waveform conversion @@ -1280,72 +1280,4 @@ def reversed(self) -> 'Waveform': return self._inner 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() \ No newline at end of file + return f"ReversedWaveform(inner={self._inner!r})" \ No newline at end of file diff --git a/tests/_program/waveforms_tests.py b/tests/_program/waveforms_tests.py index 968dff894..5da345935 100644 --- a/tests/_program/waveforms_tests.py +++ b/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, WaveformCollection + Waveform, FunctorWaveform, FunctionWaveform, ReversedWaveform from qupulse.program.transformation import LinearTransformation from qupulse.expressions import ExpressionScalar, Expression @@ -1118,57 +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])]) - - -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) \ No newline at end of file + ('A', list(1.5 - time_array[::-1]), mem[::-1])]) \ No newline at end of file