Skip to content

Commit

Permalink
Merge pull request #600 from qutech/issues/588_waveform_constant_value
Browse files Browse the repository at this point in the history 
Add automatic constant waveform optimization
  • Loading branch information
@terrorfisch
terrorfisch committed Jul 8, 2021
2 parents 87c1658 + f05e0c7 commit 16bd4f2
Show file tree
Hide file tree
Showing 11 changed files with 1,001 additions and 198 deletions.
1 change: 1 addition & 0 deletions changes.d/588.feature
View file
@@ -0,0 +1 @@
Adds the methods `is_constant`, `constant_value_dict` and `constant_value` to Waveform class to allow more efficient AWG usage.
98 changes: 68 additions & 30 deletions qupulse/_program/transformation.py
View file
@@ -1,4 +1,4 @@
from typing import Mapping, Set, Tuple, Sequence
from typing import Mapping, Set, Tuple, Sequence, AbstractSet, Union
from abc import abstractmethod
from numbers import Real

Expand All @@ -15,7 +15,8 @@ class Transformation(Comparable):
of input and output channels might differ."""

@abstractmethod
def __call__(self, time: np.ndarray, data: Mapping[ChannelID, np.ndarray]) -> Mapping[ChannelID, np.ndarray]:
def __call__(self, time: Union[np.ndarray, float],
data: Mapping[ChannelID, Union[np.ndarray, float]]) -> Mapping[ChannelID, Union[np.ndarray, float]]:
"""Apply transformation to data
Args:
time:
Expand All @@ -26,11 +27,11 @@ def __call__(self, time: np.ndarray, data: Mapping[ChannelID, np.ndarray]) -> Ma
"""

@abstractmethod
def get_output_channels(self, input_channels: Set[ChannelID]) -> Set[ChannelID]:
def get_output_channels(self, input_channels: AbstractSet[ChannelID]) -> AbstractSet[ChannelID]:
"""Return the channel identifiers"""

@abstractmethod
def get_input_channels(self, output_channels: Set[ChannelID]) -> Set[ChannelID]:
def get_input_channels(self, output_channels: AbstractSet[ChannelID]) -> AbstractSet[ChannelID]:
"""Channels that are required for getting data for the requested output channel"""

def chain(self, next_transformation: 'Transformation') -> 'Transformation':
Expand All @@ -39,19 +40,24 @@ def chain(self, next_transformation: 'Transformation') -> 'Transformation':
else:
return chain_transformations(self, next_transformation)

def is_constant_invariant(self):
"""Signals if the transformation always maps constants to constants."""
return False


class IdentityTransformation(Transformation, metaclass=SingletonABCMeta):
def __call__(self, time: np.ndarray, data: Mapping[ChannelID, np.ndarray]) -> Mapping[ChannelID, np.ndarray]:
def __call__(self, time: Union[np.ndarray, float],
data: Mapping[ChannelID, Union[np.ndarray, float]]) -> Mapping[ChannelID, Union[np.ndarray, float]]:
return data

def get_output_channels(self, input_channels: Set[ChannelID]) -> Set[ChannelID]:
def get_output_channels(self, input_channels: AbstractSet[ChannelID]) -> AbstractSet[ChannelID]:
return input_channels

@property
def compare_key(self) -> None:
return None

def get_input_channels(self, output_channels: Set[ChannelID]) -> Set[ChannelID]:
def get_input_channels(self, output_channels: AbstractSet[ChannelID]) -> AbstractSet[ChannelID]:
return output_channels

def chain(self, next_transformation: Transformation) -> Transformation:
Expand All @@ -60,6 +66,10 @@ def chain(self, next_transformation: Transformation) -> Transformation:
def __repr__(self):
return 'IdentityTransformation()'

def is_constant_invariant(self):
"""Signals if the transformation always maps constants to constants."""
return True


class ChainedTransformation(Transformation):
def __init__(self, *transformations: Transformation):
Expand All @@ -69,17 +79,18 @@ def __init__(self, *transformations: Transformation):
def transformations(self) -> Tuple[Transformation, ...]:
return self._transformations

def get_output_channels(self, input_channels: Set[ChannelID]) -> Set[ChannelID]:
def get_output_channels(self, input_channels: AbstractSet[ChannelID]) -> AbstractSet[ChannelID]:
for transformation in self._transformations:
input_channels = transformation.get_output_channels(input_channels)
return input_channels

def get_input_channels(self, output_channels: Set[ChannelID]) -> Set[ChannelID]:
def get_input_channels(self, output_channels: AbstractSet[ChannelID]) -> AbstractSet[ChannelID]:
for transformation in reversed(self._transformations):
output_channels = transformation.get_input_channels(output_channels)
return output_channels

def __call__(self, time: np.ndarray, data: Mapping[ChannelID, np.ndarray]) -> Mapping[ChannelID, np.ndarray]:
def __call__(self, time: Union[np.ndarray, float],
data: Mapping[ChannelID, Union[np.ndarray, float]]) -> Mapping[ChannelID, Union[np.ndarray, float]]:
for transformation in self._transformations:
data = transformation(time, data)
return data
Expand All @@ -88,12 +99,16 @@ def __call__(self, time: np.ndarray, data: Mapping[ChannelID, np.ndarray]) -> Ma
def compare_key(self) -> Tuple[Transformation, ...]:
return self._transformations

def chain(self, next_transformation) -> 'ChainedTransformation':
def chain(self, next_transformation) -> Transformation:
return chain_transformations(*self.transformations, next_transformation)

def __repr__(self):
return 'ChainedTransformation%r' % (self._transformations,)

def is_constant_invariant(self):
"""Signals if the transformation always maps constants to constants."""
return all(trafo.is_constant_invariant() for trafo in self._transformations)


class LinearTransformation(Transformation):
def __init__(self,
Expand Down Expand Up @@ -121,8 +136,11 @@ def __init__(self,
self._matrix = transformation_matrix
self._input_channels = tuple(sorted(input_channels))
self._output_channels = tuple(sorted(output_channels))
self._input_channels_set = frozenset(self._input_channels)
self._output_channels_set = frozenset(self._output_channels)

def __call__(self, time: np.ndarray, data: Mapping[ChannelID, np.ndarray]) -> Mapping[ChannelID, np.ndarray]:
def __call__(self, time: Union[np.ndarray, float],
data: Mapping[ChannelID, Union[np.ndarray, float]]) -> Mapping[ChannelID, Union[np.ndarray, float]]:
data_out = {forwarded_channel: data[forwarded_channel]
for forwarded_channel in set(data.keys()).difference(self._input_channels)}

Expand All @@ -138,24 +156,25 @@ def __call__(self, time: np.ndarray, data: Mapping[ChannelID, np.ndarray]) -> Ma
transformed_data = self._matrix @ data_in

for idx, out_channel in enumerate(self._output_channels):
data_out[out_channel] = transformed_data[idx, :]
data_out[out_channel] = transformed_data[idx, ...]

return data_out

def get_output_channels(self, input_channels: Set[ChannelID]) -> Set[ChannelID]:
if not input_channels.issuperset(self._input_channels):
raise KeyError('Invalid input channels', input_channels, set(self._input_channels))
def get_output_channels(self, input_channels: AbstractSet[ChannelID]) -> AbstractSet[ChannelID]:
if not input_channels >= self._input_channels_set:
# input_channels is not a superset of the required input channels
raise KeyError('Invalid input channels', input_channels, self._input_channels_set)

return input_channels.difference(self._input_channels).union(self._output_channels)
return (input_channels - self._input_channels_set) | self._output_channels_set

def get_input_channels(self, output_channels: Set[ChannelID]) -> Set[ChannelID]:
forwarded = output_channels.difference(self._output_channels)
def get_input_channels(self, output_channels: AbstractSet[ChannelID]) -> AbstractSet[ChannelID]:
forwarded = output_channels - self._output_channels_set
if not forwarded.isdisjoint(self._input_channels):
raise KeyError('Is input channel', forwarded.intersection(self._input_channels))
raise KeyError('Is input channel', forwarded & self._input_channels_set)
elif output_channels.isdisjoint(self._output_channels):
return output_channels
else:
return forwarded.union(self._input_channels)
return forwarded | self._input_channels_set

@property
def compare_key(self) -> Tuple[Tuple[ChannelID], Tuple[ChannelID], bytes]:
Expand All @@ -169,6 +188,10 @@ def __repr__(self):
input_channels=self._input_channels,
output_channels=self._output_channels)

def is_constant_invariant(self):
"""Signals if the transformation always maps constants to constants."""
return True


class OffsetTransformation(Transformation):
def __init__(self, offsets: Mapping[ChannelID, Real]):
Expand All @@ -181,14 +204,15 @@ def __init__(self, offsets: Mapping[ChannelID, Real]):
"""
self._offsets = dict(offsets.items())

