Removed register_pulse_callback and thus MATLAB dependency from Pulse…

…ControlInterface.

qctoolkit/qcmatlab/manager.py

@@ -1,8 +1,6 @@
 import matlab.engine
 """ Start a MATLAB session or connect to a connect to an existing one. """
 
-from .pulse_control import PulseControlInterface
-
 __all__ = ["Manager",
            "EngineNotFound",
            "NoConnectionSupported"]
@@ -82,8 +80,3 @@ def get_engine() -> matlab.engine.matlabengine.MatlabEngine:
 
         assert( isinstance(Manager.__engine_store, matlab.engine.matlabengine.MatlabEngine) )
         return Manager.__engine_store
-
-    @staticmethod
-    def create_pulse_control_interface(sample_rate: float, time_scaling: float=0.001) -> PulseControlInterface:
-        register_pulse_callback = lambda x: Manager.getEngine().plsreg(x)
-        return PulseControlInterface(register_pulse_callback, sample_rate, time_scaling)

qctoolkit/qcmatlab/pulse_control.py

@@ -1,6 +1,6 @@
 from math import floor
 import numpy
-from typing import Dict, Any, Callable
+from typing import Any, Dict, List, Tuple
 
 from qctoolkit.pulses.sequencing import SequencingHardwareInterface
 from qctoolkit.pulses.instructions import Waveform, EXECInstruction, STOPInstruction, InstructionSequence
@@ -10,19 +10,20 @@
 
 class PulseControlInterface(SequencingHardwareInterface):
 
-    def __init__(self, register_pulse_callback: Callable[[Dict[str, Any]], int], sample_rate: float, time_scaling: float=0.001) -> None:
+    Pulse = Dict[str, Any]
+    PulseGroup = Dict[str, Any]
+
+    def __init__(self, sample_rate: float, time_scaling: float=0.001) -> None:
         """Initialize PulseControlInterface.
 
         Arguments:
-        pulse_registration_function -- A function which registers the pulse in pulse control and returns its id.
         sample_rate -- The rate in Hz at which waveforms are sampled.
         time_scaling -- A factor that scales the time domain defined in PulseTemplates. Defaults to 0.001, meaning
         that one unit of time in a PulseTemplate corresponds to one microsecond.
         """
         super().__init__()
         self.__sample_rate = sample_rate
         self.__time_scaling = time_scaling
-        self.__register_pulse_callback = register_pulse_callback
 
     @staticmethod
     def __get_waveform_name(waveform: Waveform) -> str:
@@ -36,7 +37,7 @@ def register_waveform(self, waveform: Waveform) -> None:
         for real hardware interfaces
         """
 
-    def create_waveform_struct(self, waveform: Waveform, name: str) -> Dict[str, Any]:
+    def create_waveform_struct(self, waveform: Waveform, name: str) -> 'PulseControlInterface.Pulse':
         """Construct a dictionary adhering to the waveform struct definition in pulse control.
 
         Arguments:
