Merge branch 'master' into issues/274_prepare_removing_old_serializat…

…ion_code

Conflicts:
	qctoolkit/pulses/pulse_template_parameter_mapping.py
	tests/pulses/pulse_template_parameter_mapping_tests.py

qctoolkit/_program/_loop.py

@@ -15,8 +15,7 @@
 from qctoolkit.utils.types import MeasurementWindow
 from qctoolkit.utils import is_integer
 
-from qctoolkit.pulses.sequence_pulse_template import SequenceWaveform
-from qctoolkit.pulses.repetition_pulse_template import RepetitionWaveform
+from qctoolkit._program.waveforms import SequenceWaveform, RepetitionWaveform
 
 __all__ = ['Loop', 'MultiChannelProgram', 'make_compatible']
 

qctoolkit/_program/transformation.py

@@ -0,0 +1,55 @@
+from typing import Mapping, Set, Dict
+from abc import abstractmethod
+
+import numpy as np
+import pandas as pd
+
+from qctoolkit import ChannelID
+from qctoolkit.comparable import Comparable
+
+
+class Transformation(Comparable):
+    """Transforms numeric time-voltage values for multiple channels to other time-voltage values. The number and names
+     of input and output channels might differ."""
+
+    @abstractmethod
+    def __call__(self, time: np.ndarray, data: pd.DataFrame) -> pd.DataFrame:
+        """Apply transformation to data
+        Args:
+            time:
+            data:
+
+        Returns:
+            transformed: A DataFrame that has been transformed with index == output_channels
+        """
+
+    @abstractmethod
+    def get_output_channels(self, input_channels: Set[ChannelID]) -> Set[ChannelID]:
+        """Return the channel identifiers"""
+
+
+class LinearTransformation(Transformation):
+    def __init__(self, transformation_matrix: pd.DataFrame):
+        """
+
+        Args:
+            transformation_matrix: columns are input and index are output channels
+        """
+        self._matrix = transformation_matrix
+
+    def __call__(self, time: np.ndarray, data: pd.DataFrame) -> Mapping[ChannelID, np.ndarray]:
+        data_in = pd.DataFrame(data)
+        if set(data_in.index) != set(self._matrix.columns):
+            raise KeyError('Invalid input channels', set(data_in.index), set(self._matrix.columns))
+
+        return self._matrix @ data_in
+
+    def get_output_channels(self, input_channels: Set[ChannelID]) -> Set[ChannelID]:
+        if input_channels != set(self._matrix.columns):
+            raise KeyError('Invalid input channels', input_channels, set(self._matrix.columns))
+
+        return set(self._matrix.index)
+
+    @property
+    def compare_key(self) -> Dict[ChannelID, Dict[ChannelID, float]]:
+        return self._matrix.to_dict()

qctoolkit/_program/waveforms.py

@@ -6,21 +6,23 @@
 
 import itertools
 from abc import ABCMeta, abstractmethod
-from weakref import WeakValueDictionary
-from typing import Union, Set, Sequence, NamedTuple, Tuple, Any, List, Iterable
+from weakref import WeakValueDictionary, ref
+from typing import Union, Set, Sequence, NamedTuple, Tuple, Any, Iterable, FrozenSet, Optional
 
 import numpy as np
+import pandas as pd
 
 from qctoolkit import ChannelID
 from qctoolkit.utils import checked_int_cast
 from qctoolkit.utils.types import TimeType, time_from_float
 from qctoolkit.comparable import Comparable
 from qctoolkit.expressions import ExpressionScalar
 from qctoolkit.pulses.interpolation import InterpolationStrategy
+from qctoolkit._program.transformation import Transformation
 
 
 __all__ = ["Waveform", "TableWaveform", "TableWaveformEntry", "FunctionWaveform", "SequenceWaveform",
-           "MultiChannelWaveform", "RepetitionWaveform"]
+           "MultiChannelWaveform", "RepetitionWaveform", "TransformingWaveform"]
 
 
 class Waveform(Comparable, metaclass=ABCMeta):
@@ -481,3 +483,94 @@ def duration(self) -> TimeType:
     def unsafe_get_subset_for_channels(self, channels: Set[ChannelID]) -> 'RepetitionWaveform':
         return RepetitionWaveform(body=self._body.unsafe_get_subset_for_channels(channels),
                                   repetition_count=self._repetition_count)
