Merge 62d18ad into e74b530

qctoolkit/pulses/function_pulse_template.py

@@ -14,7 +14,7 @@
 import sympy
 
 from qctoolkit.expressions import ExpressionScalar
-from qctoolkit.serialization import Serializer
+from qctoolkit.serialization import Serializer, Serializable
 
 from qctoolkit.utils.types import ChannelID, TimeType, time_from_float
 from qctoolkit.pulses.parameters import Parameter, ParameterConstrainer, ParameterConstraint
@@ -44,7 +44,8 @@ def __init__(self,
                  identifier: Optional[str] = None,
                  *,
                  measurements: Optional[List[MeasurementDeclaration]]=None,
-                 parameter_constraints: Optional[List[Union[str, ParameterConstraint]]]=None) -> None:
+                 parameter_constraints: Optional[List[Union[str, ParameterConstraint]]]=None,
+                 registry: Dict[str, Serializable]=None) -> None:
         """
         Args:
             expression: The function represented by this FunctionPulseTemplate
@@ -60,7 +61,7 @@ def __init__(self,
             parameter_constraints: A list of parameter constraints forwarded to the
                 :class:`~`qctoolkit.pulses.measurement.ParameterConstrainer superclass
         """
-        AtomicPulseTemplate.__init__(self, identifier=identifier, measurements=measurements)
+        AtomicPulseTemplate.__init__(self, identifier=identifier, measurements=measurements, registry=registry)
         ParameterConstrainer.__init__(self, parameter_constraints=parameter_constraints)
 
         self.__expression = ExpressionScalar.make(expression)
@@ -119,31 +120,37 @@ def requires_stop(self,
             for name in parameters.keys() if (name in self.parameter_names)
         )
 
-    def get_serialization_data(self, serializer: Serializer) -> Dict[str, Any]:
-        return dict(
+    def get_serialization_data(self, serializer: Optional[Serializer]=None) -> Dict[str, Any]:
+        data = super().get_serialization_data(serializer)
+
+        if serializer: # compatibility to old serialization routines, deprecated
+            return dict(
+                duration_expression=self.__duration_expression,
+                expression=self.__expression,
+                channel=self.__channel,
+                measurement_declarations=self.measurement_declarations,
+                parameter_constraints=[str(c) for c in self.parameter_constraints]
+            )
+
+        local_data = dict(
             duration_expression=self.__duration_expression,
             expression=self.__expression,
             channel=self.__channel,
-            measurement_declarations=self.measurement_declarations,
+            measurements=self.measurement_declarations,
             parameter_constraints=[str(c) for c in self.parameter_constraints]
         )
 
-    @staticmethod
-    def deserialize(serializer: Serializer,
-                    expression: Any,
-                    duration_expression: Any,
-                    channel: 'ChannelID',
-                    measurement_declarations: List[MeasurementDeclaration],
-                    parameter_constraints: List,
-                    identifier: Optional[bool]=None) -> 'FunctionPulseTemplate':
-        return FunctionPulseTemplate(
-            expression,
-            duration_expression,
-            channel=channel,
-            identifier=identifier,
-            measurements=measurement_declarations,
-            parameter_constraints=parameter_constraints
-        )
+        data.update(**local_data)
+        return data
+
+    @classmethod
+    def deserialize(cls,
+                    serializer: Optional[Serializer]=None,
+                    **kwargs) -> 'FunctionPulseTemplate':
+        if serializer:
+            kwargs['measurements'] = kwargs['measurement_declarations'] # compatibility to old serialization routines, deprecated
+            del kwargs['measurement_declarations']
+        return super().deserialize(None, **kwargs)
 
     @property
     def integral(self) -> Dict[ChannelID, ExpressionScalar]:

qctoolkit/pulses/loop_pulse_template.py

@@ -24,8 +24,10 @@
 
 class LoopPulseTemplate(PulseTemplate):
     """Base class for loop based pulse templates. This class is still abstract and cannot be instantiated."""
-    def __init__(self, body: PulseTemplate, identifier: Optional[str]=None):
-        super().__init__(identifier=identifier)
+    def __init__(self, body: PulseTemplate,
+                 identifier: Optional[str],
+                 registry: Optional[dict]):
+        super().__init__(identifier=identifier, registry=registry)
         self.__body = body
 
     @property
@@ -109,15 +111,16 @@ def __init__(self,
                  identifier: Optional[str]=None,
                  *,
                  measurements: Optional[Sequence[MeasurementDeclaration]]=None,
-                 parameter_constraints: Optional[Sequence]=None):
+                 parameter_constraints: Optional[Sequence]=None,
+                 registry: Optional[dict]=None):
         """
         Args:
             body: The loop body. It is expected to have `loop_index` as an parameter
             loop_index: Loop index of the for loop
             loop_range: Range to loop through
             identifier: Used for serialization
         """
