Abstract pulse template.

ReleaseNotes.txt

@@ -11,6 +11,7 @@
         - Make duration equality check approximate (numeric tolerance)
     - Plotting:
         - Add `time_slice` keyword argument to render() and plot()
+    - Add `AbstractPulseTemplate` class
     - `PointPulseTemplate`:
         - Fixed bug in integral evaluation
 

doc/source/examples/12AbstractPulseTemplate.ipynb

@@ -0,0 +1,130 @@
+{
+ "cells": [
+  {
+   "cell_type": "markdown",
+   "metadata": {
+    "collapsed": true
+   },
+   "source": [
+    "# Abstract Pulse Template\n",
+    "This pulse template can be used as a place holder for a pulse template with a defined interface. Pulse template properties like `defined_channels` can be passed on initialization to declare those properties who make up the interface. Omitted properties raise an `NotSpecifiedError` exception if accessed. Properties which have been accessed are marked as \"frozen\".\n",
+    "The abstract pulse template can be linked to another pulse template by calling the `link_to` member. The target has to have the same properties for all properties marked as \"frozen\". This ensures a property always returns the same value."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 1,
+   "metadata": {
+    "scrolled": true
+   },
+   "outputs": [],
+   "source": [
+    "from qupulse.pulses import AbstractPT, FunctionPT, AtomicMultiChannelPT, PointPT\n",
+    "\n",
+    "init = PointPT([(0, (1, 0)), ('t_init', (0, 1), 'linear')], ['X', 'Y'])\n",
+    "abstract_readout = AbstractPT('readout', defined_channels={'X', 'Y'}, integral={'X': 1, 'Y': 'a*b'})\n",
+    "manip = AtomicMultiChannelPT(FunctionPT('sin(t)', 't_manip', channel='X'),\n",
+    "                             FunctionPT('cos(t)', 't_manip', channel='Y'))\n",
+    "\n",
+    "experiment = init @ manip @ abstract_readout"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "We can access declared properties like integral. If we try to get a non-declared property an exception is raised."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 2,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "The integral has been declared so we can get it\n",
+      "{'Y': Expression('a*b + sin(t_manip)'), 'X': Expression('t_init - cos(t_manip) + 2')}\n",
+      "\n",
+      "We get an error that for the pulse \"readout\" the property \"duration\" was not specified:\n",
+      "NotSpecifiedError('readout', 'duration')\n"
+     ]
+    }
+   ],
+   "source": [
+    "print('The integral has been declared so we can get it')\n",
+    "print(experiment.integral)\n",
+    "print()\n",
+    "\n",
+    "import traceback\n",
+    "try:\n",
+    "    experiment.duration\n",
+    "except Exception as err:\n",
+    "    print('We get an error that for the pulse \"readout\" the property \"duration\" was not specified:')\n",
+    "    print(repr(err))"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "We can link the abstract pulse template to an actual pulse template. By accessing the integral property above we froze it. Linking a pulse with a different property will result in an error."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 3,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "With wrong integral value:\n",
+      "RuntimeError('Cannot link to target. Wrong value of property \"integral\"')\n",
+      "the linking worked.  The new experiment has now a defined duration of Expression('t_init + t_manip + t_read') .\n"
+     ]
+    }
+   ],
+   "source": [
+    "my_readout_wrong_integral = AtomicMultiChannelPT(FunctionPT('1', 't_read', channel='X'),\n",
+    "                                                 FunctionPT('a*b', 't_read', channel='Y'))\n",
+    "\n",
+    "my_readout = AtomicMultiChannelPT(FunctionPT('1 / t_read', 't_read', channel='X'),\n",
+    "                                  FunctionPT('a*b / t_read', 't_read', channel='Y'))\n",
+    "\n",
+    "try:\n",
+    "    print('With wrong integral value:')\n",
+    "    abstract_readout.link_to(my_readout_wrong_integral)\n",
+    "except Exception as err:\n",
+    "    print(repr(err))\n",
+    "\n",
+    "abstract_readout.link_to(my_readout)\n",
+    "print('the linking worked.  The new experiment has now a defined duration of', repr(experiment.duration), '.')"
+   ]
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "Python 3",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.7.2"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 1
+}