+
+
+class TransformingWaveform(Waveform):
+    def __init__(self, inner_waveform: Waveform, transformation: Transformation):
+        """"""
+        self._inner_waveform = inner_waveform
+        self._transformation = transformation
+
+        # cache data of inner channels based identified and invalidated by the sample times
+        self._cached_data = None
+        self._cached_times = lambda: None
+
+    @property
+    def inner_waveform(self) -> Waveform:
+        return self._inner_waveform
+
+    @property
+    def transformation(self) -> Transformation:
+        return self._transformation
+
+    @property
+    def defined_channels(self) -> Set[ChannelID]:
+        return self.transformation.get_output_channels(self.inner_waveform.defined_channels)
+
+    @property
+    def compare_key(self) -> Tuple[Waveform, Transformation]:
+        return self.inner_waveform, self.transformation
+
+    @property
+    def duration(self) -> TimeType:
+        return self.inner_waveform.duration
+
+    def unsafe_get_subset_for_channels(self, channels: Set[ChannelID]) -> 'SubsetWaveform':
+        return SubsetWaveform(self, channel_subset=channels)
+
+    def unsafe_sample(self,
+                      channel: ChannelID,
+                      sample_times: np.ndarray,
+                      output_array: Union[np.ndarray, None] = None) -> np.ndarray:
+        if self._cached_times() is not sample_times:
+            inner_channels = tuple(self.inner_waveform.defined_channels)
+            inner_data = np.empty((len(inner_channels), sample_times.size))
+
+            for idx, inner_channel in enumerate(inner_channels):
+                self.inner_waveform.unsafe_sample(inner_channel, sample_times,
+                                                  output_array=inner_data[idx, :])
+
+            inner_data = pd.DataFrame(inner_data, index=inner_channels)
+
+            outer_data = self.transformation(sample_times, inner_data)
+
+            self._cached_data = outer_data
+            self._cached_times = ref(sample_times)
+
+        if output_array is None:
+            output_array = self._cached_data.loc[channel].values
+        else:
+            output_array[:] = self._cached_data.loc[channel].values
+
+        return output_array
+
+
+class SubsetWaveform(Waveform):
+    def __init__(self, inner_waveform: Waveform, channel_subset: Set[ChannelID]):
+        self._inner_waveform = inner_waveform
+        self._channel_subset = frozenset(channel_subset)
+
+    @property
+    def inner_waveform(self) -> Waveform:
+        return self._inner_waveform
+
+    @property
+    def defined_channels(self) -> FrozenSet[ChannelID]:
+        return self._channel_subset
+
+    @property
+    def duration(self) -> TimeType:
+        return self.inner_waveform.duration
+
+    @property
+    def compare_key(self) -> Tuple[frozenset, Waveform]:
+        return self.defined_channels, self.inner_waveform
+
+    def unsafe_get_subset_for_channels(self, channels: Set[ChannelID]) -> Waveform:
+        return self.inner_waveform.get_subset_for_channels(channels)
+
+    def unsafe_sample(self,
+                      channel: ChannelID,
+                      sample_times: np.ndarray,
+                      output_array: Union[np.ndarray, None]=None) -> np.ndarray:
+        return self.inner_waveform.unsafe_sample(channel, sample_times, output_array)

qctoolkit/pulses/__init__.py

@@ -4,7 +4,7 @@
 from qctoolkit.pulses.function_pulse_template import FunctionPulseTemplate as FunctionPT
 from qctoolkit.pulses.loop_pulse_template import ForLoopPulseTemplate as ForLoopPT
 from qctoolkit.pulses.multi_channel_pulse_template import AtomicMultiChannelPulseTemplate as AtomicMultiChannelPT
-from qctoolkit.pulses.pulse_template_parameter_mapping import MappingPulseTemplate as MappingPT
+from qctoolkit.pulses.mapping_pulse_template import MappingPulseTemplate as MappingPT
 from qctoolkit.pulses.repetition_pulse_template import RepetitionPulseTemplate as RepetitionPT
 from qctoolkit.pulses.sequence_pulse_template import SequencePulseTemplate as SequencePT
 from qctoolkit.pulses.table_pulse_template import TablePulseTemplate as TablePT