Merge pull request #453 from /issues/446

Make offset and amplitude handling of the Tabor AWG configurable

qupulse/hardware/awgs/base.py

@@ -16,11 +16,19 @@
 from qupulse._program.instructions import InstructionSequence
 
 __all__ = ["AWG", "Program", "ProgramOverwriteException",
-           "OutOfWaveformMemoryException"]
+           "OutOfWaveformMemoryException", "AWGAmplitudeOffsetHandling"]
 
 Program = InstructionSequence
 
 
+class AWGAmplitudeOffsetHandling:
+    IGNORE_OFFSET = 'ignore_offset'   # Offset is ignored.
+    CONSIDER_OFFSET = 'consider_offset' # Offset is discounted from the waveforms.
+    # TODO OPTIMIZED = 'optimized' # Offset and amplitude are set depending on the waveforms to maximize the waveforms resolution
+
+    _valid = [IGNORE_OFFSET, CONSIDER_OFFSET]
+
+
 class AWG(Comparable):
     """An arbitrary waveform generator abstraction class.
 
@@ -33,11 +41,26 @@ class AWG(Comparable):
 
     def __init__(self, identifier: str):
         self._identifier = identifier
+        self._amplitude_offset_handling = AWGAmplitudeOffsetHandling.IGNORE_OFFSET
 
     @property
     def identifier(self) -> str:
         return self._identifier
 
+    @property
+    def amplitude_offset_handling(self) -> str:
+        return self._amplitude_offset_handling
+
+    @amplitude_offset_handling.setter
+    def amplitude_offset_handling(self, value):
+        """
+        value (str): See possible values at `AWGAmplitudeOffsetHandling`
+        """
+        if value not in AWGAmplitudeOffsetHandling._valid:
+            raise ValueError('"{}" is invalid as AWGAmplitudeOffsetHandling'.format(value))
+
+        self._amplitude_offset_handling = value
+
     @property
     @abstractmethod
     def num_channels(self):

qupulse/hardware/awgs/tabor.py

@@ -18,7 +18,7 @@
 from qupulse.pulses.multi_channel_pulse_template import MultiChannelWaveform
 from qupulse._program._loop import Loop, make_compatible
 from qupulse.hardware.util import voltage_to_uint16, make_combined_wave, find_positions
-from qupulse.hardware.awgs.base import AWG
+from qupulse.hardware.awgs.base import AWG, AWGAmplitudeOffsetHandling
 
 
 assert(sys.byteorder == 'little')
@@ -931,7 +931,15 @@ def upload(self, name: str,
                       self.device.amplitude(self._channels[1]))
 
             voltage_amplitudes = (ranges[0]/2, ranges[1]/2)
-            voltage_offsets = (0, 0)
+
+            if self._amplitude_offset_handling == AWGAmplitudeOffsetHandling.IGNORE_OFFSET:
+                voltage_offsets = (0, 0)
+            elif self._amplitude_offset_handling == AWGAmplitudeOffsetHandling.CONSIDER_OFFSET:
+                voltage_offsets = (self.device.offset(self._channels[0]),
+                                   self.device.offset(self._channels[1]))
+            else:
+                raise ValueError('{} is invalid as AWGAmplitudeOffsetHandling'.format(self._amplitude_offset_handling))
+
             segments, segment_lengths = tabor_program.sampled_segments(sample_rate=sample_rate,
                                                                        voltage_amplitude=voltage_amplitudes,
                                                                        voltage_offset=voltage_offsets,

tests/hardware/tabor_dummy_based_tests.py

@@ -1,11 +1,14 @@
 import sys
 import unittest
+from unittest.mock import patch, MagicMock
 
 from typing import List
 from copy import copy, deepcopy
 
 import numpy as np
 
+from qupulse.hardware.awgs.base import AWGAmplitudeOffsetHandling
+from qupulse.hardware.awgs.tabor import TaborProgram, TaborAWGRepresentation
 from tests.hardware.dummy_modules import import_package
 
 
@@ -220,10 +223,17 @@ def test_upload_exceptions(self):
         with self.assertRaises(ValueError):
             channel_pair.upload('test', program, (1, 2), (3, 4), (lambda x: x,))
 
+        old = channel_pair._amplitude_offset_handling
+        with self.assertRaises(ValueError):
+            channel_pair._amplitude_offset_handling = 'invalid'
+            channel_pair.upload('test', program, (1, None), (None, None), (lambda x: x, lambda x: x))
+        channel_pair._amplitude_offset_handling = old
+
         channel_pair._known_programs['test'] = self.TaborProgramMemory(np.array([0]), None)
         with self.assertRaises(ValueError):
             channel_pair.upload('test', program, (1, 2), (3, 4), (lambda x: x, lambda x: x))
 
+
     def test_upload(self):
         segments = np.array([1, 2, 3, 4, 5])
         segment_lengths = np.array([0, 16, 0, 16, 0], dtype=np.uint16)
@@ -275,6 +285,47 @@ def dummy_amend_segments(segments_):
         finally:
             sys.modules['qupulse.hardware.awgs.tabor'].TaborProgram = to_restore
 
+    def test_upload_offset_handling(self):
+
+        program = self.Loop(waveform=self.TableWaveform(1, [(0, 0.1, self.HoldInterpolationStrategy()),
+                                                            (192, 0.1, self.HoldInterpolationStrategy())]))
+
+        channel_pair = self.TaborChannelPair(self.instrument, identifier='asd', channels=(1, 2))
+
+        channels = (1, None)
+        markers = (None, None)
+
+        tabor_program = TaborProgram(program,
+                                     channels=channels,
+                                     markers=markers,
+                                     device_properties=channel_pair.device.dev_properties)
+
+        test_sample_rate = channel_pair.sample_rate
+        test_amplitudes = (channel_pair.device.amplitude(channel_pair._channels[0]) / 2,
+                           channel_pair.device.amplitude(channel_pair._channels[1]) / 2)
+        test_offset = 0.1
+        test_transform = (lambda x: x, lambda x: x)
+
+        with patch('qupulse.hardware.awgs.tabor.TaborProgram', return_value=tabor_program) as tabor_program_mock, \
+             patch.object(tabor_program, 'sampled_segments', wraps=tabor_program.sampled_segments) as sampled_segments_mock, \
+             patch.object(channel_pair.device, 'offset', return_value=test_offset):
+
+            channel_pair.amplitude_offset_handling = AWGAmplitudeOffsetHandling.CONSIDER_OFFSET
+            channel_pair.upload('test1', program, (1, None), (None, None), test_transform)
+
+            sampled_segments_mock.assert_called_once_with(sample_rate=test_sample_rate,
+                                                          voltage_amplitude=test_amplitudes,
+                                                          voltage_offset=(test_offset, test_offset),
+                                                          voltage_transformation=test_transform)
+
+            channel_pair.amplitude_offset_handling = AWGAmplitudeOffsetHandling.IGNORE_OFFSET
+            channel_pair.upload('test2', program, (1, None), (None, None), test_transform)
+
+            sampled_segments_mock.assert_called_with(sample_rate=test_sample_rate,
+                                                     voltage_amplitude=test_amplitudes,
+                                                     voltage_offset=(0, 0),
+                                                     voltage_transformation=test_transform)
+
     def test_find_place_for_segments_in_memory(self):
         def hash_based_on_dir(ch):
             hash_list = []