-        LoopPulseTemplate.__init__(self, body=body, identifier=identifier)
+        LoopPulseTemplate.__init__(self, body=body, identifier=identifier, registry=registry)
         MeasurementDefiner.__init__(self, measurements=measurements)
         ParameterConstrainer.__init__(self, parameter_constraints=parameter_constraints)
 
@@ -232,34 +235,30 @@ def requires_stop(self,
                       conditions: Dict[str, 'Condition']) -> bool:
         return any(parameters[parameter_name].requires_stop for parameter_name in self._loop_range.parameter_names)
 
-    def get_serialization_data(self, serializer: Serializer) -> Dict[str, Any]:
-        data = dict(
-            body=serializer.dictify(self.body),
-            loop_range=self._loop_range.to_tuple(),
-            loop_index=self._loop_index,
-        )
+    def get_serialization_data(self, serializer: Optional[Serializer]=None) -> Dict[str, Any]:
+        data = super().get_serialization_data(serializer)
+
+        data['body'] = self.body
+
+        if serializer: # compatibility to old serialization routines, deprecated
+            data = dict()
+            data['body'] = serializer.dictify(self.body)
+
+        data['loop_range'] = self._loop_range.to_tuple()
+        data['loop_index'] = self._loop_index
+
         if self.parameter_constraints:
             data['parameter_constraints'] = [str(c) for c in self.parameter_constraints]
         if self.measurement_declarations:
             data['measurements'] = self.measurement_declarations
+
         return data
 
-    @staticmethod
-    def deserialize(serializer: Serializer,
-                    body: Dict[str, Any],
-                    loop_range: Tuple,
-                    loop_index: str,
-                    identifier: Optional[str]=None,
-                    measurements: Optional[Sequence[str]]=None,
-                    parameter_constraints: Optional[Sequence[str]]=None) -> 'ForLoopPulseTemplate':
-        body = cast(PulseTemplate, serializer.deserialize(body))
-        return ForLoopPulseTemplate(body=body,
-                                    identifier=identifier,
-                                    loop_range=loop_range,
-                                    loop_index=loop_index,
-                                    measurements=measurements,
-                                    parameter_constraints=parameter_constraints
-                                    )
+    @classmethod
+    def deserialize(cls, serializer: Optional[Serializer]=None, **kwargs) -> 'ForLoopTemplate':
+        if serializer: # compatibility to old serialization routines, deprecated
+            kwargs['body'] = cast(PulseTemplate, serializer.deserialize(kwargs['body']))
+        return super().deserialize(None, **kwargs)
 
     @property
     def integral(self) -> Dict[ChannelID, ExpressionScalar]:
@@ -288,15 +287,17 @@ def integral(self) -> Dict[ChannelID, ExpressionScalar]:
         return body_integrals
 
 
-
 class WhileLoopPulseTemplate(LoopPulseTemplate):
     """Conditional looping in a pulse.
     
     A LoopPulseTemplate is a PulseTemplate whose body is repeated
     during execution as long as a certain condition holds.
     """
 
-    def __init__(self, condition: str, body: PulseTemplate, identifier: Optional[str]=None) -> None:
+    def __init__(self, condition: str,
+                 body: PulseTemplate,
+                 identifier: Optional[str]=None,
+                 registry: Optional[dict]=None) -> None:
         """Create a new LoopPulseTemplate instance.
 
         Args:
@@ -306,7 +307,7 @@ def __init__(self, condition: str, body: PulseTemplate, identifier: Optional[str
                 holds.
             identifier (str): A unique identifier for use in serialization. (optional)
         """
-        super().__init__(body=body, identifier=identifier)
+        super().__init__(body=body, identifier=identifier, registry=registry)
         self._condition = condition
 
     def __str__(self) -> str:
