diff --git a/ReleaseNotes.txt b/ReleaseNotes.txt index 607028696..31669bd0f 100644 --- a/ReleaseNotes.txt +++ b/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 diff --git a/doc/source/examples/12AbstractPulseTemplate.ipynb b/doc/source/examples/12AbstractPulseTemplate.ipynb new file mode 100644 index 000000000..050e81f91 --- /dev/null +++ b/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 +} diff --git a/doc/source/examples/examples.rst b/doc/source/examples/examples.rst index c017d185c..55a5d65dc 100644 --- a/doc/source/examples/examples.rst +++ b/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. \ No newline at end of file diff --git a/qupulse/pulses/__init__.py b/qupulse/pulses/__init__.py index 99fe57f86..2d70368fb 100644 --- a/qupulse/pulses/__init__.py +++ b/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"] diff --git a/qupulse/pulses/abstract_pulse_template.py b/qupulse/pulses/abstract_pulse_template.py new file mode 100644 index 000000000..12dcbeab2 --- /dev/null +++ b/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 diff --git a/tests/pulses/abstract_pulse_template_tests.py b/tests/pulses/abstract_pulse_template_tests.py new file mode 100644 index 000000000..527fa0d4f --- /dev/null +++ b/tests/pulses/abstract_pulse_template_tests.py @@ -0,0 +1,184 @@ +import unittest +import warnings +from unittest import mock + +from qupulse.expressions import ExpressionScalar +from qupulse.pulses.abstract_pulse_template import AbstractPulseTemplate, NotSpecifiedError, UnlinkWarning + +from tests.pulses.sequencing_dummies import DummyPulseTemplate + + +class AbstractPulseTemplateTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.freezable_property_values = { + 'defined_channels': {'A', 'B'}, + 'duration': ExpressionScalar(4), + 'measurement_names': {'m', 'n'}, + 'integral': {'A': ExpressionScalar('a*b'), 'B': ExpressionScalar(32)}, + 'parameter_names': {'a', 'b', 'c'} + } + + def test_minimal_init(self): + apt = AbstractPulseTemplate(identifier='my_apt') + + self.assertEqual(apt._frozen_properties, set()) + self.assertEqual(apt._declared_properties, {}) + self.assertEqual(apt.identifier, 'my_apt') + + def test_invalid_integral(self): + with self.assertRaisesRegex(ValueError, 'Integral'): + AbstractPulseTemplate(identifier='my_apt', integral={'X': 1}, defined_channels={'A'}) + + def test_declaring(self): + apt = AbstractPulseTemplate(identifier='my_apt', defined_channels={'A'}, is_interruptable=True) + + self.assertEqual(apt._frozen_properties, set()) + self.assertEqual(apt._declared_properties, {'defined_channels': {'A'}, 'is_interruptable': True}) + self.assertEqual(apt.identifier, 'my_apt') + self.assertEqual(apt.is_interruptable, True) + + def test_freezing(self): + apt = AbstractPulseTemplate(identifier='my_apt', defined_channels={'A'}) + + # freeze + self.assertEqual(apt.defined_channels, {'A'}) + + self.assertEqual(apt._frozen_properties, {'defined_channels'}) + self.assertEqual(apt._declared_properties, {'defined_channels': {'A'}}) + + apt = AbstractPulseTemplate(identifier='my_apt', **self.freezable_property_values) + expected_frozen = set() + + for property_name, valid_value in self.freezable_property_values.items(): + self.assertEqual(apt._frozen_properties, expected_frozen) + + self.assertEqual(getattr(apt, property_name), valid_value) + expected_frozen.add(property_name) + + self.assertEqual(apt._frozen_properties, expected_frozen) + + def test_unspecified(self): + specified = {} + unspecified = self.freezable_property_values.copy() + + for property_name, valid_value in self.freezable_property_values.items(): + specified[property_name] = unspecified.pop(property_name) + + apt = AbstractPulseTemplate(identifier='my_apt', **specified) + + for x, v in specified.items(): + self.assertEqual(v, getattr(apt, x)) + + for unspecified_property_name in unspecified: + with self.assertRaisesRegex(NotSpecifiedError, unspecified_property_name, + msg=unspecified_property_name): + getattr(apt, unspecified_property_name) + + def test_linking(self): + apt = AbstractPulseTemplate(identifier='apt') + + linked = DummyPulseTemplate() + + self.assertIsNone(apt._linked_target) + + apt.link_to(linked) + + self.assertIs(linked, apt._linked_target) + + with self.assertRaisesRegex(RuntimeError, 'already linked'): + apt.link_to(DummyPulseTemplate()) + + def test_linking_wrong_frozen(self): + apt = AbstractPulseTemplate(identifier='my_apt', defined_channels={'A'}) + + dummy = DummyPulseTemplate(defined_channels={'B'}) + apt.link_to(dummy) + + self.assertEqual(apt.defined_channels, dummy.defined_channels) + + apt = AbstractPulseTemplate(identifier='my_apt', defined_channels={'A'}) + + # freeze + apt.defined_channels + + dummy = DummyPulseTemplate(defined_channels={'B'}) + + with self.assertRaisesRegex(RuntimeError, 'Wrong value of property "defined_channels"'): + apt.link_to(dummy) + + def test_method_forwarding(self): + apt = AbstractPulseTemplate(identifier='my_apt') + + args = ([], {}, 'asd') + kwargs = {'kw1': [], 'kw2': {}} + + forwarded_methods = ['build_sequence', '_create_program', 'requires_stop'] + + for method_name in forwarded_methods: + method = getattr(apt, method_name) + with self.assertRaisesRegex(RuntimeError, 'No linked target'): + method(*args, **kwargs) + + linked = mock.MagicMock() + apt.link_to(linked) + + for method_name in forwarded_methods: + method = getattr(apt, method_name) + mock_method = getattr(linked, method_name) + + method(*args, **kwargs) + + mock_method.assert_called_once_with(*args, **kwargs) + + def test_forwarded_get_attr(self): + apt = AbstractPulseTemplate(identifier='my_apt') + + self.assertFalse(hasattr(apt, 'test')) + + linked = mock.MagicMock() + + apt.link_to(linked) + + self.assertTrue(hasattr(apt, 'test')) + self.assertIs(apt.test, linked.test) + + def test_serialization(self): + defined_channels = {'X', 'Y'} + properties = {'defined_channels': defined_channels, 'duration': 5} + + apt = AbstractPulseTemplate(identifier='my_apt', **properties) + + serializer = mock.MagicMock() + with self.assertRaisesRegex(RuntimeError, "not supported"): + apt.get_serialization_data(serializer=serializer) + + expected = {**properties, + '#identifier': 'my_apt', + '#type': 'qupulse.pulses.abstract_pulse_template.AbstractPulseTemplate'} + self.assertEqual(apt.get_serialization_data(), expected) + + dummy = DummyPulseTemplate(**properties) + apt.link_to(dummy) + + self.assertEqual(apt.get_serialization_data(), expected) + apt = AbstractPulseTemplate(identifier='my_apt', **properties) + apt.link_to(dummy, serialize_linked=True) + expected = dummy.get_serialization_data() + self.assertEqual(apt.get_serialization_data(), expected) + + def test_unlink(self): + apt = AbstractPulseTemplate(identifier='my_apt') + dummy = DummyPulseTemplate() + + with warnings.catch_warnings(record=True) as w: + warnings.simplefilter("always") + apt.unlink() + + self.assertFalse(w) + + apt.link_to(dummy) + with self.assertWarns(UnlinkWarning): + apt.unlink() + + self.assertIsNone(apt._linked_target)