QPY schedule serialization (#7300)

* Schedule serialization wip

* pulse serialization

* move mapping io to separate file

* move dump numbers to parameter_values

* add alignment context serialization

* block serialization

* add interface

* bug fix

* black / lint

* documentation

* add serialization for kernel and discriminator

* migrate circuit qpy to qpy module

* keep import path

* reno

* lint

* fix import path

* update schedule qpy with cleanup. alignment classes and acquire instruction class are also updated, i.e. remove individual instance members.

* move type key enum to type_keys module

* add mapping writer (for qpy values) and cleanup sequence serializer to make sure element data is INSTRUCTION_PARAM struct.

* compress symbolic pulse expressions with zlib

* remove change to existing circuit qpy tests

* cleanup for type key

* add unittest and fix some bug
- small bug fix for wrong class usage
- exeption for None symbolic expression
- add explicit expand for symengine expression

* update documentation

* add seek(0) to data_from_binary helper function

* remove subclasses and hard coded is_sequential in every subclass

* documentation update and type fix

Co-authored-by: Matthew Treinish <[email protected]>

* move test to own module

* update typehint

* add backward compatibility test

* fix merge and cleanup

* add circuit calibrations serialization

* fix qpy_compat test

* remove compatibility test for parameterized program

* update reno for calibrations

* fix data size

Co-authored-by: Matthew Treinish <[email protected]>
Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>

qiskit/pulse/instructions/acquire.py

@@ -75,10 +75,10 @@ def __init__(
         if not (mem_slot or reg_slot):
             raise PulseError("Neither MemorySlots nor RegisterSlots were supplied.")
 
-        self._kernel = kernel
-        self._discriminator = discriminator
-
-        super().__init__(operands=(duration, channel, mem_slot, reg_slot), name=name)
+        super().__init__(
+            operands=(duration, channel, mem_slot, reg_slot, kernel, discriminator),
+            name=name,
+        )
 
     @property
     def channel(self) -> AcquireChannel:
@@ -100,12 +100,12 @@ def duration(self) -> Union[int, ParameterExpression]:
     @property
     def kernel(self) -> Kernel:
         """Return kernel settings."""
-        return self._kernel
+        return self._operands[4]
 
     @property
     def discriminator(self) -> Discriminator:
         """Return discrimination settings."""
-        return self._discriminator
+        return self._operands[5]
 
     @property
     def acquire(self) -> AcquireChannel:

qiskit/pulse/library/symbolic_pulses.py

@@ -69,8 +69,14 @@ def _lifted_gaussian(
     Returns:
         Symbolic equation.
     """
-    gauss = sym.exp(-(((t - center) / sigma) ** 2) / 2)
-    offset = sym.exp(-(((t_zero - center) / sigma) ** 2) / 2)
+    # Sympy automatically does expand.
+    # This causes expression inconsistency after qpy round-trip serializing through sympy.
+    # See issue for details: https://github.com/symengine/symengine.py/issues/409
+    t_shifted = (t - center).expand()
+    t_offset = (t_zero - center).expand()
+
+    gauss = sym.exp(-((t_shifted / sigma) ** 2) / 2)
+    offset = sym.exp(-((t_offset / sigma) ** 2) / 2)
 
     return (gauss - offset) / (1 - offset)
 

qiskit/pulse/transforms/alignments.py

@@ -12,24 +12,22 @@
 """A collection of passes to reallocate the timeslots of instructions according to context."""
 
 import abc
-from typing import Callable, Dict, Any, Union
+from typing import Callable, Dict, Any, Union, Tuple
 
 import numpy as np
 
-from qiskit.circuit.parameterexpression import ParameterExpression
+from qiskit.circuit.parameterexpression import ParameterExpression, ParameterValueType
 from qiskit.pulse.exceptions import PulseError
 from qiskit.pulse.schedule import Schedule, ScheduleComponent
-from qiskit.pulse.utils import instruction_duration_validation
+from qiskit.pulse.utils import instruction_duration_validation, deprecated_functionality
 
 
 class AlignmentKind(abc.ABC):
     """An abstract class for schedule alignment."""
 
-    is_sequential = None
-
-    def __init__(self):
+    def __init__(self, context_params: Tuple[ParameterValueType, ...]):
         """Create new context."""
-        self._context_params = tuple()
+        self._context_params = tuple(context_params)
 
     @abc.abstractmethod
     def align(self, schedule: Schedule) -> Schedule:
@@ -46,20 +44,52 @@ def align(self, schedule: Schedule) -> Schedule:
         """
         pass
 
+    @deprecated_functionality
     def to_dict(self) -> Dict[str, Any]:
         """Returns dictionary to represent this alignment."""
         return {"alignment": self.__class__.__name__}
 
+    @property
+    @abc.abstractmethod
+    def is_sequential(self) -> bool:
+        """Return ``True`` if this is sequential alignment context.
+
+        This information is used to evaluate DAG equivalency of two :class:`.ScheduleBlock`s.
+        When the context has two pulses in different channels,
+        a sequential context subtype intends to return following scheduling outcome.
+
+        .. parsed-literal::
+
+                ┌────────┐
+            D0: ┤ pulse1 ├────────────
+                └────────┘  ┌────────┐
+            D1: ────────────┤ pulse2 ├
+                            └────────┘
+
+        On the other hand, parallel context with ``is_sequential=False`` returns
+
+        .. parsed-literal::
+
+                ┌────────┐
+            D0: ┤ pulse1 ├
+                ├────────┤
+            D1: ┤ pulse2 ├
+                └────────┘
+
+        All subclasses must implement this method according to scheduling strategy.
+        """
+        pass
+
     def __eq__(self, other):
         """Check equality of two transforms."""
-        return isinstance(other, type(self)) and self.to_dict() == other.to_dict()
+        if type(self) is not type(other):
+            return False
+        if self._context_params != other._context_params:
+            return False
+        return True
 
     def __repr__(self):
-        name = self.__class__.__name__
-        opts = self.to_dict()
-        opts.pop("alignment")
-        opts_str = ", ".join(f"{key}={val}" for key, val in opts.items())
-        return f"{name}({opts_str})"
+        return f"{self.__class__.__name__}({', '.join(self._context_params)})"
 
 
 class AlignLeft(AlignmentKind):
@@ -68,7 +98,13 @@ class AlignLeft(AlignmentKind):
     Instructions are placed at earliest available timeslots.
     """
 
-    is_sequential = False
+    def __init__(self):
+        """Create new left-justified context."""
+        super().__init__(context_params=())
+
+    @property
+    def is_sequential(self) -> bool:
+        return False
 
     def align(self, schedule: Schedule) -> Schedule:
         """Reallocate instructions according to the policy.
@@ -129,7 +165,13 @@ class AlignRight(AlignmentKind):
     Instructions are placed at latest available timeslots.
     """
 
-    is_sequential = False
+    def __init__(self):
+        """Create new right-justified context."""
+        super().__init__(context_params=())
+
+    @property
+    def is_sequential(self) -> bool:
+        return False
 
     def align(self, schedule: Schedule) -> Schedule:
         """Reallocate instructions according to the policy.
@@ -192,7 +234,13 @@ class AlignSequential(AlignmentKind):
     No buffer time is inserted in between instructions.
     """
 
-    is_sequential = True
+    def __init__(self):
+        """Create new sequential context."""
+        super().__init__(context_params=())
+
+    @property
+    def is_sequential(self) -> bool:
+        return True
 
     def align(self, schedule: Schedule) -> Schedule:
         """Reallocate instructions according to the policy.
@@ -220,8 +268,6 @@ class AlignEquispaced(AlignmentKind):
     This alignment is convenient to create dynamical decoupling sequences such as PDD.
     """
 
-    is_sequential = True
-
     def __init__(self, duration: Union[int, ParameterExpression]):
         """Create new equispaced context.
 
@@ -231,9 +277,11 @@ def __init__(self, duration: Union[int, ParameterExpression]):
                 no alignment is performed and the input schedule is just returned.
                 This duration can be parametrized.
         """
-        super().__init__()
+        super().__init__(context_params=(duration,))
 
-        self._context_params = (duration,)
+    @property
+    def is_sequential(self) -> bool:
+        return True
 
     @property
     def duration(self):
@@ -281,6 +329,7 @@ def align(self, schedule: Schedule) -> Schedule:
 
         return aligned
 
+    @deprecated_functionality
     def to_dict(self) -> Dict[str, Any]:
         """Returns dictionary to represent this alignment."""
         return {"alignment": self.__class__.__name__, "duration": self.duration}
@@ -301,9 +350,13 @@ class AlignFunc(AlignmentKind):
 
         def udd10_pos(j):
         return np.sin(np.pi*j/(2*10 + 2))**2
-    """
 
-    is_sequential = True
+    .. note::
+
+        This context cannot be QPY serialized because of the callable. If you use this context,
+        your program cannot be saved in QPY format.
+
+    """
 
     def __init__(self, duration: Union[int, ParameterExpression], func: Callable):
         """Create new equispaced context.
@@ -317,16 +370,22 @@ def __init__(self, duration: Union[int, ParameterExpression], func: Callable):
                 fractional coordinate of of that sub-schedule. The returned value should be
                 defined within [0, 1]. The pulse index starts from 1.
         """
-        super().__init__()
+        super().__init__(context_params=(duration, func))
 
-        self._context_params = (duration,)
-        self._func = func
+    @property
+    def is_sequential(self) -> bool:
+        return True
 
     @property
     def duration(self):
         """Return context duration."""
         return self._context_params[0]
 
+    @property
+    def func(self):
+        """Return context alignment function."""
+        return self._context_params[1]
+
     def align(self, schedule: Schedule) -> Schedule:
         """Reallocate instructions according to the policy.
 
@@ -346,21 +405,22 @@ def align(self, schedule: Schedule) -> Schedule:
 
         aligned = Schedule.initialize_from(schedule)
         for ind, (_, child) in enumerate(schedule.children):
-            _t_center = self.duration * self._func(ind + 1)
+            _t_center = self.duration * self.func(ind + 1)
             _t0 = int(_t_center - 0.5 * child.duration)
             if _t0 < 0 or _t0 > self.duration:
                 PulseError("Invalid schedule position t=%d is specified at index=%d" % (_t0, ind))
             aligned.insert(_t0, child, inplace=True)
 
         return aligned
 
+    @deprecated_functionality
     def to_dict(self) -> Dict[str, Any]:
         """Returns dictionary to represent this alignment.
 
         .. note:: ``func`` is not presented in this dictionary. Just name.
         """
         return {
             "alignment": self.__class__.__name__,
-            "duration": self._context_params[0],
-            "func": self._func.__name__,
+            "duration": self.duration,
+            "func": self.func.__name__,
         }

qiskit/pulse/transforms/dag.py

@@ -58,6 +58,9 @@ def block_to_dag(block: ScheduleBlock) -> rx.PyDAG:
 
     Returns:
         Instructions in DAG representation.
+
+    Raises:
+        PulseError: When the context is invalid subclass.
     """
     if block.alignment_context.is_sequential:
         return _sequential_allocation(block)