def __call__(self, time: np.ndarray, data: Mapping[ChannelID, np.ndarray]) -> Mapping[ChannelID, np.ndarray]:
def __call__(self, time: Union[np.ndarray, float],
data: Mapping[ChannelID, Union[np.ndarray, float]]) -> Mapping[ChannelID, Union[np.ndarray, float]]:
return {channel: channel_values + self._offsets[channel] if channel in self._offsets else channel_values
for channel, channel_values in data.items()}

def get_input_channels(self, output_channels: Set[ChannelID]):
def get_input_channels(self, output_channels: AbstractSet[ChannelID]) -> AbstractSet[ChannelID]:
return output_channels

def get_output_channels(self, input_channels: Set[ChannelID]):
def get_output_channels(self, input_channels: AbstractSet[ChannelID]) -> AbstractSet[ChannelID]:
return input_channels

@property
Expand All @@ -198,19 +222,24 @@ def compare_key(self) -> frozenset:
def __repr__(self):
return 'OffsetTransformation(%r)' % self._offsets

def is_constant_invariant(self):
"""Signals if the transformation always maps constants to constants."""
return True


class ScalingTransformation(Transformation):
def __init__(self, factors: Mapping[ChannelID, Real]):
self._factors = dict(factors.items())

def __call__(self, time: np.ndarray, data: Mapping[ChannelID, np.ndarray]) -> Mapping[ChannelID, np.ndarray]:
def __call__(self, time: Union[np.ndarray, float],
data: Mapping[ChannelID, Union[np.ndarray, float]]) -> Mapping[ChannelID, Union[np.ndarray, float]]:
return {channel: channel_values * self._factors[channel] if channel in self._factors else channel_values
for channel, channel_values in data.items()}

