Bug Fix and tests for TaborAWGs Update Vol Parameter

qupulse/_program/tabor.py

@@ -644,11 +644,11 @@ def prepare_program_for_advanced_sequence_mode(program: Loop, min_seq_len: int,
 
 
 ParsedProgram = NamedTuple('ParsedProgram', [('advanced_sequencer_table', Sequence[TableEntry]),
-                                                 ('sequencer_tables', Sequence[Sequence[
+                                             ('sequencer_tables', Sequence[Sequence[
                                                      Tuple[TableDescription, Optional[VolatileProperty]]]]),
-                                                 ('waveforms', Tuple[Waveform, ...]),
-                                                 ('volatile_parameter_positions', Dict[Union[int, Tuple[int, int]],
-                                                                                       VolatileRepetitionCount])])
+                                             ('waveforms', Tuple[Waveform, ...]),
+                                             ('volatile_parameter_positions', Dict[Union[int, Tuple[int, int]],
+                                                                                   VolatileRepetitionCount])])
 
 
 def parse_aseq_program(program: Loop, used_channels: FrozenSet[ChannelID]) -> ParsedProgram:
@@ -683,7 +683,7 @@ def parse_aseq_program(program: Loop, used_channels: FrozenSet[ChannelID]) -> Pa
         advanced_sequencer_table.append(TableEntry(repetition_count=sequencer_table_loop.repetition_count,
                                                    element_number=sequence_no, jump_flag=0))
         if sequencer_table_loop.volatile_repetition:
-            volatile_parameter_positions[adv_position] = sequencer_table_loop.volatile_repetition
+            volatile_parameter_positions[adv_position] = sequencer_table_loop.repetition_definition
 
     # transform sequencer_tables in lists to make it indexable and mutable
     sequencer_tables = list(map(list, sequencer_tables))

qupulse/hardware/awgs/tabor.py

@@ -2,6 +2,7 @@
 import functools
 import weakref
 import logging
+import numbers
 from typing import List, Tuple, Set, Callable, Optional, Any, Sequence, cast, Union, Dict, Mapping, NamedTuple
 from collections import OrderedDict
 
@@ -15,7 +16,6 @@
 from qupulse._program._loop import Loop, make_compatible
 from qupulse.hardware.util import voltage_to_uint16, find_positions
 from qupulse.hardware.awgs.base import AWG, AWGAmplitudeOffsetHandling
-from qupulse.pulses.parameters import Parameter
 from qupulse._program.tabor import TaborSegment, TaborException, TaborProgram, PlottableProgram, TaborSequencing,\
     make_combined_wave
 
@@ -723,7 +723,7 @@ def _execute_multiple_commands_with_config_guard(self, commands: List[str]) -> N
         cmd_str = ";".join(commands)
         self.device.send_cmd(cmd_str, paranoia_level=self.internal_paranoia_level)
 
-    def set_volatile_parameters(self, program_name: str, parameters: Mapping[str, Parameter]) -> None:
+    def set_volatile_parameters(self, program_name: str, parameters: Mapping[str, numbers.Number]) -> None:
         """ Set the values of parameters which were marked as volatile on program creation. Sets volatile parameters
         in program memory and device's (adv.) sequence tables if program is current program.
 

tests/_program/tabor_tests.py

@@ -1,6 +1,7 @@
 import unittest
 import itertools
 import numpy as np
+from qupulse.utils.types import FrozenDict
 from unittest import mock
 
 try:
@@ -13,8 +14,11 @@
 from qupulse._program.tabor import TaborException, TaborProgram, \
     TaborSegment, TaborSequencing, PlottableProgram, TableDescription, make_combined_wave, TableEntry
 from qupulse._program._loop import Loop
+from qupulse._program.volatile import VolatileRepetitionCount
 from qupulse.hardware.util import voltage_to_uint16
 from qupulse.utils.types import TimeType
+from qupulse.expressions import ExpressionScalar
+from qupulse.parameter_scope import DictScope
 
 from tests.pulses.sequencing_dummies import DummyWaveform
 from tests._program.loop_tests import LoopTests, WaveformGenerator
@@ -331,7 +335,7 @@ def my_gen(gen):
         with self.assertRaisesRegex(ValueError, "non integer length"):
             root_loop = LoopTests.get_test_loop(WaveformGenerator(
                 waveform_data_generator=my_gen(self.waveform_data_generator),
-                duration_generator=itertools.repeat(1/200),
+                duration_generator=itertools.repeat(1 / 200),
                 num_channels=4))
 
             TaborProgram(root_loop, self.instr_props, ('A', 'B'), (None, None), **self.program_entry_kwargs)
@@ -364,6 +368,87 @@ def my_gen(gen):
             np.testing.assert_equal(sampled_seg.ch_a, data[0])
             np.testing.assert_equal(sampled_seg.ch_b, data[1])
 
+    def test_update_volatile_parameters_with_depth1(self):
+        parameters = {'s': 10, 'not': 13}
+        s = VolatileRepetitionCount(expression=ExpressionScalar('s'), scope=DictScope(values=FrozenDict(s=3),
+                                                                                      volatile=set('s')))
+
+        wf_1 = DummyWaveform(defined_channels={'A'}, duration=1)
+        wf_2 = DummyWaveform(defined_channels={'A'}, duration=1)
+
+        program = Loop(children=[Loop(waveform=wf_1, repetition_count=s),
+                                 Loop(waveform=wf_2, repetition_count=4),
+                                 Loop(waveform=wf_1, repetition_count=1)],
+                       repetition_count=1)
+
+        t_program = TaborProgram(program, channels=(None, 'A'), markers=(None, None),
+                                 device_properties=self.instr_props, **self.program_entry_kwargs)
+
+        self.assertEqual(t_program.get_sequencer_tables(), [[(TableDescription(3, 0, 0), s.volatile_property),
+                                                             (TableDescription(4, 1, 0), None),
+                                                             (TableDescription(1, 0, 0), None)]])
+        self.assertEqual(t_program.get_advanced_sequencer_table(), [TableDescription(1, 1, 0)])
+
+        modifications = t_program.update_volatile_parameters(parameters)
+
+        expected_seq = VolatileRepetitionCount(expression=ExpressionScalar('s'), scope=DictScope(values=FrozenDict(s=10), volatile=set('s')))
+        expected_modifications = {(0, 0): TableDescription(10, 0, 0)}
+
+        self.assertEqual(t_program.get_sequencer_tables(), [[(TableDescription(10, 0, 0), expected_seq.volatile_property),
+                                                             (TableDescription(4, 1, 0), None),
+                                                             (TableDescription(1, 0, 0), None)]])
+        self.assertEqual(t_program.get_advanced_sequencer_table(), [TableDescription(1, 1, 0)])
+        self.assertEqual(modifications, expected_modifications)
+
+    def test_update_volatile_parameters_with_depth2(self):
+        parameters = {'s': 10, 'a': 2, 'not': 13}
+        s = VolatileRepetitionCount(expression=ExpressionScalar('s'),
+                                    scope=DictScope(values=FrozenDict(s=3), volatile=set('s')))
+        a = VolatileRepetitionCount(expression=ExpressionScalar('a'),
+                                    scope=DictScope(values=FrozenDict(a=5), volatile=set('a')))
+
+        wf_1 = DummyWaveform(defined_channels={'A'}, duration=1)
+        wf_2 = DummyWaveform(defined_channels={'A'}, duration=1)
+
+        program = Loop(children=[Loop(children=[Loop(waveform=wf_1, repetition_count=s),
+                                                Loop(waveform=wf_2, repetition_count=4),
+                                                Loop(waveform=wf_1, repetition_count=2)],
+                                      repetition_count=4),
+                                 Loop(children=[Loop(waveform=wf_2, repetition_count=5),
+                                                Loop(waveform=wf_1, repetition_count=s),
+                                                Loop(waveform=wf_2, repetition_count=5)],
+                                      repetition_count=a)],
+                       repetition_count=1)
+
+        t_program = TaborProgram(program, channels=(None, 'A'), markers=(None, None),
+                                 device_properties=self.instr_props, **self.program_entry_kwargs)
+
+        self.assertEqual(t_program.get_sequencer_tables(), [[(TableDescription(3, 0, 0), s.volatile_property),
+                                                             (TableDescription(4, 1, 0), None),
+                                                             (TableDescription(2, 0, 0), None)],
+                                                            [(TableDescription(5, 1, 0), None),
+                                                             (TableDescription(3, 0, 0), s.volatile_property),
+                                                             (TableDescription(5, 1, 0), None)]
+                                                            ])
+        self.assertEqual(t_program.get_advanced_sequencer_table(), [TableEntry(4, 1, 0), TableEntry(5, 2, 0)])
+
+        modifications = t_program.update_volatile_parameters(parameters)
+
+        expected_seq = VolatileRepetitionCount(expression=ExpressionScalar('s'),
+                                               scope=DictScope(values=FrozenDict(s=10), volatile=set('s')))
+        expected_modifications = {(0, 0): TableDescription(10, 0, 0), (1, 1): TableDescription(10, 0, 0),
+                                  1: TableEntry(2, 2, 0)}
+
+        self.assertEqual(t_program.get_sequencer_tables(), [[(TableDescription(10, 0, 0), expected_seq.volatile_property),
+                                                             (TableDescription(4, 1, 0), None),
+                                                             (TableDescription(2, 0, 0), None)],
+                                                            [(TableDescription(5, 1, 0), None),
+                                                             (TableDescription(10, 0, 0), expected_seq.volatile_property),
+                                                             (TableDescription(5, 1, 0), None)]
+                                                            ])
+        self.assertEqual(t_program.get_advanced_sequencer_table(), [TableEntry(4, 1, 0), TableEntry(2, 2, 0)])
+        self.assertEqual(modifications, expected_modifications)
+
 
 class TaborSegmentTests(unittest.TestCase):
     @staticmethod

tests/hardware/tabor_dummy_based_tests.py

@@ -212,7 +212,7 @@ def test__execute_multiple_commands_with_config_guard(self):
     def test_set_volatile_parameters(self):
         channel_pair = self.TaborChannelPair(self.instrument, identifier='asd', channels=(1, 2))
 
-        parameters = {'var': ConstantParameter(2)}
+        parameters = {'var': 2}
         modifications = {1: TableEntry(repetition_count=5, element_number=1, jump_flag=0),
                          (0, 1): TableDescription(repetition_count=10, element_id=0, jump_flag=0)}
         invalid_modification = {1: TableEntry(repetition_count=0, element_number=1, jump_flag=0)}

tests/hardware/tabor_simulator_based_tests.py

@@ -262,7 +262,7 @@ def test_set_volatile_parameter(self):
         self.channel_pair._amend_segments(self.segments)
         self.arm_program(self.sequence_tables, self.advanced_sequence_table, None, np.asarray([1, 2]))
 
-        para = {'a': ConstantParameter(5)}
+        para = {'a': 5}
         actual_sequence_tables = [self.channel_pair._idle_sequence_table] + [[(rep, index + 2, jump)
                                                                               for rep, index, jump in table]
                                                                              for table in self.sequence_tables_raw]