Serializable parametric pulse

qiskit/pulse/library/symbolic_pulses.py

@@ -19,7 +19,7 @@
 """
 
 import functools
-from typing import Any, Dict, Optional, Union, Callable
+from typing import Any, Dict, List, Optional, Union, Callable
 
 import numpy as np
 
@@ -90,7 +90,30 @@ def _validate_amplitude_limit(symbolic_pulse: "SymbolicPulse") -> bool:
     return np.any(np.abs(symbolic_pulse.get_waveform().samples) > 1.0)
 
 
-class LamdifiedExpression:
+def _get_expression_args(
+    expr: sym.Expr,
+    params: Dict[str, float],
+    exclude: Optional[List[str]] = None,
+) -> List[float]:
+    """A helper function to get argument to evaluate expression.
+
+    Args:
+        expr: Symbolic expression to evaluate.
+        params: Dictionary of parameter, which is a superset of expression arguments.
+        exclude: Parameter to exclude from arguments.
+
+    Returns:
+        Arguments passed to the lambdified expression.
+    """
+    args = []
+    for symbol in sorted(expr.free_symbols, key=lambda s: s.name):
+        if exclude and symbol.name in exclude:
+            continue
+        args.append(params[symbol.name])
+    return args
+
+
+class LambdifiedExpression:
     """Descriptor to lambdify symbolic expression with cache.
 
     When new symbolic expression is set for the first time,
@@ -302,15 +325,16 @@ def Sawtooth(duration, amp, freq, name):
     __slots__ = (
         "_amp",
         "_pulse_type",
-        "_param_names",
-        "_param_vals",
+        "_params",
         "envelope",
         "constraints",
+        "eval_conditions",
     )
 
     # Lambdify caches keyed on sympy expressions. Returns the corresponding callable.