def get_input_channels(self, output_channels: Set[ChannelID]):
def get_input_channels(self, output_channels: AbstractSet[ChannelID]) -> AbstractSet[ChannelID]:
return output_channels

def get_output_channels(self, input_channels: Set[ChannelID]):
def get_output_channels(self, input_channels: AbstractSet[ChannelID]) -> AbstractSet[ChannelID]:
return input_channels

@property
Expand All @@ -220,6 +249,10 @@ def compare_key(self) -> frozenset:
def __repr__(self):
return 'ScalingTransformation(%r)' % self._factors

def is_constant_invariant(self):
"""Signals if the transformation always maps constants to constants."""
return True


try:
import pandas
Expand Down Expand Up @@ -250,7 +283,8 @@ def __init__(self, channels: Mapping[ChannelID, Real]):
self._channels = {channel: float(value)
for channel, value in channels.items()}

def __call__(self, time: np.ndarray, data: Mapping[ChannelID, np.ndarray]) -> Mapping[ChannelID, np.ndarray]:
def __call__(self, time: Union[np.ndarray, float],
data: Mapping[ChannelID, Union[np.ndarray, float]]) -> Mapping[ChannelID, Union[np.ndarray, float]]:
overwritten = {channel: np.full_like(time, fill_value=value, dtype=float)
for channel, value in self._channels.items()}
return {**data, **overwritten}
Expand All @@ -259,15 +293,19 @@ def __call__(self, time: np.ndarray, data: Mapping[ChannelID, np.ndarray]) -> Ma
def compare_key(self) -> Tuple[Tuple[ChannelID, float], ...]:
return tuple(sorted(self._channels.items()))

def get_input_channels(self, output_channels: Set[ChannelID]) -> Set[ChannelID]:
def get_input_channels(self, output_channels: AbstractSet[ChannelID]) -> AbstractSet[ChannelID]:
return output_channels - self._channels.keys()

def get_output_channels(self, input_channels: Set[ChannelID]) -> Set[ChannelID]:
def get_output_channels(self, input_channels: AbstractSet[ChannelID]) -> AbstractSet[ChannelID]:
return input_channels | self._channels.keys()

def __repr__(self):
return 'ParallelConstantChannelTransformation(%r)' % self._channels

def is_constant_invariant(self):
"""Signals if the transformation always maps constants to constants."""
return True


def chain_transformations(*transformations: Transformation) -> Transformation:
parsed_transformations = []
Expand Down

0 comments on commit 16bd4f2

Please sign in to comment.