@@ -352,24 +353,24 @@ def requires_stop(self,
                       conditions: Dict[str, Condition]) -> bool:
         return self.__obtain_condition_object(conditions).requires_stop()
 
-    def get_serialization_data(self, serializer: Serializer) -> Dict[str, Any]:
-        data = dict(
-            type=serializer.get_type_identifier(self),
-            condition=self._condition,
-            body=serializer.dictify(self.body)
-        )
+    def get_serialization_data(self, serializer: Optional[Serializer]=None) -> Dict[str, Any]:
+        data = super().get_serialization_data(serializer)
+        data['body'] = self.body
+
+        if serializer: # compatibility to old serialization routines, deprecated
+            data = dict()
+            data['body'] = serializer.dictify(self.body)
+
+        data['condition'] = self._condition
+
         return data
 
-    @staticmethod
-    def deserialize(serializer: Serializer,
-                    condition: str,
-                    body: Dict[str, Any],
-                    identifier: Optional[str]=None) -> 'WhileLoopPulseTemplate':
-        body = serializer.deserialize(body)
-        result = WhileLoopPulseTemplate(condition=condition,
-                                        body=body,
-                                        identifier=identifier)
-        return result
+    @classmethod
+    def deserialize(cls, serializer: Optional[Serializer]=None, **kwargs) -> 'WhileLoopPulseTemplate':
+        if serializer: # compatibility to old serialization routines, deprecated
+            kwargs['body'] = serializer.deserialize(kwargs['body'])
+
+        return super().deserialize(**kwargs)
 
     @property
     def integral(self) -> Dict[ChannelID, ExpressionScalar]:

qctoolkit/pulses/multi_channel_pulse_template.py

@@ -139,8 +139,9 @@ def __init__(self,
                  external_parameters: Optional[Set[str]]=None,
                  identifier: Optional[str]=None,
                  parameter_constraints: Optional[List]=None,
-                 measurements: Optional[List[MeasurementDeclaration]]=None) -> None:
-        AtomicPulseTemplate.__init__(self, identifier=identifier, measurements=measurements)
+                 measurements: Optional[List[MeasurementDeclaration]]=None,
+                 registry: Optional[dict] = None) -> None:
+        AtomicPulseTemplate.__init__(self, identifier=identifier, measurements=measurements, registry=registry)
         ParameterConstrainer.__init__(self, parameter_constraints=parameter_constraints)
 
         self._subtemplates = [st if isinstance(st, PulseTemplate) else MappingPulseTemplate.from_tuple(st) for st in
@@ -225,26 +226,30 @@ def requires_stop(self,
                       conditions: Dict[str, 'Condition']) -> bool:
         return any(st.requires_stop(parameters, conditions) for st in self._subtemplates)
 
-    def get_serialization_data(self, serializer: Serializer) -> Dict[str, Any]:
-        data = dict(subtemplates=[serializer.dictify(subtemplate) for subtemplate in self.subtemplates])
+    def get_serialization_data(self, serializer: Optional[Serializer]=None) -> Dict[str, Any]:
+        data = super().get_serialization_data(serializer)
+        data['subtemplates'] = self.subtemplates
+
+        if serializer: # compatibility to old serialization routines, deprecated
+            data = dict()
+            data['subtemplates'] = [serializer.dictify(subtemplate) for subtemplate in self.subtemplates]
 
         if self.parameter_constraints:
             data['parameter_constraints'] = [str(constraint) for constraint in self.parameter_constraints]
-
         if self.measurement_declarations:
             data['measurements'] = self.measurement_declarations
+
         return data
 
-    @staticmethod
-    def deserialize(serializer: Serializer,
-                    subtemplates: Iterable[Dict[str, Any]],
-                    parameter_constraints: Optional[Any]=None,
-                    identifier: Optional[str]=None,
-                    measurements: Optional[List[MeasurementDeclaration]]=None) -> 'AtomicMultiChannelPulseTemplate':
-        subtemplates = [serializer.deserialize(st) for st in subtemplates]
-        return AtomicMultiChannelPulseTemplate(*subtemplates,
-                                               parameter_constraints=parameter_constraints,
-                                               identifier=identifier, measurements=measurements)
+    @classmethod
+    def deserialize(cls, serializer: Optional[Serializer]=None, **kwargs) -> 'AtomicMultiChannelPulseTemplate':
+        subtemplates = kwargs['subtemplates']
+        del kwargs['subtemplates']
+
+        if serializer: # compatibility to old serialization routines, deprecated
+            subtemplates = [serializer.deserialize(st) for st in subtemplates]
+
+        return cls(*subtemplates, **kwargs)
 
     @property
     def integral(self) -> Dict[ChannelID, ExpressionScalar]:

qctoolkit/pulses/point_pulse_template.py

@@ -1,4 +1,4 @@
-from typing import Optional, List, Union, Set, Dict, Sequence
+from typing import Optional, List, Union, Set, Dict, Sequence, Any
 from numbers import Real
 import itertools
 import numbers
@@ -14,6 +14,7 @@
 from qctoolkit.pulses.pulse_template import AtomicPulseTemplate, MeasurementDeclaration
 from qctoolkit.pulses.table_pulse_template import TableEntry, EntryInInit, TableWaveform, TableWaveformEntry
 from qctoolkit.pulses.multi_channel_pulse_template import MultiChannelWaveform
+from qctoolkit.serialization import Serializer
 
 
 __all__ = ["PointWaveform", "PointPulseTemplate", "PointPulseEntry", "PointWaveformEntry", "InvalidPointDimension"]
@@ -44,9 +45,10 @@ def __init__(self,
                  *,
                  parameter_constraints: Optional[List[Union[str, ParameterConstraint]]]=None,
                  measurements: Optional[List[MeasurementDeclaration]]=None,
-                 identifier=None):
+                 identifier: Optional[str]=None,
+                 registry: Optional[dict]=None):
 
