Merge remote-tracking branch 'origin/feat/linspace_measurements' into feat/linspace_measurements/atssimple

Author: maxbeer99

qupulse/examples/expectation_checker.py

@@ -0,0 +1,142 @@
+# -*- coding: utf-8 -*-
+import numpy as np
+
+from qupulse.pulses import PointPT, ConstantPT, RepetitionPT, ForLoopPT, TablePT,\
+    FunctionPT, AtomicMultiChannelPT, SequencePT, MappingPT, ParallelConstantChannelPT
+from qupulse.program.linspace import LinSpaceBuilder
+
+from qupulse.plotting import plot
+from qupulse.utils import to_next_multiple
+
+from expectation_checker import ExpectationChecker, HDAWGAlazar
+
+#%% Get Devices
+
+# Get the Alazar and the HDAWG in hardcoded configuration (2V p-p range)
+# Connect any Marker from the HDAWG to Trig in of the Alazar
+# Connect one dummy channel from HDAWG to the external clock of the Alazar,
+# then set 0.5V-10MHz-oscillator on this channel (e.g. in HDAWG-Webinterface)
+
+ha = HDAWGAlazar("DEVXXXX","USB",)
+
+#%% Example pulse definitions
+
+# PulseTemplates must be defined on channels named as 'ZI0_X', X in A to H
+# (ensure correct mapping to Alazar)
+# Markers will be overwritten to play a marker on each channel to trigger the Alazar
+# identifiers of the final PT will be the names of the plotted object
+
+
+class ShortSingleRampTest():
+    def __init__(self, base_time=1e3):
+        hold = ConstantPT(base_time, {'a': '-1. + idx * 0.01'})
+        pt = hold.with_iteration('idx', 200)
+        self.pulse_template = MappingPT(pt,
+                                        channel_mapping={'a':'ZI0_A',},
+                                        identifier=self.__class__.__name__
+                                        )
+
+class ShortSingleRampTestWithPlay():
+    def __init__(self,base_time=1e3+8):
+        # init = PointPT([(1.0,1e4)],channel_names=('ZI0_MARKER_FRONT',))
+        init = FunctionPT('1.0+1e-9*t',base_time,channel='ZI0_A_MARKER_FRONT')#.pad_to(to_next_multiple(1.0,16,4),)
+
+        hold = ConstantPT(base_time, {'a': '-1. + idx * 0.01'})#.pad_to(to_next_multiple(1.0,16,4))
+        pt = ParallelConstantChannelPT(init,dict(ZI0_A=0.))@(ParallelConstantChannelPT(hold,dict(ZI0_A_MARKER_FRONT=0.)).with_iteration('idx', 200))
+        self.pulse_template = MappingPT(pt,
+                                        channel_mapping={'a':'ZI0_A',},
+                                        identifier=self.__class__.__name__
+                                        )
+
+class SequencedRepetitionTest():
+    def __init__(self,base_time=1e2,rep_factor=2):
+        wait = AtomicMultiChannelPT(
+            ConstantPT(f'64*{base_time}', {'a': '-1. + idx_a * 0.01 + y_gain', }),
+            ConstantPT(f'64*{base_time}', {'b': '-0.5 + idx_b * 0.02'})
+            )
+        
+        dependent_constant = AtomicMultiChannelPT(
+            ConstantPT(64*base_time, {'a': '-1.0 + y_gain'}),
+            ConstantPT(64*base_time, {'b': '-0.5 + idx_b*0.02',}),            
+            )
+        
+        dependent_constant2 = AtomicMultiChannelPT(
+            ConstantPT(64*base_time, {'a': '-0.5 + y_gain'}),
+            ConstantPT(64*base_time, {'b': '-0.3 + idx_b*0.02',}),            
+            )
+        
+    
+        pt = (dependent_constant @ dependent_constant2.with_repetition(rep_factor) @ (wait.with_iteration('idx_a', rep_factor))).with_iteration('idx_b', rep_factor)\
+
+        self.pulse_template = MappingPT(pt,parameter_mapping=dict(y_gain=0.3,),
+                                        channel_mapping={'a':'ZI0_A','b':'ZI0_C'},
+                                        identifier=self.__class__.__name__
+                                        )
+
+
+class SteppedRepetitionTest():
+    def __init__(self,base_time=1e2,rep_factor=2):
+
+        wait = ConstantPT(f'64*{base_time}*(1+idx_t)', {'a': '-0.5 + idx_a * 0.15', 'b': '-.5 + idx_a * 0.3'})
+        normal_pt = ParallelConstantChannelPT(FunctionPT("sin(t/1000)","t_sin",channel='a'),{'b':-0.2})
+        amp_pt = ParallelConstantChannelPT("amp*1/8"*FunctionPT("sin(t/2000)","t_sin",channel='a'),{'b':-0.5})
+        # amp_pt2 = ParallelConstantChannelPT("amp2*1/8"*FunctionPT("sin(t/1000)","t_sin",channel='a'),{'b':-0.5})
+        amp_inner = ParallelConstantChannelPT(FunctionPT(f"(1+amp)*1/(2*{rep_factor})*sin(4*pi*t/t_sin)","t_sin",channel='a'),{'b':-0.5})
+        amp_inner2 = ParallelConstantChannelPT(FunctionPT(f"(1+amp2)*1/(2*{rep_factor})*sin((1*freq)*4*pi*t/t_sin)+off/(2*{rep_factor})","t_sin",channel='a'),{'b':-0.3})
+
+        pt = ((((normal_pt@amp_inner2).with_iteration('off', rep_factor)@normal_pt@wait)\
+              .with_repetition(rep_factor))@amp_inner.with_iteration('amp', rep_factor))\
+              .with_iteration('amp2', rep_factor).with_iteration('freq', rep_factor).with_iteration('idx_a',rep_factor)
+       
+        self.pulse_template = MappingPT(pt,parameter_mapping=dict(t_sin=64*base_time,idx_t=1,
+                                                                  #idx_a=1,#freq=1,#amp2=1
+                                                                  ),
+                                        channel_mapping={'a':'ZI0_A','b':'ZI0_C'},
+                                        identifier=self.__class__.__name__)
+
+class TimeSweepTest():
+    def __init__(self,base_time=1e2,rep_factor=3):
+        wait = ConstantPT(f'64*{base_time}*(1+idx_t)',
+                          {'a': '-1. + idx_a * 0.01', 'b': '-.5 + idx_b * 0.02'})
+    
+        random_constant = ConstantPT(64*base_time, {'a': -.4, 'b': -.3})
+        meas = ConstantPT(64*base_time, {'a': 0.05, 'b': 0.06})
+    
+        singlet_scan = (SequencePT(random_constant,wait,meas,identifier='s')).with_iteration('idx_a', rep_factor)\
+                                                      .with_iteration('idx_b', rep_factor)\
+                                                      .with_iteration('idx_t', rep_factor)
+                                                      
+        self.pulse_template = MappingPT(singlet_scan,channel_mapping={'a':'ZI0_A','b':'ZI0_C'},
+                                        identifier=self.__class__.__name__)
+
+
+#%% Instantiate Checker
+
+# select exemplary pulse
+pulse = ShortSingleRampTest(1e3+8)
+# pulse = ShortSingleRampTestWithPlay()
+# pulse = SequencedRepetitionTest(1e3,4)
+# pulse = SteppedRepetitionTest(1e2,3)
+# pulse = TimeSweepTest(1e2,3)
+
+# Define a program builder to test program with:
+program_builder = LinSpaceBuilder(
+    #set to True to ensure triggering at Program start if program starts with constant pulse
+    play_marker_when_constant=True,
+    #in case stepped repetitions are needed, insert variables here: 
+    to_stepping_repeat={'example',},
+    )
+
+# Data will be saved as xr.Dataset in save_path
+# data_offsets corrects for offsets in Alazar (not in saved data, only in plotting)
+checker = ExpectationChecker(ha, pulse.pulse_template,
+                             program_builder=program_builder,
+                             save_path=SAVE_HERE,
+                             data_offsets={'t_offset':-100.,'v_offset':0.008,'v_scale':0.9975}
+                             )
+
+assert float(pulse.pulse_template.duration) < 1e7, "Ensure you know what you're doing when recording long data"
+
+#%% Run the checker
+
+checker.run()