@@ -52,12 +53,17 @@ def create_waveform_struct(self, waveform: Waveform, name: str) -> Dict[str, Any
         # TODO: how to deal with the second channel expected in waveform structs in pulse control?
         return struct
 
-    def create_pulse_group(self, sequence: InstructionSequence, name: str) -> Dict[str, Any]:
+    def create_pulse_group(self, sequence: InstructionSequence, name: str)\
+            -> Tuple['PulseControlInterface.PulseGroup', List['PulseControlInterface.Pulse']]:
         """Construct a dictionary adhering to the pulse group struct definition in pulse control.
 
-        All waveforms in the given InstructionBlock are converted to waveform pulse structs and registered in
-        pulse control with the pulse registration function held by the class. create_pulse_group detects
-        multiple use of waveforms and sets up the pulse group dictionary accordingly.
+        All waveforms in the given InstructionSequence are converted to waveform pulse structs and returned as a list
+        in the second component of the returned tuple. The first component of the result is a pulse group dictionary
+        denoting the sequence of waveforms using their indices in the returned list. create_pulse_group detects multiple
+        use of waveforms and sets up the pulse group dictionary accordingly.
+
+        Note that pulses are not registered in pulse control. To achieve this and update the pulse group struct
+        accordingly, the dedicated MATLAB script has to be invoked.
 
         The function will raise an Exception if the given InstructionBlock does contain branching instructions,
         which are not supported by pulse control.
@@ -77,17 +83,22 @@ def create_pulse_group(self, sequence: InstructionSequence, name: str) -> Dict[s
                            chan=1,
                            ctrl='notrig')
 
-        registered_waveforms = dict()
+        waveform_ids = dict()
+        waveform_structs = list()
 
         for waveform in waveforms:
-            if waveform not in registered_waveforms:
+            if waveform not in waveform_ids:
                 name = self.__get_waveform_name(waveform)
                 waveform_struct = self.create_waveform_struct(waveform, name)
-                registered_waveforms[waveform] = self.__register_pulse_callback(waveform_struct)
-            if pulse_group['pulses'] and pulse_group['pulses'][-1] == registered_waveforms[waveform]:
+                waveform_structs.append(waveform_struct)
+                index = len(waveform_structs) - 1
+                waveform_ids[waveform] = index
+            else:
+                index = waveform_ids[waveform]
+            if pulse_group['pulses'] and pulse_group['pulses'][-1] == index:
                 pulse_group['nrep'][-1] += 1
             else:
-                pulse_group['pulses'].append(registered_waveforms[waveform])
+                pulse_group['pulses'].append(index)
                 pulse_group['nrep'].append(1)
 
-        return pulse_group
+        return (pulse_group, waveform_structs)

tests/pulses/sequencing_dummies.py

@@ -97,7 +97,10 @@ def __init__(self, duration: float=0, sample_output: numpy.ndarray=None) -> None
 
     @property
     def _compare_key(self) -> Any:
-        return id(self)
+        if self.sample_output is not None:
+            return tuple(self.sample_output.tolist())
+        else:
+            return id(self)
 
     @property
     def duration(self) -> float:

tests/qcmatlab/pulse_control_tests.py

@@ -13,7 +13,7 @@ def test_create_waveform_struct(self) -> None:
         expected_samples = numpy.random.rand(11)
 
         waveform = DummyWaveform(duration=1, sample_output=expected_samples)
-        pci = PulseControlInterface(None, sample_rate, time_scaling=1)
+        pci = PulseControlInterface(sample_rate, time_scaling=1)
         result = pci.create_waveform_struct(waveform, name=name)
 
         expected_sample_times = numpy.linspace(0, 1, 11).tolist()
@@ -31,8 +31,8 @@ def test_create_pulse_group_empty(self) -> None:
         sample_rate = 10
         block = DummyInstructionBlock()
 
-        pci = PulseControlInterface(None, sample_rate)
-        result = pci.create_pulse_group(block.compile_sequence(), name=name)
+        pci = PulseControlInterface(sample_rate)
+        (result, _) = pci.create_pulse_group(block.compile_sequence(), name=name)
         expected_result = dict(
             name=name,
             nrep=[],
@@ -57,14 +57,15 @@ def test_create_pulse_group(self) -> None:
         block.add_instruction_exec(wf1a)
 
         registering_function = lambda x: x['data']
-        pci = PulseControlInterface(registering_function, sample_rate, time_scaling=1)
-        result = pci.create_pulse_group(block.compile_sequence(), name=name)
+        pci = PulseControlInterface(sample_rate, time_scaling=1)
+        (result, _) = pci.create_pulse_group(block.compile_sequence(), name=name)
         expected_result = dict(
             name=name,
             nrep=[2, 1, 1],
-            pulses=[registering_function(pci.create_waveform_struct(wf1a, name='')),
-                    registering_function(pci.create_waveform_struct(wf2, name='')),
-                    registering_function(pci.create_waveform_struct(wf1a, name=''))],
+            pulses=[0, 1, 0],
+            #pulses=[registering_function(pci.create_waveform_struct(wf1a, name='')),
+            #        registering_function(pci.create_waveform_struct(wf2, name='')),
+            #        registering_function(pci.create_waveform_struct(wf1a, name=''))],
             chan=1,
             ctrl='notrig'
         )
@@ -76,7 +77,7 @@ def test_create_pulse_group_invalid_instruction(self) -> None:
         block = DummyInstructionBlock()
         block.add_instruction_goto(block.create_embedded_block())
 
-        pci = PulseControlInterface(None, sample_rate)
+        pci = PulseControlInterface(sample_rate)
         with self.assertRaises(Exception):
             pci.create_pulse_group(block.compile_sequence(), name=name)