doc/source/examples/examples.rst

@@ -19,6 +19,7 @@ All examples are provided as static text in this documentation and, additionally
     09ParameterConstraints
     10FreeInductionDecayExample
     11GateConfigurationExample
+    12AbstractPulseTemplate
 
 The `/doc/source/examples` directory also contains some outdated examples for features and functionality that has been changed. These examples start with the number nine and are currently left only for reference purposes.
 If you are just learning how to get around in qupulse please ignore them.

qupulse/pulses/__init__.py

@@ -1,6 +1,7 @@
 """This is the central package for defining pulses. All :class:`~qupulse.pulses.pulse_template.PulseTemplate`
 subclasses that are final and ready to be used are imported here with their recommended abbreviation as an alias."""
 
+from qupulse.pulses.abstract_pulse_template import AbstractPulseTemplate as AbstractPT
 from qupulse.pulses.function_pulse_template import FunctionPulseTemplate as FunctionPT
 from qupulse.pulses.loop_pulse_template import ForLoopPulseTemplate as ForLoopPT
 from qupulse.pulses.multi_channel_pulse_template import AtomicMultiChannelPulseTemplate as AtomicMultiChannelPT
@@ -17,5 +18,5 @@
     import qupulse.pulses.pulse_template_parameter_mapping
 
 __all__ = ["FunctionPT", "ForLoopPT", "AtomicMultiChannelPT", "MappingPT", "RepetitionPT", "SequencePT", "TablePT",
-           "PointPT"]
+           "PointPT", "AbstractPT"]
 

qupulse/pulses/abstract_pulse_template.py