qupulse/examples/expectation_checker_example_script.py

@@ -226,7 +226,7 @@ def __init__(self, program: AllowedProgramTypes,
         if program_type == _ProgramType.Linspace:
             #!!! the voltage resolution may not be adequately represented if voltage transformations are not None?
             self._transformed_commands = self._transform_linspace_commands(to_increment_commands(program,))
-        
+            
         if waveforms is None:
             if program_type is _ProgramType.Loop:
                     waveforms = OrderedDict((node.waveform, None)

qupulse/hardware/awgs/base.py

@@ -390,13 +390,16 @@ def register_mask_for_channel(self, mask_id: str, hw_channel: int, mask_type='au
             raise NotImplementedError('Currently only can do cross buffer mask')
         self._mask_prototypes[mask_id] = (hw_channel, mask_type)
 
-    def measure_program(self, channels: Iterable[str]) -> Dict[str, np.ndarray]:
+    def measure_program(self, channels: Iterable[str] = None) -> Dict[str, np.ndarray]:
         """
         Get all measurements at once and write them in a dictionary.
         """
 
         scanline_data = self.__card.extractNextScanline()
-
+        
+        if channels is None:
+            channels = scanline_data.operationResults.keys()
+        
         scanline_definition = scanline_data.definition
         operation_definitions = {operation.identifier: operation
                                  for operation in scanline_definition.operations}

qupulse/hardware/dacs/alazar.py

@@ -111,7 +111,8 @@ def register_program(self, name: str,
             raise TypeError('The provided run_callback is not callable')
 
         if channels is None:
-            channels = next(program.get_depth_first_iterator()).waveform.defined_channels
+            # channels = next(program.get_depth_first_iterator()).waveform.defined_channels
+            channels = program.get_defined_channels()
         if channels - set(self._channel_map.keys()):
             raise KeyError('The following channels are unknown to the HardwareSetup: {}'.format(
                 channels - set(self._channel_map.keys())))

qupulse/hardware/setup.py

@@ -28,7 +28,7 @@
 from qupulse.pulses.pulse_template import PulseTemplate
 from qupulse.program.waveforms import Waveform
 from qupulse.program.loop import Loop, to_waveform
-
+from qupulse.program.linspace import LinSpaceTopLevel
 
 __all__ = ["render", "plot", "PlottingNotPossibleException"]
 
@@ -37,7 +37,9 @@ def render(program: Union[Loop],
            sample_rate: Real = 10.0,
            render_measurements: bool = False,
            time_slice: Tuple[Real, Real] = None,
-           plot_channels: Optional[Set[ChannelID]] = None) -> Tuple[np.ndarray, Dict[ChannelID, np.ndarray],
+           plot_channels: Optional[Set[ChannelID]] = None,
+           individualize_times: bool = False,
+           ) -> Tuple[np.ndarray, Dict[ChannelID, np.ndarray],
                                                                     List[MeasurementWindow]]:
     """'Renders' a pulse program.
 
@@ -50,7 +52,7 @@ def render(program: Union[Loop],
             render_measurements: If True, the third return value is a list of measurement windows.
             time_slice: The time slice to be rendered. If None, the entire pulse will be shown.
             plot_channels: Only channels in this set are rendered. If None, all will.
-
+            individualize_times: return individual time-voltage array pairs for constant value cleanup
         Returns:
             A tuple (times, values, measurements). times is a numpy.ndarray of dimensions sample_count where
             containing the time values. voltages is a dictionary of one numpy.ndarray of dimensions sample_count per
@@ -100,10 +102,36 @@ def render(program: Union[Loop],
                 for ch in channels}
     for ch, ch_voltage in voltages.items():
         waveform.get_sampled(channel=ch, sample_times=times, output_array=ch_voltage)
-
+        
+        
+    if individualize_times:
+        # new_dict = {ch: (np.copy(times),volts) for ch,volts in voltages.items()}
+        new_dict = {}
+        for ch in channels:
+            new_dict[ch] = deduplicate_with_aux(voltages[ch],np.copy(times),)
+        return times, new_dict, measurements
+        
     return times, voltages, measurements
 
 
+def deduplicate_with_aux(arr, aux, threshold=1e-4):
+    # Calculate the absolute differences between consecutive elements
+    diffs = np.abs(np.diff(arr, prepend=arr[0]))
+    
+    # Use cumsum to track the cumulative differences
+    cumulative_diffs = np.cumsum(diffs)
+    
+    # Find indices where cumulative differences exceed the threshold
+    mask = np.concatenate(([0],np.where(np.diff(np.floor_divide(cumulative_diffs,threshold),prepend=cumulative_diffs[0])>0)[0],[-1]))
+    
+    # Apply the mask to both the main and auxiliary arrays
+    dedup_arr = arr[mask]
+    dedup_aux = aux[mask]
+    
+    return dedup_aux, dedup_arr
+
+
+
 def _render_loop(loop: Loop,
                  render_measurements: bool,) -> Tuple[Waveform, List[MeasurementWindow]]:
     """Transform program into single waveform and measurement windows.
@@ -132,6 +160,7 @@ def plot(pulse: Union[PulseTemplate, Loop],
          stepped: bool=True,
          maximum_points: int=10**6,
          time_slice: Tuple[Real, Real]=None,
+         individualize_times: bool = False,
          **kwargs) -> Any:  # pragma: no cover
     """Plots a pulse using matplotlib.
 
@@ -182,7 +211,8 @@ def plot(pulse: Union[PulseTemplate, Loop],
         times, voltages, measurements = render(program,
                                                sample_rate,
                                                render_measurements=bool(plot_measurements),
-                                               time_slice=time_slice)
+                                               time_slice=time_slice,
+                                               individualize_times=individualize_times)
     else:
         times, voltages, measurements = np.array([]), dict(), []
 
@@ -214,9 +244,15 @@ def plot(pulse: Union[PulseTemplate, Loop],
     for ch_name, voltage in voltages.items():
         label = 'channel {}'.format(ch_name)
         if stepped:
-            line, = axes.step(times, voltage, **{**dict(where='post', label=label), **kwargs})
+            if individualize_times:
+                line, = axes.step(voltage[0], voltage[1], **{**dict(where='post', label=label), **kwargs})
+            else:
+                line, = axes.step(times, voltage, **{**dict(where='post', label=label), **kwargs})
         else:
-            line, = axes.plot(times, voltage, **{**dict(label=label), **kwargs})
+            if individualize_times:
+                axes.plot(voltage[0], voltage[1], **{**dict(label=label), **kwargs})
+            else:
+                line, = axes.plot(times, voltage, **{**dict(label=label), **kwargs})
         legend_handles.append(line)
 
     if plot_measurements:
@@ -235,8 +271,8 @@ def plot(pulse: Union[PulseTemplate, Loop],
 
     axes.legend(handles=legend_handles)
 
-    max_voltage = max((max(channel, default=0) for channel in voltages.values()), default=0)
-    min_voltage = min((min(channel, default=0) for channel in voltages.values()), default=0)
+    max_voltage = max((max(channel if not individualize_times else channel[1], default=0) for channel in voltages.values()), default=0)
+    min_voltage = min((min(channel if not individualize_times else channel[1], default=0) for channel in voltages.values()), default=0)
 
     # add some margins in the presentation
     axes.set_xlim(-0.5+time_slice[0], time_slice[1] + 0.5)

qupulse/plotting.py

@@ -1,13 +1,13 @@
 from typing import Optional, Union, Sequence, ContextManager, Mapping, Tuple, Generic, TypeVar, Iterable, Dict
-from typing import Protocol, runtime_checkable
+from typing import Protocol, runtime_checkable, Set
 
 from qupulse._program.waveforms import Waveform
 from qupulse.utils.types import MeasurementWindow, TimeType
 from qupulse._program.volatile import VolatileRepetitionCount
 from qupulse.parameter_scope import Scope
 from qupulse.expressions import sympy as sym_expr
 from qupulse.expressions.simple import SimpleExpression
-
+from qupulse import ChannelID
 
 RepetitionCount = Union[int, VolatileRepetitionCount, SimpleExpression[int]]
 HardwareTime = Union[TimeType, SimpleExpression[TimeType]]
@@ -74,7 +74,7 @@ def with_iteration(self, index_name: str, rng: range,
     def evaluate_nested_stepping(self, scope: Scope, parameter_names: set[str]) -> bool:
         return False
 
-    def to_program(self) -> Optional[Program]:
+    def to_program(self, defined_channels: Set[ChannelID]) -> Optional[Program]:
         """Further addition of new elements might fail after finalizing the program."""