-    _envelope_lambdify = LamdifiedExpression("envelope")
-    _constraints_lambdify = LamdifiedExpression("constraints")
+    _envelope_lam = LambdifiedExpression("envelope")
+    _constraints_lam = LambdifiedExpression("constraints")
+    _eval_conditions_lam = LambdifiedExpression("eval_conditions")
 
     def __init__(
         self,
@@ -321,6 +345,7 @@ def __init__(
         limit_amplitude: Optional[bool] = None,
         envelope: Optional[sym.Expr] = None,
         constraints: Optional[sym.Expr] = None,
+        eval_conditions: Optional[sym.Expr] = None,
     ):
         """Create a parametric pulse.
 
@@ -333,6 +358,9 @@ def __init__(
                 waveform to 1. The default is ``True`` and the amplitude is constrained to 1.
             envelope: Pulse envelope expression.
             constraints: Pulse parameter constraint expression.
+            eval_conditions: Extra conditions to evaluate full waveform to check the
+                amplitude limit. This can be provided if ``envelope`` function is not
+                normalized to 1.0.
 
         Raises:
             PulseError: When not all parameters are listed in the attribute :attr:`PARAM_DEF`.
@@ -356,21 +384,21 @@ def __init__(
             )
         self._amp = parameters.pop("amp")
         self._pulse_type = pulse_type
-        self._param_names = tuple(parameters.keys())
-        self._param_vals = tuple(parameters.values())
+        self._params = parameters
 
         self.envelope = envelope
         self.constraints = constraints
+        self.eval_conditions = eval_conditions
 
     def __getattr__(self, item):
         # Get pulse parameters with attribute-like access.
-        param_names = object.__getattribute__(self, "_param_names")
-        param_vals = object.__getattribute__(self, "_param_vals")
         if item == "amp":
             return object.__getattribute__(self, "_amp")
-        if item not in param_names:
+
+        params = object.__getattribute__(self, "_params")
+        if item not in params:
             raise AttributeError(f"'{self.__class__.__name__}' object has no attribute '{item}'")
-        return param_vals[param_names.index(item)]
+        return params[item]
 
     @property
     def pulse_type(self) -> str:
@@ -402,15 +430,9 @@ def get_waveform(self) -> Waveform:
             raise PulseError("Pulse envelope expression is not assigned.")
 
         times = np.arange(0, self.duration) + 1 / 2
-        params = self.parameters
+        fargs = [times, *_get_expression_args(self.envelope, self.parameters, exclude=["t"])]
 
-        func_args = [times]
-        for name in sorted(map(lambda s: s.name, self.envelope.free_symbols)):
-            if name == "t":
-                continue
-            func_args.append(params[name])
-
-        return Waveform(samples=self._amp * self._envelope_lambdify(*func_args), name=self.name)
+        return Waveform(samples=self._amp * self._envelope_lam(*fargs), name=self.name)
 
     def validate_parameters(self) -> None:
         """Validate parameters.
@@ -421,46 +443,56 @@ def validate_parameters(self) -> None:
         if self.is_parameterized():
             return
 
-        if any(p.imag != 0 for p in self._param_vals):
+        if any(p.imag != 0 for p in self._params.values()):
             raise PulseError("Pulse parameters must be real numbers except for 'amp'.")
 
         if self.constraints is not None:
-            func_args = []
-            params = self.parameters
-            for name in sorted(map(lambda s: s.name, self.constraints.free_symbols)):
-                func_args.append(params[name])
-
-            if not bool(self._constraints_lambdify(*func_args)):
+            fargs = _get_expression_args(self.constraints, self.parameters)
+            if not bool(self._constraints_lam(*fargs)):
                 param_repr = ", ".join(f"{p}={v}" for p, v in self.parameters.items())
                 const_repr = str(self.constraints)
                 raise PulseError(
                     f"Assigned parameters {param_repr} violate following constraint: {const_repr}."
                 )
 
-        if self.limit_amplitude and (np.abs(self._amp) > 1.0 or _validate_amplitude_limit(self)):
-            # Check max amplitude limit by generating waveform.
-            # We can avoid calling _validate_amplitude_limit when |amp| > 1.0
-            # which obviously violates the amplitude constraint by definition.
-            param_repr = ", ".join(f"{p}={v}" for p, v in self.parameters.items())
-            raise PulseError(
-                f"Maximum pulse amplitude norm exceeds 1.0 with assigned parameters {param_repr}."
-                "This can be overruled by setting Pulse.limit_amplitude."
-            )
+        if self.limit_amplitude:
+            violate_amp_limit = False
+
+            if np.abs(self._amp) > 1.0:
+                # Strong condition. |amp| > 1.0 must exceed AWG dynamic range.
+                violate_amp_limit = True
+            elif self.eval_conditions is not None:
+                # Check actual waveform.
+                # For example, if .envelope function is not normalized,
+                # |amp| < 1.0 can result in the maximum amplitude > 1.0, depending on the
+                # selection of other parameters.
+                fargs = _get_expression_args(self.eval_conditions, self.parameters)
+                if bool(self._eval_conditions_lam(*fargs)) and _validate_amplitude_limit(self):
+                    violate_amp_limit = True
+
+            if violate_amp_limit:
+                # Check max amplitude limit by generating waveform.
+                # We can avoid calling _validate_amplitude_limit when |amp| > 1.0
+                # which obviously violates the amplitude constraint by definition.
+                param_repr = ", ".join(f"{p}={v}" for p, v in self.parameters.items())
+                raise PulseError(
+                    f"Maximum pulse amplitude norm exceeds 1.0 with assigned parameters {param_repr}."
+                    "This can be overruled by setting Pulse.limit_amplitude."
+                )
 
     def is_parameterized(self) -> bool:
         """Return True iff the instruction is parameterized."""
-        args = (self.duration, self._amp, *self._param_vals)
-        return any(isinstance(val, ParameterExpression) for val in args)
+        return any(isinstance(val, ParameterExpression) for val in self.parameters.values())
 
     @property
     def parameters(self) -> Dict[str, Any]:
         params = {"duration": self.duration, "amp": self._amp}
-        params.update(dict(zip(self._param_names, self._param_vals)))
+        params.update(self._params)
         return params
 
     def __eq__(self, other: "SymbolicPulse") -> bool:
 
-        if self.envelope != other.envelope:
+        if self.envelope != getattr(other, "envelope", None):
             return False
 
         if self.parameters != other.parameters:
@@ -469,9 +501,7 @@ def __eq__(self, other: "SymbolicPulse") -> bool:
         return True
 
     def __hash__(self) -> int:
-        return hash(
-            (self._pulse_type, self.duration, self._amp, *self._param_names, *self._param_vals)
-        )
+        return hash((self._pulse_type, self.duration, self._amp, *tuple(self._params.items())))
 
     def __repr__(self) -> str:
         param_repr = ", ".join(f"{p}={v}" for p, v in self.parameters.items())
@@ -729,6 +759,7 @@ def __init__(
         # In IBM quantum backend, im(beta) == 0 is explicitly checked.
         # In Qiskit, we impose real number constraints on all parameters except for 'amp'.
         consts_expr = _sigma > 0
+        eval_conditions_expr = sym.Abs(_beta) > _sigma
 
         super().__init__(
             pulse_type=self.__class__.__name__,
@@ -738,6 +769,7 @@ def __init__(
             limit_amplitude=limit_amplitude,
             envelope=envelope_expr,
             constraints=consts_expr,
+            eval_conditions=eval_conditions_expr,
         )
         self.validate_parameters()
 

qiskit/pulse/parameter_manager.py

@@ -223,23 +223,20 @@ def visit_ParametricPulse(self, node: ParametricPulse):
     def visit_SymbolicPulse(self, node: SymbolicPulse):
         """Assign parameters to ``SymbolicPulse`` object."""
         if node.is_parameterized():
-            new_values = []
             # Assign duration
             if isinstance(node.duration, ParameterExpression):
                 node.duration = self._assign_parameter_expression(node.duration)
-            if isinstance(node.amp, ParameterExpression):
-                assigned_amp = self._assign_parameter_expression(node.amp)
+            if isinstance(node._amp, ParameterExpression):
+                assigned_amp = self._assign_parameter_expression(node._amp)
                 if not isinstance(assigned_amp, ParameterExpression):
                     # Amplitude is complex value
                     assigned_amp = complex(assigned_amp)
-                node.amp = assigned_amp
+                node._amp = assigned_amp
             # Assign other parameters
-            for op_value in node._param_vals:
-                if isinstance(op_value, ParameterExpression):
-                    new_values.append(self._assign_parameter_expression(op_value))
-                else:
-                    new_values.append(op_value)
-            node._param_vals = new_values
+            for name in node._params:
+                pval = node._params[name]
+                if isinstance(pval, ParameterExpression):
+                    node._params[name] = self._assign_parameter_expression(pval)
             node.validate_parameters()
 
         return node

test/python/pulse/test_pulse_lib.py

@@ -172,6 +172,8 @@ def test_drag_validation(self):
         wf = Drag(duration=duration, sigma=sigma, amp=amp, beta=beta)
         samples = wf.get_waveform().samples
         self.assertTrue(max(np.abs(samples)) <= 1)
+        with self.assertRaises(PulseError):
+            wf = Drag(duration=duration, sigma=sigma, amp=1.2, beta=beta)
         beta = sigma**2
         with self.assertRaises(PulseError):
             wf = Drag(duration=duration, sigma=sigma, amp=amp, beta=beta)
@@ -319,15 +321,13 @@ def test_envelope_cache(self):
         """Test speed up of instantiation with lambdify envelope cache."""
         drag_instance1 = Drag(duration=100, amp=0.1, sigma=40, beta=3)
         drag_instance2 = Drag(duration=100, amp=0.1, sigma=40, beta=3)
-        self.assertTrue(drag_instance1._envelope_lambdify is drag_instance2._envelope_lambdify)
+        self.assertTrue(drag_instance1._envelope_lam is drag_instance2._envelope_lam)
 
     def test_constraints_cache(self):
         """Test speed up of instantiation with lambdify constraints cache."""
         drag_instance1 = Drag(duration=100, amp=0.1, sigma=40, beta=3)
         drag_instance2 = Drag(duration=100, amp=0.1, sigma=40, beta=3)
-        self.assertTrue(
-            drag_instance1._constraints_lambdify is drag_instance2._constraints_lambdify
-        )
+        self.assertTrue(drag_instance1._constraints_lam is drag_instance2._constraints_lam)
 
     def test_deepcopy(self):
         """Test deep copying instance."""