-        AtomicPulseTemplate.__init__(self, identifier=identifier, measurements=measurements)
+        AtomicPulseTemplate.__init__(self, identifier=identifier, measurements=measurements, registry=registry)
         ParameterConstrainer.__init__(self, parameter_constraints=parameter_constraints)
 
         self._channels = tuple(channel_names)
@@ -99,20 +101,20 @@ def build_waveform(self,
     def point_pulse_entries(self) -> Sequence[PointPulseEntry]:
         return self._entries
 
-    def get_serialization_data(self, serializer) -> Dict:
-        data = {'time_point_tuple_list': [entry.get_serialization_data()
-                                          for entry in self._entries],
-                'channel_names':       self._channels}
+    def get_serialization_data(self, serializer: Optional[Serializer]=None) -> Dict[str, Any]:
+        data = super().get_serialization_data(serializer)
+
+        if serializer: # compatibility to old serialization routines, deprecated
+            data = dict()
+
+        data['time_point_tuple_list'] = [entry.get_serialization_data() for entry in self._entries]
+        data['channel_names'] = self._channels
         if self.parameter_constraints:
             data['parameter_constraints'] = [str(c) for c in self.parameter_constraints]
         if self.measurement_declarations:
             data['measurements'] = self.measurement_declarations
         return data
 
-    @staticmethod
-    def deserialize(serializer, **kwargs) -> 'PointPulseTemplate':
-        return PointPulseTemplate(**kwargs)
-
     @property
     def duration(self) -> Expression:
         return self._entries[-1].t

qctoolkit/pulses/pulse_template.py

@@ -42,8 +42,10 @@ class PulseTemplate(Serializable, SequencingElement, metaclass=DocStringABCMeta)
     """
 
     def __init__(self, *,
-                 identifier: Optional[str]) -> None:
-        super().__init__(identifier)
+                 identifier: Optional[str],
+                 registry: Optional[Dict[str, Serializable]]) -> None:
+        super().__init__(identifier=identifier,
+                         registry=registry)
 
     @property
     @abstractmethod
@@ -107,8 +109,9 @@ class AtomicPulseTemplate(PulseTemplate, MeasurementDefiner):
     """
     def __init__(self, *,
                  identifier: Optional[str],
-                 measurements: Optional[List[MeasurementDeclaration]]):
-        PulseTemplate.__init__(self, identifier=identifier)
+                 measurements: Optional[List[MeasurementDeclaration]],
+                 registry: Optional[Dict[str, Serializable]]):
+        PulseTemplate.__init__(self, identifier=identifier, registry=registry)
         MeasurementDefiner.__init__(self, measurements=measurements)
 
     def is_interruptable(self) -> bool: