"""This module defines LoopPulseTemplate, a higher-order hierarchical pulse template that loops
another PulseTemplate based on a condition."""
from typing import Dict, Set, Optional, Any, Union, Tuple, Generator, Sequence, cast
import warnings
import sympy
from cached_property import cached_property
from qupulse.serialization import Serializer, PulseRegistryType
from qupulse._program._loop import Loop
from qupulse.expressions import ExpressionScalar
from qupulse.utils import checked_int_cast
from qupulse.pulses.parameters import Parameter, ConstantParameter, InvalidParameterNameException, ParameterConstrainer, ParameterNotProvidedException
from qupulse.pulses.pulse_template import PulseTemplate, ChannelID
from qupulse.pulses.conditions import Condition, ConditionMissingException
from qupulse._program.instructions import InstructionBlock
from qupulse.pulses.sequencing import Sequencer
from qupulse._program.waveforms import SequenceWaveform as ForLoopWaveform
from qupulse.pulses.measurement import MeasurementDefiner, MeasurementDeclaration
__all__ = ['ForLoopPulseTemplate', 'LoopPulseTemplate', 'LoopIndexNotUsedException']
[docs]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]):
super().__init__(identifier=identifier)
self.__body = body
@property
def body(self) -> PulseTemplate:
return self.__body
@property
def defined_channels(self) -> Set['ChannelID']:
return self.__body.defined_channels
@property
def measurement_names(self) -> Set[str]:
return self.__body.measurement_names
@property
def is_interruptable(self):
raise NotImplementedError() # pragma: no cover
class ParametrizedRange:
"""Like the builtin python range but with parameters."""
def __init__(self, *args, **kwargs):
"""Positional and keyword arguments cannot be mixed.
Args:
*args: Interpreted as ``(start, )`` or ``(start, stop[, step])``
**kwargs: Expected to contain ``start``, ``stop`` and ``step``
Raises:
TypeError: If positional and keyword arguments are mixed
KeyError: If keyword arguments but one of ``start``, ``stop`` or ``step`` is missing
"""
if args and kwargs:
raise TypeError('ParametrizedRange only takes either positional or keyword arguments')
elif kwargs:
start = kwargs['start']
stop = kwargs['stop']
step = kwargs['step']
elif len(args) in (1, 2, 3):
if len(args) == 3:
start, stop, step = args
elif len(args) == 2:
(start, stop), step = args, 1
elif len(args) == 1:
start, (stop,), step = 0, args, 1
else:
raise TypeError('ParametrizedRange expected 1 to 3 arguments, got {}'.format(len(args)))
self.start = ExpressionScalar.make(start)
self.stop = ExpressionScalar.make(stop)
self.step = ExpressionScalar.make(step)
def to_tuple(self) -> Tuple[Any, Any, Any]:
"""Return a simple representation of the range which is useful for comparison and serialization"""
return (self.start.get_serialization_data(),
self.stop.get_serialization_data(),
self.step.get_serialization_data())
def to_range(self, parameters: Dict[str, Any]) -> range:
return range(checked_int_cast(self.start.evaluate_numeric(**parameters)),
checked_int_cast(self.stop.evaluate_numeric(**parameters)),
checked_int_cast(self.step.evaluate_numeric(**parameters)))
@property
def parameter_names(self) -> Set[str]:
return set(self.start.variables) | set(self.stop.variables) | set(self.step.variables)
[docs]class ForLoopPulseTemplate(LoopPulseTemplate, MeasurementDefiner, ParameterConstrainer):
"""This pulse template allows looping through an parametrized integer range and provides the loop index as a
parameter to the body. If you do not need the index in the pulse template, consider using
:class:`~qupulse.pulses.repetition_pulse_template.RepetitionPulseTemplate`"""
[docs] def __init__(self,
body: PulseTemplate,
loop_index: str,
loop_range: Union[int,
range,
str,
Tuple[Any, Any],
Tuple[Any, Any, Any],
ParametrizedRange],
identifier: Optional[str]=None,
*,
measurements: Optional[Sequence[MeasurementDeclaration]]=None,
parameter_constraints: Optional[Sequence]=None,
registry: PulseRegistryType=None) -> 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)
MeasurementDefiner.__init__(self, measurements=measurements)
ParameterConstrainer.__init__(self, parameter_constraints=parameter_constraints)
if isinstance(loop_range, ParametrizedRange):
self._loop_range = loop_range
elif isinstance(loop_range, (int, str)):
self._loop_range = ParametrizedRange(loop_range)
elif isinstance(loop_range, (tuple, list)):
self._loop_range = ParametrizedRange(*loop_range)
elif isinstance(loop_range, range):
self._loop_range = ParametrizedRange(start=loop_range.start,
stop=loop_range.stop,
step=loop_range.step)
else:
raise ValueError('loop_range is not valid')
if not loop_index.isidentifier():
raise InvalidParameterNameException(loop_index)
body_parameters = self.body.parameter_names
if loop_index not in body_parameters:
raise LoopIndexNotUsedException(loop_index, body_parameters)
self._loop_index = loop_index
if self.loop_index in self.constrained_parameters:
constraints = [str(constraint) for constraint in self.parameter_constraints
if self._loop_index in constraint.affected_parameters]
warnings.warn("ForLoopPulseTemplate was created with a constraint on a variable shadowing the loop index.\n" \
"This will not constrain the actual loop index but introduce a new parameter.\n" \
"To constrain the loop index, put the constraint in the body subtemplate.\n" \
"Loop index is {} and offending constraints are: {}".format(self._loop_index, constraints))
self._register(registry=registry)
@property
def loop_index(self) -> str:
return self._loop_index
@property
def loop_range(self) -> ParametrizedRange:
return self._loop_range
@property
def measurement_names(self) -> Set[str]:
return LoopPulseTemplate.measurement_names.fget(self) | MeasurementDefiner.measurement_names.fget(self)
@cached_property
def duration(self) -> ExpressionScalar:
step_size = self._loop_range.step.sympified_expression
loop_index = sympy.symbols(self._loop_index)
sum_index = sympy.symbols(self._loop_index)
# replace loop_index with sum_index dependable expression
body_duration = self.body.duration.sympified_expression.subs({loop_index: self._loop_range.start.sympified_expression + sum_index*step_size})
# number of sum contributions
step_count = sympy.ceiling((self._loop_range.stop.sympified_expression-self._loop_range.start.sympified_expression) / step_size)
sum_start = 0
sum_stop = sum_start + (sympy.functions.Max(step_count, 1) - 1)
# expression used if step_count >= 0
finite_duration_expression = sympy.Sum(body_duration, (sum_index, sum_start, sum_stop))
duration_expression = sympy.Piecewise((0, step_count <= 0),
(finite_duration_expression, True))
return ExpressionScalar(duration_expression)
@property
def parameter_names(self) -> Set[str]:
parameter_names = self.body.parameter_names.copy()
parameter_names.remove(self._loop_index)
return parameter_names | self._loop_range.parameter_names | self.constrained_parameters | self.measurement_parameters
def _body_parameter_generator(self, parameters: Dict[str, Parameter], forward=True) -> Generator:
loop_range_parameters = dict((parameter_name, parameters[parameter_name].get_value())
for parameter_name in self._loop_range.parameter_names)
loop_range = self._loop_range.to_range(loop_range_parameters)
parameters = dict((parameter_name, parameters[parameter_name])
for parameter_name in self.body.parameter_names if parameter_name != self._loop_index)
loop_range = loop_range if forward else reversed(loop_range)
for loop_index_value in loop_range:
local_parameters = parameters.copy()
local_parameters[self._loop_index] = ConstantParameter(loop_index_value)
yield local_parameters
[docs] def build_sequence(self,
sequencer: Sequencer,
parameters: Dict[str, Parameter],
conditions: Dict[str, Condition],
measurement_mapping: Dict[str, str],
channel_mapping: Dict[ChannelID, ChannelID],
instruction_block: InstructionBlock) -> None:
self.validate_parameter_constraints(parameters=parameters)
self.insert_measurement_instruction(instruction_block=instruction_block,
parameters=parameters,
measurement_mapping=measurement_mapping)
for local_parameters in self._body_parameter_generator(parameters, forward=False):
sequencer.push(self.body,
parameters=local_parameters,
conditions=conditions,
window_mapping=measurement_mapping,
channel_mapping=channel_mapping,
target_block=instruction_block)
def _internal_create_program(self, *,
parameters: Dict[str, Parameter],
measurement_mapping: Dict[str, Optional[str]],
channel_mapping: Dict[ChannelID, Optional[ChannelID]],
global_transformation: Optional['Transformation'],
to_single_waveform: Set[Union[str, 'PulseTemplate']],
parent_loop: Loop) -> None:
self.validate_parameter_constraints(parameters=parameters)
try:
measurement_parameters = {parameter_name: parameters[parameter_name].get_value()
for parameter_name in self.measurement_parameters}
duration_parameters = {parameter_name: parameters[parameter_name].get_value()
for parameter_name in self.duration.variables}
except KeyError as e:
raise ParameterNotProvidedException(str(e)) from e
if self.duration.evaluate_numeric(**duration_parameters) > 0:
measurements = self.get_measurement_windows(measurement_parameters, measurement_mapping)
if measurements:
parent_loop.add_measurements(measurements)
for local_parameters in self._body_parameter_generator(parameters, forward=True):
self.body._create_program(parameters=local_parameters,
measurement_mapping=measurement_mapping,
channel_mapping=channel_mapping,
global_transformation=global_transformation,
to_single_waveform=to_single_waveform,
parent_loop=parent_loop)
[docs] def requires_stop(self,
parameters: Dict[str, Parameter],
conditions: Dict[str, 'Condition']) -> bool:
return any(parameters[parameter_name].requires_stop for parameter_name in self._loop_range.parameter_names)
[docs] 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
[docs] @classmethod
def deserialize(cls, serializer: Optional[Serializer]=None, **kwargs) -> 'ForLoopPulseTemplate':
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]:
step_size = self._loop_range.step.sympified_expression
loop_index = sympy.symbols(self._loop_index)
sum_index = sympy.symbols(self._loop_index)
body_integrals = self.body.integral
body_integrals = {
c: body_integrals[c].sympified_expression.subs(
{loop_index: self._loop_range.start.sympified_expression + sum_index*step_size}
)
for c in body_integrals
}
# number of sum contributions
step_count = sympy.ceiling((self._loop_range.stop.sympified_expression-self._loop_range.start.sympified_expression) / step_size)
sum_start = 0
sum_stop = sum_start + (sympy.functions.Max(step_count, 1) - 1)
for c in body_integrals:
channel_integral_expr = sympy.Sum(body_integrals[c], (sum_index, sum_start, sum_stop))
body_integrals[c] = ExpressionScalar(channel_integral_expr)
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,
registry: PulseRegistryType=None) -> None:
"""Create a new LoopPulseTemplate instance.
Args:
condition (str): A unique identifier for the looping condition. Will be used to obtain
the Condition object from the mapping passed in during the sequencing process.
body (PulseTemplate): The PulseTemplate which will be repeated as long as the condition
holds.
identifier (str): A unique identifier for use in serialization. (optional)
"""
super().__init__(body=body, identifier=identifier)
self._condition = condition
self._register(registry=registry)
def __str__(self) -> str:
return "LoopPulseTemplate: Condition <{}>, Body <{}>".format(self._condition, self.body)
@property
def condition(self) -> str:
"""This LoopPulseTemplate's condition."""
return self._condition
@property
def parameter_names(self) -> Set[str]:
return self.body.parameter_names
@property
def duration(self) -> ExpressionScalar:
return ExpressionScalar('nan')
def __obtain_condition_object(self, conditions: Dict[str, Condition]) -> Condition:
try:
return conditions[self._condition]
except:
raise ConditionMissingException(self._condition)
def build_sequence(self,
sequencer: Sequencer,
parameters: Dict[str, Parameter],
conditions: Dict[str, Condition],
measurement_mapping: Dict[str, str],
channel_mapping: Dict[ChannelID, ChannelID],
instruction_block: InstructionBlock) -> None:
self.__obtain_condition_object(conditions).build_sequence_loop(self,
self.body,
sequencer,
parameters,
conditions,
measurement_mapping,
channel_mapping,
instruction_block)
def _internal_create_program(self, *, # pragma: no cover
parameters: Dict[str, Parameter],
measurement_mapping: Dict[str, Optional[str]],
channel_mapping: Dict[ChannelID, Optional[ChannelID]],
parent_loop: Loop) -> None:
raise NotImplementedError("create_program() does not handle conditions/triggers right now and cannot "
"be meaningfully implemented for a WhileLoopPulseTemplate")
def requires_stop(self,
parameters: Dict[str, Parameter],
conditions: Dict[str, Condition]) -> bool:
return self.__obtain_condition_object(conditions).requires_stop()
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
@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]:
return {c: ExpressionScalar('nan') for c in self.body.defined_channels}
[docs]class LoopIndexNotUsedException(Exception):
def __init__(self, loop_index: str, body_parameter_names: Set[str]):
self.loop_index = loop_index
self.body_parameter_names = body_parameter_names
def __str__(self) -> str:
return "The parameter {} is missing in the body's parameter names: {}".format(self.loop_index,
self.body_parameter_names)