@@ -0,0 +1,157 @@
+from typing import Set, Optional, Dict, Any, cast
+from functools import partial, partialmethod
+import warnings
+
+from qupulse import ChannelID
+from qupulse.expressions import ExpressionScalar
+from qupulse.serialization import PulseRegistryType
+from qupulse.pulses.pulse_template import PulseTemplate
+
+
+__all__ = ["AbstractPulseTemplate", "UnlinkWarning"]
+
+
+class AbstractPulseTemplate(PulseTemplate):
+    def __init__(self, identifier: str,
+                 *,
+                 defined_channels: Optional[Set[ChannelID]]=None,
+                 parameter_names: Optional[Set[str]]=None,
+                 measurement_names: Optional[Set[str]]=None,
+                 integral: Optional[Dict[ChannelID, ExpressionScalar]]=None,
+                 duration: Optional[ExpressionScalar]=None,
+                 is_interruptable: Optional[bool]=None,
+                 registry: Optional[PulseRegistryType]=None):
+        """This pulse template can be used as a place holder for a pulse template with a defined interface. Pulse
+        template properties like `defined_channels` can be passed on initialization to declare those properties who make
+        up the interface. Omitted properties raise an `NotSpecifiedError` exception if accessed. Properties which have
+        been accessed are marked as "frozen".
+
+        The abstract pulse template can be linked to another pulse template by calling the `link_to` member. The target
+        has to have the same properties for all properties marked as "frozen". This ensures a property always returns
+        the same value.
+
+        Example:
+            >>> abstract_readout = AbstractPulseTemplate('readout', defined_channels={'X', 'Y'})
+            >>> assert abstract_readout.defined_channels == {'X', 'Y'}
+
+            This will raise an exception
+            >>> print(abstract_readout.duration)
+
+        Args:
+            identifier: Mandatory property
+            defined_channels: Optional property
+            parameter_names: Optional property
+            measurement_names: Optional property
+            integral: Optional property
+            duration: Optional property
+            is_interruptable: Optional property
+            registry: Instance is registered here if specified
+        """
+        super().__init__(identifier=identifier)
+
+        self._declared_properties = {}
+        self._frozen_properties = set()
+
+        if defined_channels is not None:
+            self._declared_properties['defined_channels'] = set(defined_channels)
+
+        if parameter_names is not None:
+            self._declared_properties['parameter_names'] = set(map(str, parameter_names))
+
+        if measurement_names is not None:
+            self._declared_properties['measurement_names'] = set(map(str, measurement_names))
+
+        if integral is not None:
+            if defined_channels is not None and integral.keys() != defined_channels:
+                raise ValueError('Integral does not fit to defined channels', integral.keys(), defined_channels)
+            self._declared_properties['integral'] = {channel: ExpressionScalar(value)
+                                                     for channel, value in integral.items()}
+
+        if duration:
+            self._declared_properties['duration'] = ExpressionScalar(duration)
+
+        if is_interruptable is not None:
+            self._declared_properties['is_interruptable'] = bool(is_interruptable)
+
+        self._linked_target = None
+        self.serialize_linked = False
+
+        self._register(registry=registry)
+
+    def link_to(self, target: PulseTemplate, serialize_linked: bool=None):
+        """Link to another pulse template.
+
+        Args:
+            target: Forward all getattr calls to this pulse template
+            serialize_linked: If true, serialization will be forwarded. Otherwise serialization will ignore the link
+        """
+        if self._linked_target:
+            raise RuntimeError('Cannot is already linked. If you REALLY need to relink call unlink() first.')
+
+        for frozen_property in self._frozen_properties:
+            if self._declared_properties[frozen_property] != getattr(target, frozen_property):
+                raise RuntimeError('Cannot link to target. Wrong value of property "%s"' % frozen_property)
+
+        if serialize_linked is not None:
+            self.serialize_linked = serialize_linked
+        self._linked_target = target
+
+    def unlink(self):
+        """Unlink a linked target. This might lead to unexpected behaviour as forwarded get attributes are not frozen"""
+        if self._linked_target:
+            warnings.warn("This might lead to unexpected behaviour as forwarded attributes are not frozen. Parent pulse"
+                          " templates might rely on certain properties to be constant (for example due to caching).",
+                          UnlinkWarning)
+        self._linked_target = None
+
+    def __getattr__(self, item: str) -> Any:
+        """Forward all unknown attribute accesses."""
+        return getattr(self._linked_target, item)
+
+    def get_serialization_data(self, serializer=None) -> Dict:
+        if self._linked_target and self.serialize_linked:
+            return self._linked_target.get_serialization_data(serializer=serializer)
+
+        if serializer:
+            raise RuntimeError('Old serialization not supported in new class')
+
+        data = super().get_serialization_data()
+        data.update(self._declared_properties)
+        return data
+
+    def _get_property(self, property_name: str) -> Any:
+        if self._linked_target:
+            return getattr(self._linked_target, property_name)
+        elif property_name in self._declared_properties:
+            self._frozen_properties.add(property_name)
+            return self._declared_properties[property_name]
+        else:
+            raise NotSpecifiedError(self.identifier, property_name)
+
+    def _forward_if_linked(self, method_name: str, *args, **kwargs) -> Any:
+        if self._linked_target:
+            return getattr(self._linked_target, method_name)(*args, **kwargs)
+        else:
+            raise RuntimeError('Cannot call "%s". No linked target to refer to', method_name)
+
+    def _internal_create_program(self, **kwargs):
+        raise NotImplementedError('this should never be called as we overrode _create_program')  # pragma: no cover
+
+    _create_program = partialmethod(_forward_if_linked, '_create_program')
+    build_sequence = partialmethod(_forward_if_linked, 'build_sequence')
+    requires_stop = partialmethod(_forward_if_linked, 'requires_stop')
+
+    is_interruptable = property(partial(_get_property, property_name='is_interruptable'))
+    defined_channels = property(partial(_get_property, property_name='defined_channels'))
+    duration = property(partial(_get_property, property_name='duration'))
+    measurement_names = property(partial(_get_property, property_name='measurement_names'))
+    integral = property(partial(_get_property, property_name='integral'))
+    parameter_names = property(partial(_get_property, property_name='parameter_names'))
+
+
+class NotSpecifiedError(RuntimeError):
+    pass
+
+
+class UnlinkWarning(UserWarning):
+    pass