Merge pull request qutech#552 from qutech/issues/549_initial_and_fina…

…l_value

Add initial_values and final_values

changes.d/549.feature

@@ -0,0 +1,4 @@
+Add `initial_values` and `final_values` attributes to `PulseTemplate`.
+
+This allows pulse template construction that depends on features of arbitrary existing pulses i.e. like extension until
+a certain length.

qupulse/pulses/abstract_pulse_template.py

@@ -146,6 +146,10 @@ def _internal_create_program(self, **kwargs):
                         doc=_PROPERTY_DOC.format(name='integral'))
     parameter_names = property(partial(_get_property, property_name='parameter_names'),
                                doc=_PROPERTY_DOC.format(name='parameter_names'))
+    initial_values = property(partial(_get_property, property_name='initial_values'),
+                              doc=_PROPERTY_DOC.format(name='initial_values'))
+    final_values = property(partial(_get_property, property_name='final_values'),
+                            doc=_PROPERTY_DOC.format(name='final_values'))
 
     __hash__ = None
 

qupulse/pulses/arithmetic_pulse_template.py

@@ -2,6 +2,7 @@
 from typing import Any, Dict, List, Set, Optional, Union, Mapping, FrozenSet, cast, Callable
 from numbers import Real
 import warnings
+import operator
 
 import sympy
 
@@ -18,15 +19,17 @@
     IdentityTransformation
 
 
-def _apply_operation_to_channel_dict(operator: str,
-                                     lhs: Mapping[ChannelID, Any],
-                                     rhs: Mapping[ChannelID, Any]) -> Dict[ChannelID, Any]:
+def _apply_operation_to_channel_dict(lhs: Mapping[ChannelID, Any],
+                                     rhs: Mapping[ChannelID, Any],
+                                     operator_both: Optional[Callable[[Any, Any], Any]],
+                                     rhs_only: Optional[Callable[[Any], Any]]
+                                     ) -> Dict[ChannelID, Any]:
     result = dict(lhs)
     for channel, rhs_value in rhs.items():
         if channel in result:
-            result[channel] = ArithmeticWaveform.operator_map[operator](result[channel], rhs_value)
+            result[channel] = operator_both(result[channel], rhs_value)
         else:
-            result[channel] = ArithmeticWaveform.rhs_only_map[operator](rhs_value)
+            result[channel] = rhs_only(rhs_value)
     return result
 
 
@@ -106,14 +109,30 @@ def duration(self) -> ExpressionScalar:
         """Duration of the lhs operand if it is larger zero. Else duration of the rhs."""
         return ExpressionScalar(sympy.Max(self.lhs.duration, self.rhs.duration))
 
+    def _apply_operation(self, lhs: Mapping[str, Any], rhs: Mapping[str, Any]) -> Dict[str, Any]:
+        operator_both = ArithmeticWaveform.operator_map[self._arithmetic_operator]
+        rhs_only = ArithmeticWaveform.rhs_only_map[self._arithmetic_operator]
+        return _apply_operation_to_channel_dict(lhs, rhs,
+                                                operator_both=operator_both,
+                                                rhs_only=rhs_only)
+
     @property
     def integral(self) -> Dict[ChannelID, ExpressionScalar]:
-        return _apply_operation_to_channel_dict(self._arithmetic_operator, self.lhs.integral, self.rhs.integral)
+        # this is a guard for possible future changes
+        assert self._arithmetic_operator in ('+', '-'), \
+            f"Integral not correctly implemented for '{self._arithmetic_operator}'"
+        return self._apply_operation(self.lhs.integral, self.rhs.integral)
 
     def _as_expression(self) -> Dict[ChannelID, ExpressionScalar]:
-        return _apply_operation_to_channel_dict(self._arithmetic_operator,
-                                                self.lhs._as_expression(),
-                                                self.rhs._as_expression())
+        return self._apply_operation(self.lhs._as_expression(), self.rhs._as_expression())
+
+    @property
+    def initial_values(self) -> Dict[ChannelID, ExpressionScalar]:
+        return self._apply_operation(self.lhs.initial_values, self.rhs.initial_values)
+
+    @property
+    def final_values(self) -> Dict[ChannelID, ExpressionScalar]:
+        return self._apply_operation(self.lhs.final_values, self.rhs.final_values)
 
     def build_waveform(self,
                        parameters: Dict[str, Real],
@@ -173,15 +192,21 @@ def deserialize(cls, serializer: Optional[Serializer] = None, **kwargs) -> 'Arit
 
 
 class ArithmeticPulseTemplate(PulseTemplate):
-    """"""
-
     def __init__(self,
                  lhs: Union[PulseTemplate, ExpressionLike, Mapping[ChannelID, ExpressionLike]],
                  arithmetic_operator: str,
                  rhs: Union[PulseTemplate, ExpressionLike, Mapping[ChannelID, ExpressionLike]],
                  *,
                  identifier: Optional[str] = None):
-        """
+        """Allowed operations
+
+        scalar + pulse_template
+        scalar - pulse_template
+        scalar * pulse_template
+        pulse_template + scalar
+        pulse_template - scalar
+        pulse_template * scalar
+        pulse_template / scalar
 
         Args:
             lhs: Left hand side operand
@@ -385,47 +410,67 @@ def defined_channels(self):
     def duration(self) -> ExpressionScalar:
         return self._pulse_template.duration
 
+    def _scalar_as_dict(self) -> Dict[ChannelID, ExpressionScalar]:
+        if isinstance(self._scalar, ExpressionScalar):
+            return {channel: self._scalar
+                    for channel in self.defined_channels}
+        else:
+            return dict(self._scalar)
+
     @property
     def integral(self) -> Dict[ChannelID, ExpressionScalar]:
         integral = {channel: value.sympified_expression for channel, value in self._pulse_template.integral.items()}
+        scalar = self._scalar_as_dict()
 
-        if isinstance(self._scalar, ExpressionScalar):
-            scalar = {channel: self._scalar.sympified_expression
-                      for channel in self.defined_channels}
+        if self._arithmetic_operator in ('+', '-'):
+            for ch, value in scalar.items():
+                scalar[ch] = value * self.duration.sympified_expression
+
+        return self._apply_operation_to_channel_dict(integral, scalar)
+
+    def _apply_operation_to_channel_dict(self,
+                                         pt_values: Dict[ChannelID, ExpressionScalar],
+                                         scalar_values: Dict[ChannelID, ExpressionScalar]):
+        operator_map = {
+            '+': operator.add,
+            '-': operator.sub,
+            '/': operator.truediv,
+            '*': operator.mul
+        }
+
+        rhs_only_map = {
+            '+': operator.pos,
+            '-': operator.neg,
+            '*': lambda x: x,
+            '/': lambda x: 1 / x
+        }
+
+        if self._pulse_template is self.lhs:
+            lhs, rhs = pt_values, scalar_values
         else:
-            scalar = {channel: value.sympified_expression
-                      for channel, value in self._scalar.items()}
+            lhs, rhs = scalar_values, pt_values
+            # cannot divide by pulse templates
+            operator_map.pop('/')
+            rhs_only_map.pop('/')
 
-        if self._arithmetic_operator == '+':
-            for channel, value in scalar.items():
-                integral[channel] = integral[channel] + (value * self.duration.sympified_expression)
+        operator_both = operator_map.get(self._arithmetic_operator, None)
+        rhs_only = rhs_only_map.get(self._arithmetic_operator, None)
 
-        elif self._arithmetic_operator == '*':
-            for channel, value in scalar.items():
-                integral[channel] = integral[channel] * value
+        return _apply_operation_to_channel_dict(lhs, rhs, operator_both=operator_both, rhs_only=rhs_only)
 
-        elif self._arithmetic_operator == '/':
-            assert self._pulse_template is self.lhs
-            for channel, value in scalar.items():
-                integral[channel] = integral[channel] / value
+    @property
+    def initial_values(self) -> Dict[ChannelID, ExpressionScalar]:
+        return self._apply_operation_to_channel_dict(
+            self._pulse_template.initial_values,
+            self._scalar_as_dict()
+        )
 
-        else:
-            assert self._arithmetic_operator == '-'
-            if self._pulse_template is self.rhs:
-                # we need to negate all existing values
-                for channel, inner_value in integral.items():
-                    if channel in scalar:
-                        integral[channel] = scalar[channel] * self.duration.sympified_expression - inner_value
-                    else:
-                        integral[channel] = -inner_value
-
-            else:
-                for channel, value in scalar.items():
-                    integral[channel] = integral[channel] - value * self.duration.sympified_expression
-
-        for channel, value in integral.items():
-            integral[channel] = ExpressionScalar(value)
-        return integral
+    @property
+    def final_values(self) -> Dict[ChannelID, ExpressionScalar]:
+        return self._apply_operation_to_channel_dict(
+            self._pulse_template.final_values,
+            self._scalar_as_dict()
+        )
 
     @property
     def measurement_names(self) -> Set[str]:

qupulse/pulses/constant_pulse_template.py

@@ -120,3 +120,11 @@ def build_waveform(self,
             if constant_values:
                 return ConstantWaveform.from_mapping(duration, constant_values)
         return None
+
+    @property
+    def initial_values(self) -> Dict[ChannelID, ExpressionScalar]:
+        return {ch: ExpressionScalar(val) for ch, val in self._amplitude_dict.items()}
+
+    @property
+    def final_values(self) -> Dict[ChannelID, ExpressionScalar]:
+        return {ch: ExpressionScalar(val) for ch, val in self._amplitude_dict.items()}

qupulse/pulses/function_pulse_template.py

@@ -151,4 +151,14 @@ def _as_expression(self) -> Dict[ChannelID, ExpressionScalar]:
         expr = ExpressionScalar.make(self.__expression.underlying_expression.subs({'t': self._AS_EXPRESSION_TIME}))
         return {self.__channel: expr}
 
+    @property
+    def initial_values(self) -> Dict[ChannelID, ExpressionScalar]:
+        expr = ExpressionScalar.make(self.__expression.underlying_expression.subs('t', 0))
+        return {self.__channel: expr}
+
+    @property
+    def final_values(self) -> Dict[ChannelID, ExpressionScalar]:
+        expr = ExpressionScalar.make(self.__expression.underlying_expression.subs('t', self.__duration_expression.underlying_expression))
+        return {self.__channel: expr}
+
 

qupulse/pulses/loop_pulse_template.py

@@ -225,6 +225,24 @@ def integral(self) -> Dict[ChannelID, ExpressionScalar]:
 
         return body_integrals
 
+    @property
+    def initial_values(self) -> Dict[ChannelID, ExpressionScalar]:
+        values = self.body.initial_values
+        initial_idx = self._loop_range.start
+        for ch, value in values.items():
+            values[ch] = ExpressionScalar(value.underlying_expression.subs(self._loop_index, initial_idx))
+        return values
+
+    @property
+    def final_values(self) -> Dict[ChannelID, ExpressionScalar]:
+        values = self.body.final_values
+        start, step, stop = self._loop_range.start.sympified_expression, self._loop_range.step.sympified_expression, self._loop_range.stop.sympified_expression
+        n = (stop - start) // step
+        final_idx = start + sympy.Max(n - 1, 0) * step
+        for ch, value in values.items():
+            values[ch] = ExpressionScalar(value.underlying_expression.subs(self._loop_index, final_idx))
+        return values
+
 
 class LoopIndexNotUsedException(Exception):
     def __init__(self, loop_index: str, body_parameter_names: Set[str]):

qupulse/pulses/mapping_pulse_template.py

@@ -344,38 +344,31 @@ def get_measurement_windows(self,
             measurement_mapping=self.get_updated_measurement_mapping(measurement_mapping=measurement_mapping)
         )
 
-    @property
-    def integral(self) -> Dict[ChannelID, ExpressionScalar]:
-        internal_integral = self.__template.integral
-        expressions = dict()
-
-        # sympy.subs() does not work if one of the mappings in the provided dict is an Expression object
-        # the following is an ugly workaround
-        # todo: make Expressions compatible with sympy.subs()
-        parameter_mapping = {parameter_name: expression.underlying_expression
-                             for parameter_name, expression in self.__parameter_mapping.items()}
-        for channel, ch_integral in internal_integral.items():
-            channel_out = self.__channel_mapping.get(channel, channel)
-            if channel_out is None:
-                continue
-
-            expressions[channel_out] = ExpressionScalar(
-                ch_integral.sympified_expression.subs(parameter_mapping, simultaneous=True)
-            )
-
-        return expressions
-
-    def _as_expression(self) -> Dict[ChannelID, ExpressionScalar]:
+    def _apply_mapping_to_inner_channel_dict(self, to_map: Dict[ChannelID, ExpressionScalar]) -> Dict[ChannelID, ExpressionScalar]:
         parameter_mapping = {parameter_name: expression.underlying_expression
                              for parameter_name, expression in self.__parameter_mapping.items()}
-        inner = self.__template._as_expression()
         return {
             self.__channel_mapping.get(ch, ch): ExpressionScalar(ch_expr.sympified_expression.subs(parameter_mapping,
                                                                                                    simultaneous=True))
-            for ch, ch_expr in inner.items()
+            for ch, ch_expr in to_map.items()
             if self.__channel_mapping.get(ch, ch) is not None
         }
 
+    @property
+    def integral(self) -> Dict[ChannelID, ExpressionScalar]:
+        return self._apply_mapping_to_inner_channel_dict(self.__template.integral)
+
+    def _as_expression(self) -> Dict[ChannelID, ExpressionScalar]:
+        return self._apply_mapping_to_inner_channel_dict(self.__template._as_expression())
+
+    @property
+    def initial_values(self) -> Dict[ChannelID, ExpressionScalar]:
+        return self._apply_mapping_to_inner_channel_dict(self.__template.initial_values)
+
+    @property
+    def final_values(self) -> Dict[ChannelID, ExpressionScalar]:
+        return self._apply_mapping_to_inner_channel_dict(self.__template.final_values)
+
 
 class MissingMappingException(Exception):
     """Indicates that no mapping was specified for some parameter declaration of a

qupulse/pulses/multi_channel_pulse_template.py

@@ -194,6 +194,20 @@ def _as_expression(self) -> Dict[ChannelID, ExpressionScalar]:
             expressions.update(subtemplate._as_expression())
         return expressions
 
+    @property
+    def initial_values(self) -> Dict[ChannelID, ExpressionScalar]:
+        values = {}
+        for subtemplate in self._subtemplates:
+            values.update(subtemplate.initial_values)
+        return values
+
+    @property
+    def final_values(self) -> Dict[ChannelID, ExpressionScalar]:
+        values = {}
+        for subtemplate in self._subtemplates:
+            values.update(subtemplate.final_values)
+        return values
+
 
 class ParallelConstantChannelPulseTemplate(PulseTemplate):
     def __init__(self,
@@ -280,6 +294,18 @@ def integral(self) -> Dict[ChannelID, ExpressionScalar]:
             integral[channel] = value * duration
         return integral
 
+    @property
+    def initial_values(self) -> Dict[ChannelID, ExpressionScalar]:
+        values = self._template.initial_values
+        values.update(self._overwritten_channels)
+        return values
+
+    @property
+    def final_values(self) -> Dict[ChannelID, ExpressionScalar]:
+        values = self._template.final_values
+        values.update(self._overwritten_channels)
+        return values
+
     def get_serialization_data(self, serializer: Optional[Serializer]=None) -> Dict[str, Any]:
         if serializer:
             raise NotImplementedError('Legacy serialization not implemented for new class')

qupulse/pulses/point_pulse_template.py

@@ -168,6 +168,22 @@ def value_trafo(v):
             expressions[channel] = pw
         return expressions
 
+    @property
+    def initial_values(self) -> Dict[ChannelID, ExpressionScalar]:
+        shape = (len(self._channels),)
+        return {
+            ch: ExpressionScalar(IndexedBroadcast(self._entries[0].v, shape, ch_idx))
+            for ch_idx, ch in enumerate(self._channels)
+        }
+
+    @property
+    def final_values(self) -> Dict[ChannelID, ExpressionScalar]:
+        shape = (len(self._channels),)
+        return {
+            ch: ExpressionScalar(IndexedBroadcast(self._entries[-1].v, shape, ch_idx))
+            for ch_idx, ch in enumerate(self._channels)
+        }
+
 
 class InvalidPointDimension(Exception):
     def __init__(self, expected, received):