Merge 5558f79 into 17dc51f

qctoolkit/_program/_loop.py

@@ -43,7 +43,8 @@ def __init__(self,
 
     @property
     def compare_key(self) -> Tuple:
-        return self._waveform, self.repetition_count, self._measurements, super().compare_key
+        return self._waveform, self.repetition_count, self._measurements if self._measurements else None,\
+               super().compare_key
 
     def append_child(self, loop: Optional['Loop']=None, **kwargs) -> None:
         # do not invalidate but update cached duration

qctoolkit/_program/transformation.py

@@ -1,14 +1,16 @@
-from typing import Mapping, Set, Dict
+from typing import Mapping, Set, Dict, Tuple
 from abc import abstractmethod
 
 import numpy as np
 import pandas as pd
 
 from qctoolkit import ChannelID
 from qctoolkit.comparable import Comparable
+from qctoolkit.utils.types import SingletonABCMeta
 
 
 class Transformation(Comparable):
+    _identity_singleton = None
     """Transforms numeric time-voltage values for multiple channels to other time-voltage values. The number and names
      of input and output channels might differ."""
 
@@ -27,6 +29,53 @@ def __call__(self, time: np.ndarray, data: pd.DataFrame) -> pd.DataFrame:
     def get_output_channels(self, input_channels: Set[ChannelID]) -> Set[ChannelID]:
         """Return the channel identifiers"""
 
+    def chain(self, next_transformation: 'Transformation') -> 'Transformation':
+        if next_transformation is IdentityTransformation():
+            return self
+        else:
+            return chain_transformations(self, next_transformation)
+
+
+class IdentityTransformation(Transformation, metaclass=SingletonABCMeta):
+    def __call__(self, time: np.ndarray, data: pd.DataFrame) -> pd.DataFrame:
+        return data
+
+    def get_output_channels(self, input_channels: Set[ChannelID]) -> Set[ChannelID]:
+        return input_channels
+
+    @property
+    def compare_key(self) -> None:
+        return None
+
+    def chain(self, next_transformation: Transformation) -> Transformation:
+        return next_transformation
+
+
+class ChainedTransformation(Transformation):
+    def __init__(self, *transformations: Transformation):
+        self._transformations = transformations
+
+    @property
+    def transformations(self) -> Tuple[Transformation, ...]:
+        return self._transformations
+
+    def get_output_channels(self, input_channels: Set[ChannelID]) -> Set[ChannelID]:
+        for transformation in self._transformations:
+            input_channels = transformation.get_output_channels(input_channels)
+        return input_channels
+
+    def __call__(self, time: np.ndarray, data: pd.DataFrame) -> pd.DataFrame:
+        for transformation in self._transformations:
+            data = transformation(time, data)
+        return data
+
+    @property
+    def compare_key(self) -> Tuple[Transformation, ...]:
+        return self._transformations
+
+    def chain(self, next_transformation) -> 'ChainedTransformation':
+        return chain_transformations(*self.transformations, next_transformation)
+
 
 class LinearTransformation(Transformation):
     def __init__(self, transformation_matrix: pd.DataFrame):
@@ -53,3 +102,20 @@ def get_output_channels(self, input_channels: Set[ChannelID]) -> Set[ChannelID]:
     @property
     def compare_key(self) -> Dict[ChannelID, Dict[ChannelID, float]]:
         return self._matrix.to_dict()
+
+
+def chain_transformations(*transformations: Transformation) -> Transformation:
+    parsed_transformations = []
+    for transformation in transformations:
+        if transformation is IdentityTransformation():
+            pass
+        elif isinstance(transformation, ChainedTransformation):
+            parsed_transformations.extend(transformation.transformations)
+        else:
+            parsed_transformations.append(transformation)
+    if len(parsed_transformations) == 0:
+        return IdentityTransformation()
+    elif len(parsed_transformations) == 1:
+        return parsed_transformations[0]
+    else:
+        return ChainedTransformation(*parsed_transformations)

qctoolkit/pulses/loop_pulse_template.py

@@ -232,6 +232,8 @@ def _internal_create_program(self, *,
                                  parameters: Dict[str, Parameter],
                                  measurement_mapping: Dict[str, Optional[str]],
                                  channel_mapping: Dict[ChannelID, Optional[ChannelID]],
+                                 global_transformation: Optional['Transformation'],
+                                 to_single_waveform: Set[Union[str, 'PulseTemplate']],
                                  parent_loop: Loop) -> None:
         self.validate_parameter_constraints(parameters=parameters)
 
@@ -242,16 +244,19 @@ def _internal_create_program(self, *,
                                       for parameter_name in self.duration.variables}
         except KeyError as e:
             raise ParameterNotProvidedException(str(e)) from e
-        measurements = self.get_measurement_windows(measurement_parameters, measurement_mapping)
+
         if self.duration.evaluate_numeric(**duration_parameters) > 0:
+            measurements = self.get_measurement_windows(measurement_parameters, measurement_mapping)
             if measurements:
                 parent_loop.add_measurements(measurements)
 
             for local_parameters in self._body_parameter_generator(parameters, forward=True):
-                self.body._internal_create_program(parameters=local_parameters,
-                                                   measurement_mapping=measurement_mapping,
-                                                   channel_mapping=channel_mapping,
-                                                   parent_loop=parent_loop)
+                self.body._create_program(parameters=local_parameters,
+                                          measurement_mapping=measurement_mapping,
+                                          channel_mapping=channel_mapping,
+                                          global_transformation=global_transformation,
+                                          to_single_waveform=to_single_waveform,
+                                          parent_loop=parent_loop)
 
     def build_waveform(self, parameters: Dict[str, Parameter]) -> ForLoopWaveform:
         return ForLoopWaveform([self.body.build_waveform(local_parameters)

qctoolkit/pulses/mapping_pulse_template.py

@@ -271,12 +271,16 @@ def _internal_create_program(self, *,
                                  parameters: Dict[str, Parameter],
                                  measurement_mapping: Dict[str, Optional[str]],
                                  channel_mapping: Dict[ChannelID, Optional[ChannelID]],
+                                 global_transformation: Optional['Transformation'],
+                                 to_single_waveform: Set[Union[str, 'PulseTemplate']],
                                  parent_loop: Loop) -> None:
         # parameters are validated in map_parameters() call, no need to do it here again explicitly
-        self.template._internal_create_program(parameters=self.map_parameters(parameters),
-                                               measurement_mapping=self.get_updated_measurement_mapping(measurement_mapping),
-                                               channel_mapping=self.get_updated_channel_mapping(channel_mapping),
-                                               parent_loop=parent_loop)
+        self.template._create_program(parameters=self.map_parameters(parameters),
+                                      measurement_mapping=self.get_updated_measurement_mapping(measurement_mapping),
+                                      channel_mapping=self.get_updated_channel_mapping(channel_mapping),
+                                      global_transformation=global_transformation,
+                                      to_single_waveform=to_single_waveform,
+                                      parent_loop=parent_loop)
 
     def build_waveform(self,
                        parameters: Dict[str, numbers.Real],
@@ -286,6 +290,14 @@ def build_waveform(self,
             parameters=self.map_parameters(parameters),
             channel_mapping=self.get_updated_channel_mapping(channel_mapping))
 
+    def get_measurement_windows(self,
+                                parameters: Dict[str, numbers.Real],
+                                measurement_mapping: Dict[str, Optional[str]]) -> List:
+        return self.template.get_measurement_windows(
+            parameters=self.map_parameters(parameters=parameters),
+            measurement_mapping=self.get_updated_measurement_mapping(measurement_mapping=measurement_mapping)
+        )
+
     def requires_stop(self,
                       parameters: Dict[str, Parameter],
                       conditions: Dict[str, Condition]) -> bool:

qctoolkit/pulses/pulse_template.py

@@ -7,20 +7,22 @@
         directly translated into a waveform.
 """
 from abc import abstractmethod
-from typing import Dict, Tuple, Set, Optional, Union, List
+from typing import Dict, Tuple, Set, Optional, Union, List, Callable, Any, Generic, TypeVar, Mapping
 import itertools
+from contextlib import contextmanager
 from numbers import Real
 
 from qctoolkit.utils.types import ChannelID, DocStringABCMeta
 from qctoolkit.serialization import Serializable
 from qctoolkit.expressions import ExpressionScalar
-from qctoolkit._program._loop import Loop
+from qctoolkit._program._loop import Loop, to_waveform
+from qctoolkit._program.transformation import Transformation, IdentityTransformation, ChainedTransformation, chain_transformations
 
 
 from qctoolkit.pulses.conditions import Condition
 from qctoolkit.pulses.parameters import Parameter, ConstantParameter, ParameterNotProvidedException
 from qctoolkit.pulses.sequencing import Sequencer, SequencingElement, InstructionBlock
-from qctoolkit._program.waveforms import Waveform
+from qctoolkit._program.waveforms import Waveform, TransformingWaveform
 from qctoolkit.pulses.measurement import MeasurementDefiner, MeasurementDeclaration
 
 __all__ = ["PulseTemplate", "AtomicPulseTemplate", "DoubleParameterNameException", "MappingTuple"]
@@ -98,7 +100,9 @@ def integral(self) -> Dict[ChannelID, ExpressionScalar]:
     def create_program(self, *,
                        parameters: Optional[Dict[str, Parameter]]=None,
                        measurement_mapping: Optional[Dict[str, Optional[str]]]=None,
-                       channel_mapping: Optional[Dict[ChannelID, Optional[ChannelID]]]=None) -> Optional['Loop']:
+                       channel_mapping: Optional[Dict[ChannelID, Optional[ChannelID]]]=None,
+                       global_transformation: Optional[Transformation]=None,
+                       to_single_waveform: Set[Union[str, 'PulseTemplate']]=None) -> Optional['Loop']:
         """Translates this PulseTemplate into a program Loop.
 
         The returned Loop represents the PulseTemplate with all parameter values instantiated provided as dictated by
@@ -108,6 +112,9 @@ def create_program(self, *,
         :param parameters: A mapping of parameter names to Parameter objects.
         :param measurement_mapping: A mapping of measurement window names. Windows that are mapped to None are omitted.
         :param channel_mapping: A mapping of channel names. Channels that are mapped to None are omitted.
+        :param global_transformation: This transformation is applied to every waveform
+        :param to_single_waveform: A set of pulse templates (or identifiers) which are directly translated to a
+        waveform. This might change how transformations are applied. TODO: clarify
         :return: A Loop object corresponding to this PulseTemplate.
         """
         if parameters is None:
@@ -116,6 +123,8 @@ def create_program(self, *,
             measurement_mapping = {name: name for name in self.measurement_names}
         if channel_mapping is None:
             channel_mapping = dict()
+        if to_single_waveform is None:
+            to_single_waveform = set()
 
         # make sure all values in the parameters dict are of type Parameter
         for (key, value) in parameters.items():
@@ -124,10 +133,12 @@ def create_program(self, *,
 
         root_loop = Loop()
         # call subclass specific implementation
-        self._internal_create_program(parameters=parameters,
-                                      measurement_mapping=measurement_mapping,
-                                      channel_mapping=channel_mapping,
-                                      parent_loop=root_loop)
+        self._create_program(parameters=parameters,
+                             measurement_mapping=measurement_mapping,
+                             channel_mapping=channel_mapping,
+                             global_transformation=global_transformation,
+                             to_single_waveform=to_single_waveform,
+                             parent_loop=root_loop)
 
         if root_loop.waveform is None and len(root_loop.children) == 0:
             return None # return None if no program
@@ -138,6 +149,8 @@ def _internal_create_program(self, *,
                                  parameters: Dict[str, Parameter],
                                  measurement_mapping: Dict[str, Optional[str]],
                                  channel_mapping: Dict[ChannelID, Optional[ChannelID]],
+                                 global_transformation: Optional[Transformation],
+                                 to_single_waveform: Set[Union[str, 'PulseTemplate']],
                                  parent_loop: Loop) -> None:
         """The subclass specific implementation of create_program().
 
@@ -153,6 +166,47 @@ def _internal_create_program(self, *,
         implementations of this method must throw an adequate exception. They do not have to ensure that the parent_loop
         remains unchanged in this case."""
 
+    def _create_program(self, *,
+                        parameters: Dict[str, Parameter],
+                        measurement_mapping: Dict[str, Optional[str]],
+                        channel_mapping: Dict[ChannelID, Optional[ChannelID]],
+                        global_transformation: Optional[Transformation],
+                        to_single_waveform: Set[Union[str, 'PulseTemplate']],
+                        parent_loop: Loop):
+        """Generic part of create program. This method handles to_single_waveform and the configuration of the
+        transformer."""
+        if self.identifier in to_single_waveform or self in to_single_waveform:
+            root = Loop()
+
+            self._internal_create_program(parameters=parameters,
+                                          measurement_mapping=measurement_mapping,
+                                          channel_mapping=channel_mapping,
+                                          global_transformation=None,
+                                          to_single_waveform=to_single_waveform,
+                                          parent_loop=root)
+
+            waveform = to_waveform(root)
+
+            if global_transformation:
+                waveform = TransformingWaveform(waveform, global_transformation)
+
+            # convert the nicely formatted measurement windows back into the old format again :(
+            measurements = root.get_measurement_windows()
+            measurement_window_list = []
+            for measurement_name, (begins, lengths) in measurements.items():
+                measurement_window_list.extend(zip(itertools.repeat(measurement_name), begins, lengths))
+
+            parent_loop.add_measurements(measurement_window_list)
+            parent_loop.append_child(waveform=waveform)
+
+        else:
+            self._internal_create_program(parameters=parameters,
+                                          measurement_mapping=measurement_mapping,
+                                          channel_mapping=channel_mapping,
+                                          to_single_waveform=to_single_waveform,
+                                          global_transformation=global_transformation,
+                                          parent_loop=parent_loop)
+
 
 class AtomicPulseTemplate(PulseTemplate, MeasurementDefiner):
     """A PulseTemplate that does not imply any control flow disruptions and can be directly
@@ -185,7 +239,7 @@ def build_sequence(self,
         parameters = {parameter_name: parameter_value.get_value()
                       for parameter_name, parameter_value in parameters.items()
                       if parameter_name in self.parameter_names}
-        waveform = self.build_waveform(parameters,
+        waveform = self.build_waveform(parameters=parameters,
                                        channel_mapping=channel_mapping)
         if waveform:
             measurements = self.get_measurement_windows(parameters=parameters, measurement_mapping=measurement_mapping)
@@ -196,7 +250,11 @@ def _internal_create_program(self, *,
                                  parameters: Dict[str, Parameter],
                                  measurement_mapping: Dict[str, Optional[str]],
                                  channel_mapping: Dict[ChannelID, Optional[ChannelID]],
+                                 global_transformation: Optional[Transformation],
+                                 to_single_waveform: Set[Union[str, 'PulseTemplate']],
                                  parent_loop: Loop) -> None:
+        """Parameter constraints are validated in build_waveform because build_waveform is guaranteed to be called
+        during sequencing"""
         ### current behavior (same as previously): only adds EXEC Loop and measurements if a waveform exists.
         ### measurements are directly added to parent_loop (to reflect behavior of Sequencer + MultiChannelProgram)
         # todo (2018-08-08): could move measurements into own Loop object?
@@ -205,16 +263,18 @@ def _internal_create_program(self, *,
         try:
             parameters = {parameter_name: parameters[parameter_name].get_value()
                           for parameter_name in self.parameter_names}
-
-            measurement_parameters = {parameter_name: parameters[parameter_name].get_value()
-                                      for parameter_name in self.measurement_parameters}
         except KeyError as e:
             raise ParameterNotProvidedException(str(e)) from e
 
-        measurements = self.get_measurement_windows(parameters=measurement_parameters, measurement_mapping=measurement_mapping)
-        waveform = self.build_waveform(parameters,
+        waveform = self.build_waveform(parameters=parameters,
                                        channel_mapping=channel_mapping)
         if waveform:
+            measurements = self.get_measurement_windows(parameters=parameters,
+                                                        measurement_mapping=measurement_mapping)
+
+            if global_transformation:
+                waveform = TransformingWaveform(waveform, global_transformation)
+
             parent_loop.add_measurements(measurements=measurements)
             parent_loop.append_child(waveform=waveform)