diff --git a/changes.d/933.feature b/changes.d/933.feature new file mode 100644 index 00000000..2f44e23f --- /dev/null +++ b/changes.d/933.feature @@ -0,0 +1 @@ +Replace PulseTemplate._create_program(**a_lot_of__kwargs) with PulseTemplate._build_program(program_builder). \ No newline at end of file diff --git a/qupulse/hardware/awgs/base.py b/qupulse/hardware/awgs/base.py index 7b949b18..2a62fdc3 100644 --- a/qupulse/hardware/awgs/base.py +++ b/qupulse/hardware/awgs/base.py @@ -20,7 +20,7 @@ from qupulse.hardware.util import get_sample_times, not_none_indices from qupulse.utils.types import ChannelID -from qupulse.program.linspace import LinSpaceNode, LinSpaceTopLevel, Play, \ +from qupulse.program.linspace import LinSpaceNode, LinSpaceProgram, Play, \ transform_linspace_commands, to_increment_commands from qupulse.program.loop import Loop from qupulse.program.waveforms import Waveform @@ -178,7 +178,7 @@ def __str__(self) -> str: #!!! typehint obsolete -AllowedProgramTypes = Union[Loop,LinSpaceTopLevel,] +AllowedProgramTypes = Union[Loop,LinSpaceProgram,] class ChannelTransformation(NamedTuple): diff --git a/qupulse/program/linspace.py b/qupulse/program/linspace.py index 870e3d29..c610ff98 100644 --- a/qupulse/program/linspace.py +++ b/qupulse/program/linspace.py @@ -13,14 +13,15 @@ from dataclasses import dataclass from abc import ABC, abstractmethod from typing import Mapping, Optional, Sequence, ContextManager, Iterable, Tuple, \ - Union, Dict, List, Set, ClassVar, Callable, Any + Union, Dict, List, Set, ClassVar, Callable, Any, AbstractSet from collections import OrderedDict from qupulse import ChannelID, MeasurementWindow from qupulse.parameter_scope import Scope, MappedScope, FrozenDict -from qupulse.program.protocol import (ProgramBuilder, Waveform, ) +from qupulse.program.protocol import (ProgramBuilder, Waveform, BaseProgramBuilder, Program, ) from qupulse.program.values import RepetitionCount, HardwareTime, HardwareVoltage, DynamicLinearValue, TimeType from qupulse.program.volatile import VolatileRepetitionCount, InefficientVolatility +from qupulse.program.waveforms import TransformingWaveform # this resolution is used to unify increments # the increments themselves remain floats @@ -173,7 +174,7 @@ def reversed(self, offset: int, lengths: list): return reversed_iter -class LinSpaceBuilder(ProgramBuilder): +class LinSpaceBuilder(BaseProgramBuilder): """This program builder supports efficient translation of pulse templates that use symbolic linearly spaced voltages and durations. @@ -208,7 +209,7 @@ def inner_scope(self, scope: Scope) -> Scope: def _get_ranges(self): return dict(self._ranges) - def hold_voltage(self, duration: HardwareTime, voltages: Mapping[ChannelID, HardwareVoltage]): + def _transformed_hold_voltage(self, duration: HardwareTime, voltages: Mapping[ChannelID, HardwareVoltage]): voltages = sorted((self._name_to_idx[ch_name], value) for ch_name, value in voltages.items()) voltages = [value for _, value in voltages] @@ -249,7 +250,7 @@ def hold_voltage(self, duration: HardwareTime, voltages: Mapping[ChannelID, Hard self._stack[-1].append(set_cmd) - def play_arbitrary_waveform(self, waveform: Waveform): + def _transformed_play_arbitrary_waveform(self, waveform: Waveform): return self._stack[-1].append(LinSpaceArbitraryWaveform(waveform, self._idx_to_name)) def measure(self, measurements: Optional[Sequence[MeasurementWindow]]): @@ -272,7 +273,7 @@ def with_repetition(self, repetition_count: RepetitionCount, @contextlib.contextmanager def with_sequence(self, - measurements: Optional[Sequence[MeasurementWindow]] = None) -> ContextManager['ProgramBuilder']: + measurements: Optional[Sequence[MeasurementWindow]] = None) -> Iterable['ProgramBuilder']: yield self def new_subprogram(self, global_transformation: 'Transformation' = None) -> ContextManager['ProgramBuilder']: @@ -284,28 +285,30 @@ def with_iteration(self, index_name: str, rng: range, return self._stack.append([]) self._ranges.append((index_name, rng)) - yield self + scope = self.build_context.scope.overwrite({index_name: DynamicLinearValue(base=0, factors={index_name: 1})}) + with self._with_patched_context(scope=scope): + yield self cmds = self._stack.pop() self._ranges.pop() if cmds: self._stack[-1].append(LinSpaceIter(body=tuple(cmds), length=len(rng))) @contextlib.contextmanager - def time_reversed(self) -> ContextManager['LinSpaceBuilder']: + def time_reversed(self) -> Iterable['LinSpaceBuilder']: self._stack.append([]) yield self inner = self._stack.pop() offset = len(self._ranges) self._stack[-1].extend(node.reversed(offset, []) for node in reversed(inner)) - def to_program(self, defined_channels: set[ChannelID]) -> Optional['LinSpaceTopLevel']: - if self._root(): - return LinSpaceTopLevel(body=tuple(self._root()), - _defined_channels=defined_channels) - #TODO: the argument defined_channels currently does not do anything, - #it is included for convenience for the HDAWGLinspaceBuilder - #The extra class for the top level is to shift program specific functionality - #from awg/base to the respective program files. + def to_program(self) -> Optional['LinSpaceProgram']: + if root := self._root(): + return LinSpaceProgram( + root=tuple(root), + defined_channels=self._idx_to_name, + ) + else: + return None @dataclass @@ -510,10 +513,10 @@ def add_node(self, node: Union[LinSpaceNode, Sequence[LinSpaceNode]]): raise TypeError("The node type is not handled", type(node), node) -def to_increment_commands(linspace_nodes: 'LinSpaceTopLevel') -> List[Command]: +def to_increment_commands(linspace_nodes: 'LinSpaceProgram') -> List[Command]: """translate the given linspace node tree to a minimal sequence of set and increment commands as well as loops.""" state = _TranslationState() - state.add_node(linspace_nodes.body) + state.add_node(linspace_nodes.root) return state.commands @@ -550,21 +553,24 @@ def _get_waveforms_dict(transformed_commands: Sequence[Command]) -> Mapping[Wave @dataclass -class LinSpaceTopLevel(LinSpaceNode): - - body: Tuple[LinSpaceNode, ...] - _defined_channels: set[ChannelID] +class LinSpaceProgram(Program): + root: Tuple[LinSpaceNode, ...] + defined_channels: Tuple[ChannelID, ...] + + @property + def duration(self) -> TimeType: + raise NotImplementedError("TODO") + + def get_defined_channels(self) -> AbstractSet[ChannelID]: + return set(self.defined_channels) def dependencies(self): dependencies = {} - for node in self.body: + for node in self.root: for idx, deps in node.dependencies().items(): dependencies.setdefault(idx, set()).update(deps) return dependencies - def get_defined_channels(self) -> set[ChannelID]: - return self._defined_channels - def get_waveforms_dict(self, channels: Sequence[ChannelID], #!!! this argument currently does not do anything. channel_transformations: Mapping[ChannelID,'ChannelTransformation'], diff --git a/qupulse/program/loop.py b/qupulse/program/loop.py index 46860ebe..952d94c1 100644 --- a/qupulse/program/loop.py +++ b/qupulse/program/loop.py @@ -3,7 +3,8 @@ # SPDX-License-Identifier: LGPL-3.0-or-later """Program builder implementation that creates the legacy :py:class:`.Loop`.""" - +import contextlib +import copy import reprlib import warnings from collections import defaultdict, OrderedDict @@ -11,18 +12,21 @@ from dataclasses import dataclass from enum import Enum from typing import Set, Union, Iterable, Optional, Sequence, Tuple, List, \ - Generator, Mapping, cast, Dict, ContextManager, Any + Generator, Mapping, cast, Dict, ContextManager, Any, AbstractSet, Iterator import numpy as np -from qupulse.parameter_scope import Scope +from qupulse.expressions import Expression +from qupulse.parameter_scope import Scope, DictScope from qupulse.program import ProgramBuilder +from qupulse.program.protocol import BuildContext, BuildSettings, BaseProgramBuilder from qupulse.program.values import RepetitionCount, HardwareTime, HardwareVoltage -from qupulse.program.transformation import Transformation +from qupulse.program.transformation import Transformation, chain_transformations from qupulse.program.volatile import VolatileRepetitionCount, VolatileProperty, VolatileModificationWarning from qupulse.program.waveforms import SequenceWaveform, RepetitionWaveform from qupulse.program.waveforms import TransformingWaveform from qupulse.program.waveforms import Waveform, ConstantWaveform +from qupulse.pulses.metadata import TemplateMetadata from qupulse.pulses.range import RangeScope from qupulse.utils import is_integer from qupulse.utils.numeric import smallest_factor_ge @@ -136,7 +140,7 @@ def add_measurements(self, measurements: Iterable[MeasurementWindow]): else: self._measurements.extend(measurements) - def get_defined_channels(self) -> Set[ChannelID]: + def get_defined_channels(self) -> AbstractSet[ChannelID]: return next(self.get_depth_first_iterator()).waveform.defined_channels @property @@ -768,11 +772,10 @@ def add_measurements(self, measurements: List[MeasurementWindow]): @dataclass class StackFrame: loop: Union[Loop, LoopGuard] - iterating: Optional[Tuple[str, int]] -class LoopBuilder(ProgramBuilder): +class LoopBuilder(BaseProgramBuilder): """ Notes fduring implementation: @@ -780,19 +783,14 @@ class LoopBuilder(ProgramBuilder): """ - def __init__(self): + def __init__(self, initial_context: BuildContext = None, initial_settings: BuildSettings = None): + super().__init__(initial_context, initial_settings) + self._root: Loop = Loop() self._top: Union[Loop, LoopGuard] = self._root self._stack: List[StackFrame] = [StackFrame(self._root, None)] - def inner_scope(self, scope: Scope) -> Scope: - local_vars = self._stack[-1].iterating - if local_vars is None: - return scope - else: - return RangeScope(scope, *local_vars) - def _push(self, stack_entry: StackFrame): self._top = stack_entry.loop self._stack.append(stack_entry) @@ -824,13 +822,18 @@ def with_iteration(self, index_name: str, rng: range, measurements: Optional[Sequence[MeasurementWindow]] = None) -> Iterable['ProgramBuilder']: with self.with_sequence(measurements): top_frame = self._stack[-1] + context = copy.copy(self.build_context) + base_scope = context.scope + self._build_context_stack.append(context) for value in rng: top_frame.iterating = (index_name, value) + context.scope = RangeScope(base_scope, index_name, value) yield self + self._build_context_stack.pop() @contextmanager - def time_reversed(self) -> ContextManager['LoopBuilder']: - inner_builder = LoopBuilder() + def time_reversed(self) -> Iterator[ProgramBuilder]: + inner_builder = LoopBuilder(self.build_context, self.build_settings) yield inner_builder inner_program = inner_builder.to_program() @@ -839,21 +842,21 @@ def time_reversed(self) -> ContextManager['LoopBuilder']: self._try_append(inner_program, None) @contextmanager - def with_sequence(self, measurements: Optional[Sequence[MeasurementWindow]] = None) -> ContextManager['ProgramBuilder']: + def with_sequence(self, measurements: Optional[Sequence[MeasurementWindow]] = None) -> Iterator[ProgramBuilder]: top_frame = StackFrame(LoopGuard(self._top, measurements), None) self._push(top_frame) yield self self._pop() - def hold_voltage(self, duration: HardwareTime, voltages: Mapping[str, HardwareVoltage]): + def _transformed_hold_voltage(self, duration: HardwareTime, voltages: Mapping[str, HardwareVoltage]): self.play_arbitrary_waveform(ConstantWaveform.from_mapping(duration, voltages)) - def play_arbitrary_waveform(self, waveform: Waveform): + def _transformed_play_arbitrary_waveform(self, waveform: Waveform): self._top.append_child(waveform=waveform) @contextmanager - def new_subprogram(self, global_transformation: Transformation = None) -> ContextManager['ProgramBuilder']: - inner_builder = LoopBuilder() + def new_subprogram(self, global_transformation: Transformation = None) -> Iterator[ProgramBuilder]: + inner_builder = LoopBuilder(self.build_context, self.build_settings) yield inner_builder inner_program = inner_builder.to_program() @@ -867,13 +870,13 @@ def new_subprogram(self, global_transformation: Transformation = None) -> Contex waveform = TransformingWaveform.from_transformation(waveform, global_transformation) self.play_arbitrary_waveform(waveform) - def to_program(self, defined_channels: Set[ChannelID] = set()) -> Optional[Loop]: - #!!! the defined_channels argument is for convenience for HDAWGLinspacebuilder - # and not needed here. + def to_program(self) -> Optional[Loop]: if len(self._stack) != 1: warnings.warn("Creating program with active build stack.") if self._root.waveform or len(self._root.children) != 0: return self._root + else: + return None @classmethod def _testing_dummy(cls, stack): diff --git a/qupulse/program/protocol.py b/qupulse/program/protocol.py index a3692e4d..a7656a8e 100644 --- a/qupulse/program/protocol.py +++ b/qupulse/program/protocol.py @@ -1,15 +1,20 @@ """Definition of the program builder protocol.""" - -from abc import abstractmethod -from typing import runtime_checkable, Protocol, Mapping, Optional, Sequence, Iterable, ContextManager +import copy +import dataclasses +from abc import abstractmethod, ABC +from contextlib import contextmanager +from typing import runtime_checkable, Protocol, Mapping, Optional, Sequence, Iterable, ContextManager, AbstractSet, \ + Union from qupulse import MeasurementWindow -from qupulse.parameter_scope import Scope -from qupulse.program.waveforms import Waveform -from qupulse.program.transformation import Transformation +from qupulse.expressions import Expression +from qupulse.parameter_scope import Scope, MappedScope +from qupulse.program.waveforms import Waveform, ConstantWaveform, TransformingWaveform +from qupulse.program.transformation import Transformation, chain_transformations from qupulse.program.values import RepetitionCount, HardwareTime, HardwareVoltage +from qupulse.pulses.metadata import TemplateMetadata -from qupulse.utils.types import TimeType +from qupulse.utils.types import TimeType, ChannelID @runtime_checkable @@ -22,6 +27,44 @@ class Program(Protocol): def duration(self) -> TimeType: """The duration of the program in nanoseconds.""" + @abstractmethod + def get_defined_channels(self) -> AbstractSet[ChannelID]: + """Get the set of channels that are used in this program.""" + + +@dataclasses.dataclass +class BuildContext: + """This dataclass bundles the mutable context information during the build.""" + + scope: Scope = None + measurement_mapping: Mapping[str, Optional[str]] = None + channel_mapping: Mapping[ChannelID, Optional[ChannelID]] = None + transformation: Optional[Transformation] = None + minimal_sample_rate: Optional[TimeType] = None + + def apply_mappings(self, + parameter_mapping: Mapping[str, Expression] = None, + measurement_mapping: Mapping[str, Optional[str]] = None, + channel_mapping: Mapping[ChannelID, Optional[ChannelID]] = None, + ) -> "BuildContext": + scope = self.scope + if parameter_mapping is not None: + scope = MappedScope(scope=scope, mapping=parameter_mapping) + mapped_measurement_mapping = self.measurement_mapping + if measurement_mapping is not None: + mapped_measurement_mapping = {k: mapped_measurement_mapping[v] for k, v in measurement_mapping.items()} + mapped_channel_mapping = self.channel_mapping + if channel_mapping is not None: + mapped_channel_mapping = {inner_ch: None if outer_ch is None else mapped_channel_mapping[outer_ch] + for inner_ch, outer_ch in channel_mapping.items()} + return BuildContext(scope=scope, measurement_mapping=mapped_measurement_mapping, channel_mapping=mapped_channel_mapping, transformation=self.transformation, minimal_sample_rate=self.minimal_sample_rate) + + +@dataclasses.dataclass(frozen=True) +class BuildSettings: + """This dataclass bundles the immutable settings.""" + to_single_waveform: AbstractSet[str | object] + @runtime_checkable class ProgramBuilder(Protocol): @@ -30,10 +73,10 @@ class ProgramBuilder(Protocol): The pulse templates call the methods that correspond to their functionality on the program builder. For example, :py:class:`.ConstantPulseTemplate` translates itself into a simple :py:meth:`.ProgramBuilder.hold_voltage` call while - :class:`SequencePulseTemplate` uses :py:meth:`.ProgramBuilder.with_sequence` to signify a logical unit with + :py:class:`SequencePulseTemplate` uses :py:meth:`.ProgramBuilder.with_sequence` to signify a logical unit with attached measurements and passes the resulting object to the sequenced sub-templates. - Due to backward compatibility the handling of measurements is a bit weird since they have to be omitted in certain + Due to backward compatibility, the handling of measurements is a bit weird since they have to be omitted in certain cases. However, this is not relevant for HDAWG specific implementations because these are expected to ignore :py:meth:`.ProgramBuilder.measure` calls. @@ -41,19 +84,47 @@ class ProgramBuilder(Protocol): and repetition implementation. """ + @property @abstractmethod - def inner_scope(self, scope: Scope) -> Scope: - """This function is part of the iteration protocol and necessary to inject program builder specific parameter - implementations into the build process. :py:meth:`.ProgramBuilder.with_iteration` and - `.ProgramBuilder.with_iteration` callers *must* call this function inside the iteration. + def build_context(self) -> BuildContext: + """Get the current build context.""" - Args: - scope: The parameter scope outside the iteration. + @property + @abstractmethod + def build_settings(self) -> BuildSettings: + """Get the current build settings""" - Returns: - The parameter scope inside the iteration. + @abstractmethod + def override(self, + scope: Scope = None, + measurement_mapping: Optional[Mapping[str, Optional[str]]] = None, + channel_mapping: Optional[Mapping[ChannelID, Optional[ChannelID]]] = None, + global_transformation: Optional[Transformation] = None, + to_single_waveform: AbstractSet[str | object] = None,): + """Override the non-None values in context and settings""" + + @abstractmethod + def with_mappings(self, *, + parameter_mapping: Mapping[str, Expression], + measurement_mapping: Mapping[str, Optional[str]], + channel_mapping: Mapping[ChannelID, Optional[ChannelID]], + ) -> ContextManager['ProgramBuilder']: + """Modify the build context for the duration of the context manager. + + Args: + parameter_mapping: A mapping of parameter names to expressions. + measurement_mapping: A mapping of measurement names to measurement names or None. + channel_mapping: A mapping of channel IDs to channel IDs or None. """ + @abstractmethod + def with_transformation(self, transformation: Transformation) -> ContextManager['ProgramBuilder']: + """Modify the build context for the duration of the context manager.""" + + @abstractmethod + def with_metadata(self, metadata: TemplateMetadata) -> ContextManager['ProgramBuilder']: + """Modify the build context for the duration of the context manager.""" + @abstractmethod def hold_voltage(self, duration: HardwareTime, voltages: Mapping[str, HardwareVoltage]): """Hold the specified voltage for a given time. Advances the current time by ``duration``. The values are @@ -115,13 +186,10 @@ def with_sequence(self, """ @abstractmethod - def new_subprogram(self, global_transformation: 'Transformation' = None) -> ContextManager['ProgramBuilder']: + def new_subprogram(self) -> ContextManager['ProgramBuilder']: """Create a context managed program builder whose contents are translated into a single waveform upon exit if it is not empty. - Args: - global_transformation: This transformation is applied to the waveform - Returns: A context manager that returns a :py:class:`ProgramBuilder` on entering. """ @@ -153,3 +221,109 @@ def to_program(self) -> Optional[Program]: Returns: A program implementation. None if nothing was added to this program builder. """ + + +class BaseProgramBuilder(ProgramBuilder, ABC): + """Helper base class for program builder to reduce code duplication. The interface is defined by :py:class:`ProgramBuilder`. + + This class provides shared functionality for context and settings and correct transformation handling. + """ + + def __init__(self, initial_context: BuildContext = None, initial_settings: BuildSettings = None): + self._build_context_stack: list[BuildContext] = [BuildContext() if initial_context is None else initial_context] + self._build_settings_stack: list[BuildSettings] = [BuildSettings(set()) if initial_settings is None else initial_settings] + + @property + def build_context(self) -> BuildContext: + return self._build_context_stack[-1] + + @property + def build_settings(self) -> BuildSettings: + return self._build_settings_stack[-1] + + def override(self, + scope: Scope = None, + measurement_mapping: Optional[Mapping[str, Optional[str]]] = None, + channel_mapping: Optional[Mapping[ChannelID, Optional[ChannelID]]] = None, + global_transformation: Optional[Transformation] = None, + to_single_waveform: AbstractSet[Union[str, 'PulseTemplate']] = None): + old_context = self._build_context_stack[-1] + context = BuildContext( + scope=old_context.scope if scope is None else scope, + measurement_mapping=old_context.measurement_mapping if measurement_mapping is None else measurement_mapping, + channel_mapping=old_context.channel_mapping if channel_mapping is None else channel_mapping, + transformation=old_context.transformation if global_transformation is None else global_transformation, + ) + old_settings = self._build_settings_stack[-1] + settings = BuildSettings( + to_single_waveform=old_settings.to_single_waveform if to_single_waveform is None else to_single_waveform, + ) + + self._build_context_stack.append(context) + self._build_settings_stack.append(settings) + + @contextmanager + def _with_patched_context(self, **kwargs): + context = copy.copy(self._build_context_stack[-1]) + for name, value in kwargs.items(): + setattr(context, name, value) + self._build_context_stack.append(context) + yield + self._build_context_stack.pop() + + @contextmanager + def with_metadata(self, metadata: TemplateMetadata): + # metadata.to_single_waveform == "always" is handled in PulseTemplate._build_program + if metadata.minimal_sample_rate is not None: + with self._with_patched_context(minimal_sample_rate=metadata.minimal_sample_rate) as builder: + yield builder + else: + yield self + + @contextmanager + def with_transformation(self, transformation: Transformation): + context = copy.copy(self.build_context) + context.transformation = chain_transformations(context.transformation, transformation) + self._build_context_stack.append(context) + yield self + self._build_context_stack.pop() + + @contextmanager + def with_mappings(self, *, + parameter_mapping: Mapping[str, Expression], + measurement_mapping: Mapping[str, Optional[str]], + channel_mapping: Mapping[ChannelID, Optional[ChannelID]], + ): + context = self.build_context.apply_mappings(parameter_mapping, measurement_mapping, channel_mapping) + self._build_context_stack.append(context) + yield self + self._build_context_stack.pop() + + @abstractmethod + def _transformed_hold_voltage(self, duration: HardwareTime, voltages: Mapping[str, HardwareVoltage]): + """This internal function gets the constant voltage values transformed by the current built context's transformation. + """ + + @abstractmethod + def _transformed_play_arbitrary_waveform(self, waveform: Waveform): + """This internal function gets the waveform transformed by the current built context's transformation.""" + + def play_arbitrary_waveform(self, waveform: Waveform): + transformation = self.build_context.transformation + if transformation: + transformed_waveform = TransformingWaveform(waveform, transformation) + self._transformed_play_arbitrary_waveform(transformed_waveform) + else: + self._transformed_play_arbitrary_waveform(waveform) + + def hold_voltage(self, duration: HardwareTime, voltages: Mapping[str, HardwareVoltage]): + transformation = self.build_context.transformation + if transformation: + if transformation.get_constant_output_channels(voltages.keys()) != transformation.get_output_channels(voltages.keys()): + waveform = TransformingWaveform(ConstantWaveform.from_mapping(duration, voltages), transformation) + self._transformed_play_arbitrary_waveform(waveform) + else: + transformed_voltages = transformation(0.0, voltages) + self._transformed_hold_voltage(duration, transformed_voltages) + else: + self._transformed_hold_voltage(duration, voltages) diff --git a/qupulse/program/transformation.py b/qupulse/program/transformation.py index 607c8dc4..27a2e755 100644 --- a/qupulse/program/transformation.py +++ b/qupulse/program/transformation.py @@ -107,6 +107,9 @@ def is_constant_invariant(self): def get_constant_output_channels(self, input_channels: AbstractSet[ChannelID]) -> AbstractSet[ChannelID]: return input_channels + def __bool__(self): + return False + class ChainedTransformation(Transformation): __slots__ = ('_transformations',) diff --git a/qupulse/program/waveforms.py b/qupulse/program/waveforms.py index 2a41ab91..627bec5e 100644 --- a/qupulse/program/waveforms.py +++ b/qupulse/program/waveforms.py @@ -11,6 +11,7 @@ import itertools import operator import warnings +import dataclasses from abc import ABCMeta, abstractmethod from numbers import Real from typing import ( @@ -53,13 +54,41 @@ def _to_time_type(duration: Real) -> TimeType: return time_from_float(float(duration), absolute_error=PULSE_TO_WAVEFORM_ERROR) +@dataclasses.dataclass(frozen=False, eq=False, repr=False) +class WaveformMetadata: + """Metadata for a waveform. Does not participate in equality and hashing!""" + + minimal_sample_rate: Optional[TimeType] = None + + def __init__(self, **kwargs): + for key, value in kwargs.items(): + setattr(self, key, value) + + def as_dict(self): + data = vars(self).copy() + for field in dataclasses.fields(self): + if field.default is not dataclasses.MISSING: + if data[field.name] == field.default: + del data[field.name] + return data + + + def __repr__(self): + args = ",".join(f"{name}={value!r}" + for name, value in self.as_dict().items()) + return f'{self.__class__.__name__}({args})' + + class Waveform(metaclass=ABCMeta): """Represents an instantiated PulseTemplate which can be sampled to retrieve arrays of voltage values for the hardware.""" __sampled_cache = WeakValueDictionary() - __slots__ = ('_duration',) + __slots__ = ( + '_duration', # included in __hash__ and __eq__ + '_metadata', # excluded from __hash__ and __eq__ + ) def __init__(self, duration: TimeType): self._duration = duration @@ -69,6 +98,14 @@ def duration(self) -> TimeType: """The duration of the waveform in time units.""" return self._duration + @property + def metadata(self): + try: + return self._metadata + except AttributeError: + metadata = self._metadata = WaveformMetadata() + return metadata + @abstractmethod def unsafe_sample(self, channel: ChannelID, @@ -146,15 +183,23 @@ def get_sampled(self, def __hash__(self): if self.__class__.__base__ is not Waveform: + # we require direct inheritance because self.__slots__ are the slots of the subclass + # we manually add self._duration but not self._metadata here raise NotImplementedError("Waveforms __hash__ and __eq__ implementation requires direct inheritance") return hash(tuple(getattr(self, slot) for slot in self.__slots__)) ^ hash(self._duration) def __eq__(self, other): + if self.__class__.__base__ is not Waveform: + # we require direct inheritance because self.__slots__ are the slots of the subclass + # we manually add self._duration but not self._metadata here + raise NotImplementedError("Waveforms __hash__ and __eq__ implementation requires direct inheritance") slots = self.__slots__ if slots is getattr(other, '__slots__', None): + # the __slots__ attribute of self and other are identical objects -> other is of the same type return self._duration == other._duration and all(getattr(self, slot) == getattr(other, slot) for slot in slots) - # The other class might be more lenient - return NotImplemented + else: + # The other class might be more lenient + return NotImplemented @property @abstractmethod diff --git a/qupulse/pulses/abstract_pulse_template.py b/qupulse/pulses/abstract_pulse_template.py index 0ac3c1e8..54e21ae0 100644 --- a/qupulse/pulses/abstract_pulse_template.py +++ b/qupulse/pulses/abstract_pulse_template.py @@ -137,10 +137,8 @@ def _forward_if_linked(self, method_name: str, *args, **kwargs) -> Any: 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_program = partialmethod(_forward_if_linked, '_build_program') defined_channels = property(partial(_get_property, property_name='defined_channels'), doc=_PROPERTY_DOC.format(name='defined_channels')) diff --git a/qupulse/pulses/arithmetic_pulse_template.py b/qupulse/pulses/arithmetic_pulse_template.py index cba0d9a6..0ac2fce8 100644 --- a/qupulse/pulses/arithmetic_pulse_template.py +++ b/qupulse/pulses/arithmetic_pulse_template.py @@ -10,6 +10,7 @@ import sympy from qupulse.expressions import ExpressionScalar, ExpressionLike +from qupulse.program import ProgramBuilder from qupulse.serialization import Serializer, PulseRegistryType from qupulse.parameter_scope import Scope @@ -381,29 +382,17 @@ def _get_transformation(self, ScalingTransformation(scalar_value) ) - def _internal_create_program(self, *, - scope: Scope, - measurement_mapping: Dict[str, Optional[str]], - channel_mapping: Dict[ChannelID, Optional[ChannelID]], - global_transformation: Optional[Transformation], - to_single_waveform: Set[Union[str, 'PulseTemplate']], - program_builder: 'ProgramBuilder'): - """The operation is applied by modifying the transformation the pulse template operand sees.""" + def _internal_build_program(self, program_builder: ProgramBuilder): + build_context = program_builder.build_context + scope = build_context.scope if not scope.get_volatile_parameters().keys().isdisjoint(self._scalar_operand_parameters): raise NotImplementedError('The scalar operand of arithmetic pulse template cannot be volatile') - # put arithmetic into transformation inner_transformation = self._get_transformation(parameters=scope, - channel_mapping=channel_mapping) - - transformation = inner_transformation.chain(global_transformation) + channel_mapping=build_context.channel_mapping) + with program_builder.with_transformation(inner_transformation): + self._pulse_template._build_program(program_builder=program_builder) - self._pulse_template._create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - global_transformation=transformation, - to_single_waveform=to_single_waveform, - program_builder=program_builder) def build_waveform(self, parameters: Dict[str, Real], diff --git a/qupulse/pulses/loop_pulse_template.py b/qupulse/pulses/loop_pulse_template.py index 86b5589c..15fde39c 100644 --- a/qupulse/pulses/loop_pulse_template.py +++ b/qupulse/pulses/loop_pulse_template.py @@ -170,28 +170,15 @@ def _body_scope_generator(self, scope: Scope, forward=True) -> Iterator[Scope]: for loop_index_value in loop_range: yield _ForLoopScope(scope, loop_index_name, loop_index_value) - def _internal_create_program(self, *, - scope: Scope, - measurement_mapping: Dict[str, Optional[str]], - channel_mapping: Dict[ChannelID, Optional[ChannelID]], - global_transformation: Optional['Transformation'], - to_single_waveform: Set[Union[str, 'PulseTemplate']], - program_builder: ProgramBuilder) -> None: - self.validate_scope(scope=scope) - + def _internal_build_program(self, program_builder: ProgramBuilder): + build_context = program_builder.build_context + scope = build_context.scope loop_range = self._loop_range.to_range(scope) loop_index_name = self._loop_index - measurements = self.get_measurement_windows(scope, measurement_mapping) - - for iteration_program_builder in program_builder.with_iteration(loop_index_name, loop_range, - measurements=measurements): - self.body._create_program(scope=iteration_program_builder.inner_scope(scope), - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - global_transformation=global_transformation, - to_single_waveform=to_single_waveform, - program_builder=iteration_program_builder) + measurements = self.get_measurement_windows(scope, build_context.measurement_mapping) + for iteration_program_builder in program_builder.with_iteration(loop_index_name, loop_range, measurements=measurements): + self.body._build_program(program_builder=iteration_program_builder) def build_waveform(self, parameter_scope: Scope) -> ForLoopWaveform: return ForLoopWaveform([self.body.build_waveform(local_scope) diff --git a/qupulse/pulses/mapping_pulse_template.py b/qupulse/pulses/mapping_pulse_template.py index d120c602..af768f59 100644 --- a/qupulse/pulses/mapping_pulse_template.py +++ b/qupulse/pulses/mapping_pulse_template.py @@ -347,22 +347,12 @@ def get_updated_channel_mapping(self, channel_mapping: Dict[ChannelID, return {inner_ch: None if outer_ch is None else channel_mapping[outer_ch] for inner_ch, outer_ch in self.__channel_mapping.items()} - def _internal_create_program(self, *, - scope: Scope, - measurement_mapping: Dict[str, Optional[str]], - channel_mapping: Dict[ChannelID, Optional[ChannelID]], - global_transformation: Optional['Transformation'], - to_single_waveform: Set[Union[str, 'PulseTemplate']], - program_builder: ProgramBuilder) -> None: - self.validate_scope(scope) - - # parameters are validated in map_parameters() call, no need to do it here again explicitly - self.template._create_program(scope=self.map_scope(scope), - measurement_mapping=self.get_updated_measurement_mapping(measurement_mapping), - channel_mapping=self.get_updated_channel_mapping(channel_mapping), - global_transformation=global_transformation, - to_single_waveform=to_single_waveform, - program_builder=program_builder) + def _internal_build_program(self, program_builder: ProgramBuilder): + with program_builder.with_mappings( + parameter_mapping=self.__parameter_mapping, + channel_mapping=self.__channel_mapping, + measurement_mapping=self.__measurement_mapping) as inner_builder: + self.__template._build_program(program_builder=inner_builder) def build_waveform(self, parameters: Dict[str, numbers.Real], diff --git a/qupulse/pulses/measurement.py b/qupulse/pulses/measurement.py index 76145e35..4e545aa4 100644 --- a/qupulse/pulses/measurement.py +++ b/qupulse/pulses/measurement.py @@ -35,7 +35,7 @@ def __init__(self, measurements: Optional[List[MeasurementDeclaration]]): def get_measurement_windows(self, parameters: Union[Mapping[str, Real], Scope], - measurement_mapping: Dict[str, Optional[str]]) -> List[MeasurementWindow]: + measurement_mapping: Mapping[str, Optional[str]]) -> List[MeasurementWindow]: """Calculate measurement windows with the given parameter set and rename them with the measurement mapping. This method only returns the measurement windows that are defined on `self`. It does _not_ collect the measurement windows defined on eventual child objects that `self` has/is composed of. diff --git a/qupulse/pulses/metadata.py b/qupulse/pulses/metadata.py index 4d38ea78..2454674e 100644 --- a/qupulse/pulses/metadata.py +++ b/qupulse/pulses/metadata.py @@ -3,6 +3,7 @@ from typing import Literal, Optional +from qupulse.utils.types import TimeType SingleWaveformStrategy = Literal['always'] @@ -17,11 +18,16 @@ class TemplateMetadata: """ to_single_waveform: Optional[SingleWaveformStrategy] = dataclasses.field(default=None) + minimal_sample_rate: Optional[TimeType] = dataclasses.field(default=None) - def __init__(self, to_single_waveform: Optional[SingleWaveformStrategy] = None, **kwargs): + def __init__(self, + to_single_waveform: Optional[SingleWaveformStrategy] = None, + minimal_sample_rate: Optional[TimeType] = None, + **kwargs): # TODO: generate this init automatically # The reason for the custom init is that we want to allow additional kwargs self.to_single_waveform = to_single_waveform + self.minimal_sample_rate = minimal_sample_rate for key, value in kwargs.items(): setattr(self, key, value) diff --git a/qupulse/pulses/multi_channel_pulse_template.py b/qupulse/pulses/multi_channel_pulse_template.py index cf79bb18..46c19e0b 100644 --- a/qupulse/pulses/multi_channel_pulse_template.py +++ b/qupulse/pulses/multi_channel_pulse_template.py @@ -15,6 +15,7 @@ import numbers import warnings +from qupulse.program import ProgramBuilder from qupulse.serialization import Serializer, PulseRegistryType from qupulse.parameter_scope import Scope @@ -269,21 +270,13 @@ def _get_overwritten_channels_values(self, for name, value in self.overwritten_channels.items() if channel_mapping[name] is not None} - def _internal_create_program(self, *, - scope: Scope, - global_transformation: Optional[Transformation], - channel_mapping: Dict[ChannelID, Optional[ChannelID]], - **kwargs): - overwritten_channels = self._get_overwritten_channels_values(parameters=scope, channel_mapping=channel_mapping) + def _internal_build_program(self, program_builder: ProgramBuilder): + context = program_builder.build_context + overwritten_channels = self._get_overwritten_channels_values(parameters=context.scope, + channel_mapping=context.channel_mapping) transformation = ParallelChannelTransformation(overwritten_channels) - - if global_transformation is not None: - transformation = chain_transformations(global_transformation, transformation) - - self._template._create_program(scope=scope, - channel_mapping=channel_mapping, - global_transformation=transformation, - **kwargs) + with program_builder.with_transformation(transformation) as trafo_program_builder: + self._template._build_program(program_builder=trafo_program_builder) def build_waveform(self, parameters: Dict[str, numbers.Real], channel_mapping: Dict[ChannelID, Optional[ChannelID]]) -> Optional[Waveform]: diff --git a/qupulse/pulses/pulse_template.py b/qupulse/pulses/pulse_template.py index c7c30097..9c2b0bc3 100644 --- a/qupulse/pulses/pulse_template.py +++ b/qupulse/pulses/pulse_template.py @@ -26,7 +26,7 @@ from qupulse.expressions import ExpressionScalar, Expression, ExpressionLike from qupulse.program.transformation import Transformation -from qupulse.program.waveforms import Waveform, TransformingWaveform +from qupulse.program.waveforms import Waveform, TransformingWaveform, WaveformMetadata from qupulse.pulses.measurement import MeasurementDefiner, MeasurementDeclaration from qupulse.parameter_scope import Scope, DictScope @@ -237,37 +237,42 @@ def create_program(self, *, category=UnknownVolatileParameter, stacklevel=2) + program_builder.override( + scope=scope, + measurement_mapping=measurement_mapping, + channel_mapping=complete_channel_mapping, + global_transformation=global_transformation, + to_single_waveform=to_single_waveform, + ) + # call subclass specific implementation - self._create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=complete_channel_mapping, - global_transformation=global_transformation, - to_single_waveform=to_single_waveform, - program_builder=program_builder) + self._build_program(program_builder=program_builder) + return program_builder.to_program() - return program_builder.to_program(set(complete_channel_mapping.values())) + def _build_program(self, program_builder: ProgramBuilder): + """New method that keeps state in program builder""" + if (validate_scope := getattr(self, "validate_scope", None)) is not None: + validate_scope(program_builder.build_context.scope) - @abstractmethod - def _internal_create_program(self, *, - scope: Scope, - measurement_mapping: Dict[str, Optional[str]], - channel_mapping: Dict[ChannelID, Optional[ChannelID]], - global_transformation: Optional[Transformation], - to_single_waveform: Set[Union[str, 'PulseTemplate']], - program_builder: ProgramBuilder) -> None: - """The subclass specific implementation of create_program(). - - Receives a Loop instance parent_loop to which it should append measurements and its own Loops as children. - - Subclasses should not overwrite create_program() directly but provide their implementation here. This method - is called by create_program(). - Implementations should not call create_program() of any subtemplates to obtain Loop objects for them but - call subtemplate._internal_create_program() instead, providing an adequate parent_loop object to which - the subtemplate will append. Implementations must make sure not to append invalid Loop objects (no waveform or no children). - - In case of an error (e.g. invalid measurement mapping, missing parameters, violated parameter constraints, etc), - implementations of this method must throw an adequate exception. They do not have to ensure that the parent_loop - remains unchanged in this case.""" + to_single_waveform = program_builder.build_settings.to_single_waveform + if self.metadata.to_single_waveform == 'always' or self.identifier in to_single_waveform or self in to_single_waveform: + with program_builder.new_subprogram() as inner_program_builder: + self._internal_build_program(inner_program_builder) + else: + self._internal_build_program(program_builder) + + def _internal_build_program(self, program_builder: ProgramBuilder): + """The subclass specific implementation of create_program().""" + context = program_builder.build_context + settings = program_builder.build_settings + self._internal_create_program( + scope=context.scope, + measurement_mapping=dict(context.measurement_mapping), + channel_mapping=dict(context.channel_mapping), + global_transformation=context.transformation, + to_single_waveform=set(settings.to_single_waveform), + program_builder=program_builder + ) def _create_program(self, *, scope: Scope, @@ -278,6 +283,9 @@ def _create_program(self, *, program_builder: ProgramBuilder): """Generic part of create program. This method handles to_single_waveform and the configuration of the transformer.""" + warnings.warn("_create_program is deprecated. " + "Update your driver to use _build_program", + DeprecationWarning, stacklevel=2) if self.metadata.to_single_waveform == 'always' or self.identifier in to_single_waveform or self in to_single_waveform: with program_builder.new_subprogram(global_transformation=global_transformation) as inner_program_builder: @@ -689,35 +697,24 @@ def _is_atomic(self) -> bool: measurement_names = MeasurementDefiner.measurement_names - def _internal_create_program(self, *, - scope: Scope, - measurement_mapping: Dict[str, Optional[str]], - channel_mapping: Dict[ChannelID, Optional[ChannelID]], - global_transformation: Optional[Transformation], - to_single_waveform: Set[Union[str, 'PulseTemplate']], - program_builder: ProgramBuilder) -> None: - """Parameter constraints are validated in build_waveform because build_waveform is guaranteed to be called - during sequencing""" - ### current behavior (same as previously): only adds EXEC Loop and measurements if a waveform exists. - ### measurements are directly added to parent_loop (to reflect behavior of Sequencer + MultiChannelProgram) + def _internal_build_program(self, program_builder: ProgramBuilder): + context = program_builder.build_context + scope = context.scope assert not scope.get_volatile_parameters().keys() & self.parameter_names, "AtomicPT cannot be volatile" waveform = self.build_waveform(parameters=scope, - channel_mapping=channel_mapping) - if waveform: - measurements = self.get_measurement_windows(parameters=scope, - measurement_mapping=measurement_mapping) - program_builder.measure(measurements) - - if global_transformation: - waveform = TransformingWaveform.from_transformation(waveform, global_transformation) - - constant_values = waveform.constant_value_dict() - if constant_values is None: - program_builder.play_arbitrary_waveform(waveform) - else: - program_builder.hold_voltage(waveform.duration, constant_values) - + channel_mapping=context.channel_mapping) + if not waveform: + return + + measurements = self.get_measurement_windows(parameters=scope, + measurement_mapping=context.measurement_mapping) + program_builder.measure(measurements) + constant_values = waveform.constant_value_dict() + if constant_values is None: + program_builder.play_arbitrary_waveform(waveform) + else: + program_builder.hold_voltage(waveform.duration, constant_values) @abstractmethod def build_waveform(self, diff --git a/qupulse/pulses/repetition_pulse_template.py b/qupulse/pulses/repetition_pulse_template.py index 6c0bd2e8..fd4f0eca 100644 --- a/qupulse/pulses/repetition_pulse_template.py +++ b/qupulse/pulses/repetition_pulse_template.py @@ -132,34 +132,23 @@ def measurement_names(self) -> AbstractSet[str]: def duration(self) -> ExpressionScalar: return self.repetition_count * self.body.duration - def _internal_create_program(self, *, - scope: Scope, - measurement_mapping: Dict[str, Optional[str]], - channel_mapping: Dict[ChannelID, Optional[ChannelID]], - global_transformation: Optional['Transformation'], - to_single_waveform: AbstractSet[Union[str, 'PulseTemplate']], - program_builder: ProgramBuilder) -> None: - self.validate_scope(scope) - - repetition_count = max(0, self.get_repetition_count_value(scope)) - - if repetition_count > 0: - if scope.get_volatile_parameters().keys() & self.repetition_count.variables: - repetition_definition = VolatileRepetitionCount(self.repetition_count, scope) - assert int(repetition_definition) == repetition_count - else: - repetition_definition = repetition_count - - measurements = self.get_measurement_windows(scope, measurement_mapping) - - for repetition_program_builder in program_builder.with_repetition(repetition_definition, - measurements=measurements): - self.body._create_program(scope=repetition_program_builder.inner_scope(scope), - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - global_transformation=global_transformation, - to_single_waveform=to_single_waveform, - program_builder=repetition_program_builder) + def _internal_build_program(self, program_builder: ProgramBuilder): + build_context = program_builder.build_context + scope = build_context.scope + + repetition_count = self.get_repetition_count_value(scope) + if not (repetition_count > 0): + return + + if scope.get_volatile_parameters().keys() & self.repetition_count.variables: + repetition_definition = VolatileRepetitionCount(self.repetition_count, scope) + assert int(repetition_definition) == repetition_count + else: + repetition_definition = repetition_count + + measurements = self.get_measurement_windows(scope, build_context.measurement_mapping) + for repetition_program_builder in program_builder.with_repetition(repetition_definition, measurements=measurements): + self.body._build_program(program_builder=repetition_program_builder) def get_serialization_data(self, serializer: Optional[Serializer]=None) -> Dict[str, Any]: data = super().get_serialization_data(serializer) diff --git a/qupulse/pulses/sequence_pulse_template.py b/qupulse/pulses/sequence_pulse_template.py index 07fe07df..c1fe64ba 100644 --- a/qupulse/pulses/sequence_pulse_template.py +++ b/qupulse/pulses/sequence_pulse_template.py @@ -134,24 +134,12 @@ def build_waveform(self, [sub_template.build_waveform(parameters, channel_mapping=channel_mapping) for sub_template in self.__subtemplates]) - def _internal_create_program(self, *, - scope: Scope, - measurement_mapping: Dict[str, Optional[str]], - channel_mapping: Dict[ChannelID, Optional[ChannelID]], - global_transformation: Optional['Transformation'], - to_single_waveform: Set[Union[str, 'PulseTemplate']], - program_builder: ProgramBuilder) -> None: - self.validate_scope(scope) - - measurements = self.get_measurement_windows(scope, measurement_mapping) + def _internal_build_program(self, program_builder: ProgramBuilder): + build_context = program_builder.build_context + measurements = self.get_measurement_windows(build_context.scope, build_context.measurement_mapping) with program_builder.with_sequence(measurements=measurements) as sequence_program_builder: for subtemplate in self.subtemplates: - subtemplate._create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - global_transformation=global_transformation, - to_single_waveform=to_single_waveform, - program_builder=sequence_program_builder) + subtemplate._build_program(program_builder=sequence_program_builder) def get_serialization_data(self, serializer: Optional[Serializer]=None) -> Dict[str, Any]: data = super().get_serialization_data(serializer) diff --git a/qupulse/pulses/time_reversal_pulse_template.py b/qupulse/pulses/time_reversal_pulse_template.py index 12377f2c..e0b9376c 100644 --- a/qupulse/pulses/time_reversal_pulse_template.py +++ b/qupulse/pulses/time_reversal_pulse_template.py @@ -57,10 +57,10 @@ def initial_values(self) -> Dict[ChannelID, ExpressionScalar]: @property def final_values(self) -> Dict[ChannelID, ExpressionScalar]: return self._inner.initial_values - - def _internal_create_program(self, *, program_builder: ProgramBuilder, **kwargs) -> None: - with program_builder.time_reversed() as reversed_builder: - self._inner._internal_create_program(program_builder=reversed_builder, **kwargs) + + def _internal_build_program(self, program_builder: ProgramBuilder): + with program_builder.time_reversed() as inner_program_builder: + self._inner._internal_build_program(inner_program_builder) def build_waveform(self, *args, **kwargs) -> Optional[Waveform]: diff --git a/tests/program/linspace_tests.py b/tests/program/linspace_tests.py index 77138f57..deab804a 100644 --- a/tests/program/linspace_tests.py +++ b/tests/program/linspace_tests.py @@ -23,7 +23,7 @@ def setUp(self): hold = ConstantPT(10 ** 6, {'a': '-1. + idx * 0.01'}) self.pulse_template = hold.with_iteration('idx', 200) - self.program = LinSpaceTopLevel((LinSpaceIter( + self.program = LinSpaceProgram((LinSpaceIter( length=200, body=(LinSpaceHold( bases=(-1.,), @@ -31,8 +31,7 @@ def setUp(self): duration_base=TimeType(10**6), duration_factors=None ),) - ),), - {'a',}) + ),), ("a",)) key = DepKey.from_voltages((0.01,), DEFAULT_INCREMENT_RESOLUTION) @@ -112,7 +111,7 @@ def setUp(self): duration_factors=None ) - self.program = LinSpaceTopLevel((LinSpaceIter( + self.program = LinSpaceProgram((LinSpaceIter( length=rep_factor, body=( dependent_hold_1, @@ -120,7 +119,7 @@ def setUp(self): LinSpaceIter(body=(wait_hold,), length=rep_factor), ) ),), - {'a','b'}) + ('a','b')) self.commands = [ Set(channel=0, value=-1.0, key=DepKey(factors=())), @@ -209,7 +208,7 @@ def setUp(self): # self.pulse_template = (hold_random@(hold_random@hold).with_repetition(10)@hold_random@hold)\ self.pulse_template = (hold_random @ hold.with_repetition(10)).with_iteration('idx', 200) - self.program = LinSpaceTopLevel((LinSpaceIter( + self.program = LinSpaceProgram((LinSpaceIter( length=200, body=( LinSpaceHold(bases=(-.4,), factors=(None,), duration_base=TimeType(10**5), duration_factors=None), @@ -218,7 +217,9 @@ def setUp(self): ), count=10), # LinSpaceHold(bases=(-.4),factors=None,duration_base=TimeType(10**6),duration_factors=None), # LinSpaceHold(bases=(-1.,),factors=((0.01,),),duration_base=TimeType(10**6),duration_factors=None) - ),),),{'a',}) + ),),), + ('a',) + ) self.commands = [ Set(channel=0, value=-0.4, key=DepKey(factors=())), @@ -270,7 +271,7 @@ def setUp(self): scan_a = hold.with_iteration('idx_a', 200) self.pulse_template = scan_a.with_iteration('idx_b', 100) - self.program = LinSpaceTopLevel((LinSpaceIter(length=100, body=(LinSpaceIter( + self.program = LinSpaceProgram((LinSpaceIter(length=100, body=(LinSpaceIter( length=200, body=(LinSpaceHold( bases=(-1., -0.5), @@ -279,7 +280,7 @@ def setUp(self): duration_base=TimeType(10**6), duration_factors=None ),) - ),)),),{'a','b'}) + ),)),), ('a', 'b')) key_0 = DepKey.from_voltages((0, 0.01,), DEFAULT_INCREMENT_RESOLUTION) key_1 = DepKey.from_voltages((0.02,), DEFAULT_INCREMENT_RESOLUTION) @@ -342,7 +343,7 @@ def setUp(self): self.pulse_template = scan_a.with_iteration('idx_b', 100) self.repeated_pt = self.pulse_template.with_repetition(repetition_count) - self.program = LinSpaceTopLevel((LinSpaceIter(length=100, body=(LinSpaceIter( + self.program = LinSpaceProgram((LinSpaceIter(length=100, body=(LinSpaceIter( length=200, body=(LinSpaceHold( bases=(-1., -0.5), @@ -351,8 +352,8 @@ def setUp(self): duration_base=TimeType(10**6), duration_factors=None ),) - ),)),),{'a','b'}) - self.repeated_program = LinSpaceTopLevel((LinSpaceRepeat(body=self.program.body, count=repetition_count),),{'a','b'}) + ),)),), ('a', 'b')) + self.repeated_program = LinSpaceProgram((LinSpaceRepeat(body=self.program.root, count=repetition_count),), ('a', 'b')) key_0 = DepKey.from_voltages((1e-3, 0.01,), DEFAULT_INCREMENT_RESOLUTION) key_1 = DepKey.from_voltages((0.02, -3e-3), DEFAULT_INCREMENT_RESOLUTION) @@ -440,7 +441,7 @@ def setUp(self): singlet_scan = (load_random @ wait @ meas).with_iteration('idx_a', 200).with_iteration('idx_b', 100) self.pulse_template = singlet_scan - self.program = LinSpaceTopLevel((LinSpaceIter(length=100, body=(LinSpaceIter( + self.program = LinSpaceProgram((LinSpaceIter(length=100, body=(LinSpaceIter( length=200, body=( LinSpaceHold(bases=(-0.4, -0.3), factors=(None, None), duration_base=TimeType(10 ** 5), @@ -453,7 +454,7 @@ def setUp(self): LinSpaceHold(bases=(0.05, 0.06), factors=(None, None), duration_base=TimeType(10 ** 5), duration_factors=None), ) - ),)),),{'a','b'}) + ),)),), ('a', 'b')) key_0 = DepKey.from_voltages((0, 0.01,), DEFAULT_INCREMENT_RESOLUTION) key_1 = DepKey.from_voltages((0.02,), DEFAULT_INCREMENT_RESOLUTION) @@ -549,7 +550,7 @@ def setUp(self): self.pulse_template = hold.with_iteration('idx', 200) self.transformation = ScalingTransformation({'a': 2.0}) - self.program = LinSpaceTopLevel((LinSpaceIter( + self.program = LinSpaceProgram((LinSpaceIter( length=200, body=(LinSpaceHold( bases=(-2.,), @@ -557,7 +558,7 @@ def setUp(self): duration_base=TimeType(10 ** 6), duration_factors=None ),) - ),),{'a',}) + ),), ('a',)) def test_global_trafo_program(self): program_builder = LinSpaceBuilder(('a',)) @@ -584,7 +585,7 @@ def setUp(self): self.sine_waveform = sine.build_waveform(parameters={}, channel_mapping={'a': 'a'}) - self.program = LinSpaceTopLevel((LinSpaceIter( + self.program = LinSpaceProgram((LinSpaceIter( length=100, body=(LinSpaceHold( bases=(-1.,), @@ -597,7 +598,7 @@ def setUp(self): channels=('a',) ) ) - ),),{'a',}) + ),),('a',)) key = DepKey.from_voltages((0.01,), DEFAULT_INCREMENT_RESOLUTION) self.commands = [ diff --git a/tests/pulses/arithmetic_pulse_template_tests.py b/tests/pulses/arithmetic_pulse_template_tests.py index 919fdc7d..8c9aac2d 100644 --- a/tests/pulses/arithmetic_pulse_template_tests.py +++ b/tests/pulses/arithmetic_pulse_template_tests.py @@ -423,17 +423,18 @@ def test_time_dependent_integral(self): evaluated = symbolic.evaluate_in_scope({'t_gauss': t_gauss}) np.testing.assert_allclose(expected, evaluated) - def test_internal_create_program(self): + def test_internal_build_program(self): lhs = 'x + y' rhs = DummyPulseTemplate(defined_channels={'u', 'v', 'w'}) arith = ArithmeticPulseTemplate(lhs, '-', rhs) scope = DictScope.from_kwargs(x=3, y=5, z=8, volatile={'some_parameter'}) channel_mapping = dict(u='a', v='b', w=None) - measurement_mapping = dict(m1='m2') - global_transformation = OffsetTransformation({'unrelated': 1.}) - to_single_waveform = {'something_else'} - program_builder = mock.Mock() + program_builder = mock.MagicMock() + + build_context = program_builder.build_context + build_context.scope = scope + build_context.channel_mapping = channel_mapping expected_transformation = mock.create_autospec(IdentityTransformation,instance=True) with self.assertWarns(DeprecationWarning): @@ -442,35 +443,20 @@ def test_internal_create_program(self): inner_trafo = mock.create_autospec(spec=IdentityTransformation,instance=True) inner_trafo.chain.return_value = expected_transformation - with mock.patch.object(rhs, '_create_program') as inner_create_program: + with mock.patch.object(rhs, '_build_program') as inner_build_program: with mock.patch.object(arith, '_get_transformation', return_value=inner_trafo) as get_transformation: - arith._internal_create_program( - scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - global_transformation=global_transformation, - to_single_waveform=to_single_waveform, + arith._internal_build_program( program_builder=program_builder ) get_transformation.assert_called_once_with(parameters=scope, channel_mapping=channel_mapping) - - inner_trafo.chain.assert_called_once_with(global_transformation) - inner_create_program.assert_called_once_with( - scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - global_transformation=expected_transformation, - to_single_waveform=to_single_waveform, + program_builder.with_transformation.assert_called_once_with(inner_trafo) + inner_build_program.assert_called_once_with( program_builder=program_builder ) + build_context.scope = DictScope.from_kwargs(x=3, y=5, z=8, volatile={'x'}) with self.assertRaisesRegex(NotImplementedError, 'volatile'): - arith._internal_create_program( - scope=DictScope.from_kwargs(x=3, y=5, z=8, volatile={'x'}), - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - global_transformation=global_transformation, - to_single_waveform=to_single_waveform, + arith._internal_build_program( program_builder=program_builder ) diff --git a/tests/pulses/constant_pulse_template_tests.py b/tests/pulses/constant_pulse_template_tests.py index 197aa101..df33a527 100644 --- a/tests/pulses/constant_pulse_template_tests.py +++ b/tests/pulses/constant_pulse_template_tests.py @@ -1,4 +1,5 @@ import unittest +from unittest import mock import qupulse.plotting import qupulse.program.waveforms @@ -44,10 +45,7 @@ def test_zero_duration(self): self.assertEqual(pulse.duration, 12) def test_regression_duration_conversion(self): - old_value = qupulse.program.waveforms.PULSE_TO_WAVEFORM_ERROR - - try: - qupulse.program.waveforms.PULSE_TO_WAVEFORM_ERROR = 1e-6 + with mock.patch("qupulse.program.waveforms.PULSE_TO_WAVEFORM_ERROR", 1e-6): for duration_in_samples in [64, 936320, 24615392]: p = ConstantPulseTemplate(duration_in_samples / 2.4, {'a': 0}) number_of_samples = p.create_program().duration * 2.4 @@ -56,33 +54,21 @@ def test_regression_duration_conversion(self): p2 = ConstantPulseTemplate((duration_in_samples + 1) / 2.4, {'a': 0}) self.assertNotEqual(p.create_program().duration, p2.create_program().duration) - finally: - qupulse.program.waveforms.PULSE_TO_WAVEFORM_ERROR = old_value def test_regression_duration_conversion_functionpt(self): - old_value = qupulse.program.waveforms.PULSE_TO_WAVEFORM_ERROR - - try: - qupulse.program.waveforms.PULSE_TO_WAVEFORM_ERROR = 1e-6 + with mock.patch("qupulse.program.waveforms.PULSE_TO_WAVEFORM_ERROR", 1e-6): for duration_in_samples in [64, 2000, 936320]: p = FunctionPT('1', duration_expression=duration_in_samples / 2.4, channel='a') number_of_samples = p.create_program().duration * 2.4 self.assertEqual(number_of_samples.denominator, 1) - finally: - qupulse.program.waveforms.PULSE_TO_WAVEFORM_ERROR = old_value def test_regression_template_combination(self): - old_value = qupulse.utils.sympy.SYMPY_DURATION_ERROR_MARGIN - - try: - qupulse.utils.sympy.SYMPY_DURATION_ERROR_MARGIN = 1e-9 + with mock.patch("qupulse.utils.sympy.SYMPY_DURATION_ERROR_MARGIN", 1e-9): duration_in_seconds = 2e-6 full_template = ConstantPulseTemplate(duration=duration_in_seconds * 1e9, amplitude_dict={'C1': 1.1}) duration_in_seconds_derived = 1e-9 * full_template.duration marker_pulse = TablePT({'marker': [(0, 0), (duration_in_seconds_derived * 1e9, 0)]}) full_template = AtomicMultiChannelPT(full_template, marker_pulse) - finally: - qupulse.utils.sympy.SYMPY_DURATION_ERROR_MARGIN = old_value def test_regression_sequencept_with_mappingpt(self): t1 = TablePT({'C1': [(0, 0), (100, 0)], 'C2': [(0, 1), (100, 1)]}) diff --git a/tests/pulses/loop_pulse_template_tests.py b/tests/pulses/loop_pulse_template_tests.py index cbeec61f..b74d45f1 100644 --- a/tests/pulses/loop_pulse_template_tests.py +++ b/tests/pulses/loop_pulse_template_tests.py @@ -211,14 +211,10 @@ def test_create_program_constraint_on_loop_var_exception(self): children = [Loop(waveform=DummyWaveform(duration=2.0))] program = Loop(children=children) program_builder = LoopBuilder._testing_dummy([program]) + program_builder.override(scope=scope) with self.assertRaises(ParameterNotProvidedException): - flt._internal_create_program(scope=scope, - measurement_mapping=dict(), - channel_mapping=dict(), - program_builder=program_builder, - to_single_waveform=set(), - global_transformation=None) + flt._build_program(program_builder=program_builder) self.assertEqual(children, list(program.children)) self.assertEqual(1, program.repetition_count) self.assertIsNone(program._measurements) @@ -237,14 +233,10 @@ def test_create_program_invalid_params(self) -> None: children = [Loop(waveform=DummyWaveform(duration=2.0))] program = Loop(children=children) program_builder = LoopBuilder._testing_dummy([program]) + program_builder.override(scope=invalid_scope, measurement_mapping=measurement_mapping, channel_mapping=channel_mapping) with self.assertRaises(ParameterConstraintViolation): - flt._internal_create_program(scope=invalid_scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - program_builder=program_builder, - to_single_waveform=set(), - global_transformation=None) + flt._build_program(program_builder=program_builder) self.assertEqual(children, list(program.children)) self.assertEqual(1, program.repetition_count) @@ -264,14 +256,10 @@ def test_create_program_invalid_measurement_mapping(self) -> None: children = [Loop(waveform=DummyWaveform(duration=2.0))] program = Loop(children=children) program_builder = LoopBuilder._testing_dummy([program]) + program_builder.override(scope=invalid_scope, measurement_mapping=measurement_mapping, channel_mapping=channel_mapping) with self.assertRaises(KeyError): - flt._internal_create_program(scope=invalid_scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - program_builder=program_builder, - to_single_waveform=set(), - global_transformation=None) + flt._build_program(program_builder=program_builder) self.assertEqual(children, list(program.children)) self.assertEqual(1, program.repetition_count) @@ -280,13 +268,9 @@ def test_create_program_invalid_measurement_mapping(self) -> None: # test for broken mapping on child level. no guarantee that parent_loop is not changed, only check for exception measurement_mapping = dict(A='B') + program_builder.override(measurement_mapping=measurement_mapping) with self.assertRaises(KeyError): - flt._internal_create_program(scope=invalid_scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - program_builder=program_builder, - to_single_waveform=set(), - global_transformation=None) + flt._build_program(program_builder=program_builder) def test_create_program_missing_params(self) -> None: dt = DummyPulseTemplate(parameter_names={'i'}, waveform=DummyWaveform(duration=4.0), measurements=[('b', 2, 1)]) @@ -300,32 +284,24 @@ def test_create_program_missing_params(self) -> None: children = [Loop(waveform=DummyWaveform(duration=2.0))] program = Loop(children=children) program_builder = LoopBuilder._testing_dummy([program]) + program_builder.override(scope=scope, measurement_mapping=measurement_mapping, channel_mapping=channel_mapping) # test parameter in constraints with self.assertRaises(ParameterNotProvidedException): - flt._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - program_builder=program_builder, - to_single_waveform=set(), - global_transformation=None) + flt._build_program(program_builder=program_builder) # test parameter in measurement mappings scope = DictScope.from_kwargs(a=1, b=4, c=2) + program_builder.override(scope=scope) with self.assertRaises(ParameterNotProvidedException): - flt._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - program_builder=program_builder, - to_single_waveform=set(), - global_transformation=None) + flt._build_program(program_builder=program_builder) self.assertEqual(children, list(program.children)) self.assertEqual(1, program.repetition_count) self.assertIsNone(program._measurements) self.assert_measurement_windows_equal({}, program.get_measurement_windows()) - def test_create_program_body_none(self) -> None: + def test_build_program_body_none(self) -> None: dt = DummyPulseTemplate(parameter_names={'i'}, waveform=None, duration=0, measurements=[('b', 2, 1)]) flt = ForLoopPulseTemplate(body=dt, loop_index='i', loop_range=('a', 'b', 'c'), @@ -337,17 +313,16 @@ def test_create_program_body_none(self) -> None: program_builder = LoopBuilder() - flt._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - program_builder=program_builder, - to_single_waveform=set(), - global_transformation=None) + program_builder.override( + scope=scope, measurement_mapping=measurement_mapping, channel_mapping=channel_mapping, + ) + + flt._internal_build_program(program_builder=program_builder) program = program_builder.to_program() self.assertIsNone(program) - def test_create_program(self) -> None: + def test_build_program(self) -> None: dt = DummyPulseTemplate(parameter_names={'i'}, waveform=DummyWaveform(duration=4.0, defined_channels={'A'}), duration=4, @@ -363,33 +338,20 @@ def test_create_program(self) -> None: global_transformation = TransformationStub() program_builder = LoopBuilder() + program_builder.override(scope=scope, measurement_mapping=measurement_mapping, channel_mapping=channel_mapping) # inner _create_program does nothing expected_program = Loop(measurements=[('B', .1, 1)]) - expected_create_program_kwargs = dict(measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - global_transformation=global_transformation, - to_single_waveform=to_single_waveform, - program_builder=program_builder) - expected_create_program_calls = [mock.call(**expected_create_program_kwargs, - scope=_ForLoopScope(scope, 'i', i)) + expected_create_program_calls = [mock.call(program_builder=program_builder) for i in (1, 3)] - with mock.patch.object(flt, 'validate_scope') as validate_scope: - with mock.patch.object(dt, '_create_program') as body_create_program: - with mock.patch.object(flt, 'get_measurement_windows', - wraps=flt.get_measurement_windows) as get_measurement_windows: - flt._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - program_builder=program_builder, - to_single_waveform=to_single_waveform, - global_transformation=global_transformation) - - validate_scope.assert_called_once_with(scope=scope) - get_measurement_windows.assert_called_once_with(scope, measurement_mapping) - self.assertEqual(body_create_program.call_args_list, expected_create_program_calls) + with mock.patch.object(dt, '_build_program') as body_create_program: + with mock.patch.object(flt, 'get_measurement_windows', + wraps=flt.get_measurement_windows) as get_measurement_windows: + flt._internal_build_program(program_builder=program_builder) + get_measurement_windows.assert_called_once_with(scope, measurement_mapping) + self.assertEqual(body_create_program.call_args_list, expected_create_program_calls) def test_create_program_append(self) -> None: dt = DummyPulseTemplate(parameter_names={'i'}, waveform=DummyWaveform(duration=4.0), duration=4, @@ -405,13 +367,9 @@ def test_create_program_append(self) -> None: program = Loop(children=children) program_builder = LoopBuilder._testing_dummy([program]) + program_builder.override(scope=scope, measurement_mapping=measurement_mapping, channel_mapping=channel_mapping) - flt._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - to_single_waveform=set(), - global_transformation=None, - program_builder=program_builder) + flt._internal_build_program(program_builder=program_builder) program_builder.to_program() diff --git a/tests/pulses/mapping_pulse_template_tests.py b/tests/pulses/mapping_pulse_template_tests.py index ae86f91f..0d40f52b 100644 --- a/tests/pulses/mapping_pulse_template_tests.py +++ b/tests/pulses/mapping_pulse_template_tests.py @@ -308,30 +308,17 @@ def test_create_program(self) -> None: parameter_names={'t'}) st = MappingPulseTemplate(template, parameter_mapping=parameter_mapping, measurement_mapping=measurement_mapping, channel_mapping=channel_mapping) - - pre_scope = DictScope.from_kwargs(k=5) - pre_measurement_mapping = {'meas2': 'meas3'} - pre_channel_mapping = {'default': 'A'} - - program_builder = default_program_builder() - expected_inner_args = dict(scope=st.map_scope(pre_scope), - measurement_mapping=st.get_updated_measurement_mapping(pre_measurement_mapping), - channel_mapping=st.get_updated_channel_mapping(pre_channel_mapping), - to_single_waveform=to_single_waveform, - global_transformation=global_transformation, - program_builder=program_builder) - - with mock.patch.object(template, '_create_program') as inner_create_program: - st._internal_create_program(scope=pre_scope, - measurement_mapping=pre_measurement_mapping, - channel_mapping=pre_channel_mapping, - to_single_waveform=to_single_waveform, - global_transformation=global_transformation, - program_builder=program_builder) - inner_create_program.assert_called_once_with(**expected_inner_args) - - # as we mock the inner function there shouldnt be any changes - self.assertIsNone(program_builder.to_program()) + program_builder = mock.Mock() + program_builder.with_mappings = mock.MagicMock() + + with mock.patch.object(template, '_build_program') as inner_build_program: + st._internal_build_program(program_builder=program_builder) + program_builder.with_mappings.assert_called_once_with( + parameter_mapping=parameter_mapping, + measurement_mapping=measurement_mapping, + channel_mapping=channel_mapping + ) + inner_build_program.assert_called_once_with(program_builder=program_builder.with_mappings.return_value.__enter__.return_value) def test_create_program_invalid_measurement_mapping(self) -> None: measurement_mapping = {'meas1': 'meas2'} @@ -351,13 +338,13 @@ def test_create_program_invalid_measurement_mapping(self) -> None: pre_channel_mapping = {'default': 'A'} program_builder = default_program_builder() + program_builder.override( + scope=pre_scope, + measurement_mapping=pre_measurement_mapping, + channel_mapping=pre_channel_mapping + ) with self.assertRaises(KeyError): - st._internal_create_program(scope=pre_scope, - measurement_mapping=pre_measurement_mapping, - channel_mapping=pre_channel_mapping, - to_single_waveform=set(), - global_transformation=None, - program_builder=program_builder) + st._build_program(program_builder=program_builder) def test_create_program_parameter_constraint_violation(self) -> None: measurement_mapping = {'meas1': 'meas2'} @@ -378,13 +365,9 @@ def test_create_program_parameter_constraint_violation(self) -> None: pre_channel_mapping = {'default': 'A'} program_builder = default_program_builder() + program_builder.override(scope=pre_scope, measurement_mapping=pre_measurement_mapping, channel_mapping=pre_channel_mapping) with self.assertRaises(ParameterConstraintViolation): - st._internal_create_program(scope=pre_scope, - measurement_mapping=pre_measurement_mapping, - channel_mapping=pre_channel_mapping, - to_single_waveform=set(), - global_transformation=None, - program_builder=program_builder) + st._build_program(program_builder=program_builder) def test_create_program_subtemplate_none(self) -> None: measurement_mapping = {'meas1': 'meas2'} @@ -405,20 +388,10 @@ def test_create_program_subtemplate_none(self) -> None: pre_channel_mapping = {'default': 'A'} program_builder = LoopBuilder() - st._internal_create_program(scope=pre_scope, - measurement_mapping=pre_measurement_mapping, - channel_mapping=pre_channel_mapping, - to_single_waveform=set(), - global_transformation=None, - program_builder=program_builder) - - self.assertEqual(1, len(template.create_program_calls)) - self.assertEqual((st.map_scope(pre_scope), - st.get_updated_measurement_mapping(pre_measurement_mapping), - st.get_updated_channel_mapping(pre_channel_mapping), - program_builder), - template.create_program_calls[-1]) + program_builder.override(scope=pre_scope, measurement_mapping=pre_measurement_mapping, channel_mapping=pre_channel_mapping) + st._internal_build_program(program_builder=program_builder) + template._internal_build_program.assert_called_once_with(program_builder) self.assertIsNone(program_builder.to_program()) def test_same_channel_error(self): diff --git a/tests/pulses/multi_channel_pulse_template_tests.py b/tests/pulses/multi_channel_pulse_template_tests.py index 617ccb93..d204c420 100644 --- a/tests/pulses/multi_channel_pulse_template_tests.py +++ b/tests/pulses/multi_channel_pulse_template_tests.py @@ -414,7 +414,9 @@ def test_internal_create_program(self): measurement_names={'M'}, waveform=DummyWaveform()) overwritten_channels = {'Y': 'c', 'Z': 'a', 'ToNone': 'foo'} - program_builder = object() + program_builder = mock.Mock() + program_builder.with_transformation = mock.MagicMock() + measurement_mapping = object() channel_mapping = {'Y': 'O', 'Z': 'Z', 'X': 'X', 'ToNone': None} to_single_waveform = object() @@ -426,24 +428,17 @@ def test_internal_create_program(self): pccpt = ParallelChannelPulseTemplate(template, overwritten_channels) scope = DictScope.from_kwargs(c=1.2, a=3.4) - kwargs = {**other_kwargs, 'scope': scope, 'global_transformation': None} - expected_overwritten_channels = {'O': 1.2, 'Z': 3.4} expected_transformation = ParallelChannelTransformation(expected_overwritten_channels) - expected_kwargs = {**kwargs, 'global_transformation': expected_transformation} - - with mock.patch.object(template, '_create_program', spec=template._create_program) as cp_mock: - pccpt._internal_create_program(**kwargs) - cp_mock.assert_called_once_with(**expected_kwargs) - global_transformation = LinearTransformation(numpy.zeros((0, 0)), [], []) - expected_transformation = chain_transformations(global_transformation, expected_transformation) - kwargs = {**other_kwargs, 'scope': scope, 'global_transformation': global_transformation} - expected_kwargs = {**kwargs, 'global_transformation': expected_transformation} + program_builder.build_context = mock.Mock() + program_builder.build_context.scope = scope + program_builder.build_context.channel_mapping = channel_mapping - with mock.patch.object(template, '_create_program', spec=template._create_program) as cp_mock: - pccpt._internal_create_program(**kwargs) - cp_mock.assert_called_once_with(**expected_kwargs) + with mock.patch.object(template, '_build_program', spec=template._build_program) as cp_mock: + pccpt._internal_build_program(program_builder) + program_builder.with_transformation.assert_called_once_with(expected_transformation) + cp_mock.assert_called_once_with(program_builder=program_builder.with_transformation.return_value.__enter__.return_value) def test_build_waveform(self): template = DummyPulseTemplate(duration='t1', defined_channels={'X', 'Y'}, parameter_names={'a', 'b'}, diff --git a/tests/pulses/pulse_template_tests.py b/tests/pulses/pulse_template_tests.py index a0dd165b..38df1079 100644 --- a/tests/pulses/pulse_template_tests.py +++ b/tests/pulses/pulse_template_tests.py @@ -8,6 +8,7 @@ import sympy from qupulse.parameter_scope import Scope, DictScope +from qupulse.program.transformation import IdentityTransformation from qupulse.pulses.sequence_pulse_template import SequencePulseTemplate from qupulse.utils.types import ChannelID from qupulse.expressions import Expression, ExpressionScalar @@ -81,7 +82,7 @@ def _internal_create_program(self, *, channel_mapping: Dict[ChannelID, Optional[ChannelID]], global_transformation: Optional[Transformation], to_single_waveform: Set[Union[str, 'PulseTemplate']], - parent_loop: Loop): + program_builder): raise NotImplementedError() @property @@ -104,6 +105,19 @@ def final_values(self) -> Dict[ChannelID, ExpressionScalar]: return self._final_values +def get_appending_internal_build_program(waveform=DummyWaveform(), + always_append=False, + measurements: list=None): + def internal_create_program(program_builder: ProgramBuilder): + scope = program_builder.build_context.scope + if always_append or 'append_a_child' in scope: + if measurements is not None: + program_builder.measure(measurements=measurements) + program_builder.play_arbitrary_waveform(waveform=waveform) + + return internal_create_program + + def get_appending_internal_create_program(waveform=DummyWaveform(), always_append=False, measurements: list=None): @@ -176,7 +190,7 @@ def test_create_program(self) -> None: expected_scope = DictScope.from_kwargs(foo=2.126, bar=-26.2, hugo=math.exp(math.sin(math.pi/2)), volatile=volatile, append_a_child=1) to_single_waveform = {'voll', 'toggo'} - global_transformation = TransformationStub() + global_transformation = IdentityTransformation() expected_internal_kwargs = dict(scope=expected_scope, measurement_mapping=measurement_mapping, @@ -190,16 +204,18 @@ def test_create_program(self) -> None: program_builder = LoopBuilder() with mock.patch.object(template, - '_create_program', - wraps=get_appending_internal_create_program(dummy_waveform)) as _create_program: - with mock.patch('qupulse.pulses.pulse_template.default_program_builder', return_value=program_builder): - program = template.create_program(parameters=parameters, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - to_single_waveform=to_single_waveform, - global_transformation=global_transformation, - volatile=volatile) - _create_program.assert_called_once_with(**expected_internal_kwargs, program_builder=program_builder) + '_build_program', + wraps=get_appending_internal_build_program(dummy_waveform)) as _build_program: + with mock.patch.object(program_builder, "override", wraps=program_builder.override) as override: + with mock.patch('qupulse.pulses.pulse_template.default_program_builder', return_value=program_builder): + program = template.create_program(parameters=parameters, + measurement_mapping=measurement_mapping, + channel_mapping=channel_mapping, + to_single_waveform=to_single_waveform, + global_transformation=global_transformation, + volatile=volatile) + override.assert_called_once_with(**expected_internal_kwargs) + _build_program.assert_called_once_with(program_builder=program_builder) self.assertEqual(expected_program, program) self.assertEqual(previos_measurement_mapping, measurement_mapping) self.assertEqual(previous_channel_mapping, channel_mapping) @@ -215,12 +231,13 @@ def test__create_program(self): template = PulseTemplateStub() with mock.patch.object(template, '_internal_create_program') as _internal_create_program: - template._create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - global_transformation=global_transformation, - to_single_waveform=to_single_waveform, - program_builder=program_builder) + with self.assertWarns(DeprecationWarning): + template._create_program(scope=scope, + measurement_mapping=measurement_mapping, + channel_mapping=channel_mapping, + global_transformation=global_transformation, + to_single_waveform=to_single_waveform, + program_builder=program_builder) _internal_create_program.assert_called_once_with( scope=scope, @@ -234,12 +251,13 @@ def test__create_program(self): with self.assertRaisesRegex(NotImplementedError, "volatile"): template._parameter_names = {'c'} - template._create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - global_transformation=global_transformation, - to_single_waveform={template}, - program_builder=program_builder) + with self.assertWarns(DeprecationWarning): + template._create_program(scope=scope, + measurement_mapping=measurement_mapping, + channel_mapping=channel_mapping, + global_transformation=global_transformation, + to_single_waveform={template}, + program_builder=program_builder) def test__create_program_single_waveform(self): template = PulseTemplateStub(identifier='pt_identifier', parameter_names={'alpha'}) @@ -276,12 +294,13 @@ def test__create_program_single_waveform(self): with mock.patch('qupulse.program.loop.to_waveform', return_value=single_waveform) as to_waveform: with mock.patch('qupulse.program.loop.LoopBuilder', return_value=inner_program_builder): - template._create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - global_transformation=global_transformation, - to_single_waveform=to_single_waveform, - program_builder=program_builder) + with self.assertWarns(DeprecationWarning): + template._create_program(scope=scope, + measurement_mapping=measurement_mapping, + channel_mapping=channel_mapping, + global_transformation=global_transformation, + to_single_waveform=to_single_waveform, + program_builder=program_builder) _internal_create_program.assert_called_once_with(scope=scope, measurement_mapping=measurement_mapping, @@ -330,10 +349,11 @@ def test_create_program_channel_mapping(self): to_single_waveform=set()) with mock.patch('qupulse.pulses.pulse_template.default_program_builder') as pb: - with mock.patch.object(template, '_internal_create_program') as _internal_create_program: + with mock.patch.object(template, '_build_program') as _build_program: template.create_program(channel_mapping={'A': 'C'}) pb.assert_called_once_with() - _internal_create_program.assert_called_once_with(**expected_internal_kwargs, program_builder=pb.return_value) + pb.return_value.override.assert_called_once_with(**expected_internal_kwargs) + _build_program.assert_called_once_with(program_builder=pb.return_value) def test_create_program_volatile(self): template = PulseTemplateStub(defined_channels={'A', 'B'}) @@ -696,7 +716,7 @@ def identifier_map(identifier): class AtomicPulseTemplateTests(unittest.TestCase): - def test_internal_create_program(self) -> None: + def test_internal_build_program(self) -> None: measurement_windows = [('M', 0, 5)] single_wf = DummyWaveform(duration=6, defined_channels={'A'}) wf = MultiChannelWaveform([single_wf]) @@ -709,19 +729,21 @@ def test_internal_create_program(self) -> None: expected_program = Loop(children=[Loop(waveform=wf)], measurements=[('N', 0, 5)]) + program_builder.override( + scope=scope, + measurement_mapping=measurement_mapping, + channel_mapping=channel_mapping, + global_transformation=None, + to_single_waveform=set() + ) with mock.patch.object(template, 'build_waveform', return_value=wf) as build_waveform: - template._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - program_builder=program_builder, - to_single_waveform=set(), - global_transformation=None) + template._internal_build_program(program_builder=program_builder) build_waveform.assert_called_once_with(parameters=scope, channel_mapping=channel_mapping) program = program_builder.to_program() self.assertEqual(expected_program, program) - def test_internal_create_program_transformation(self): + def test_internal_build_program_transformation(self): inner_wf = DummyWaveform() template = AtomicPulseTemplateStub(parameter_names=set()) program_builder = LoopBuilder() @@ -729,17 +751,14 @@ def test_internal_create_program_transformation(self): scope = DictScope.from_kwargs() expected_program = Loop(children=[Loop(waveform=TransformingWaveform(inner_wf, global_transformation))]) + program_builder.override(scope=scope, global_transformation=global_transformation) + with mock.patch.object(template, 'build_waveform', return_value=inner_wf): - template._internal_create_program(scope=scope, - measurement_mapping={}, - channel_mapping={}, - program_builder=program_builder, - to_single_waveform=set(), - global_transformation=global_transformation) + template._internal_build_program(program_builder=program_builder) program = program_builder.to_program() self.assertEqual(expected_program, program) - def test_internal_create_program_no_waveform(self) -> None: + def test_internal_build_program_no_waveform(self) -> None: measurement_windows = [('M', 0, 5)] template = AtomicPulseTemplateStub(measurements=measurement_windows, parameter_names={'foo'}) @@ -747,37 +766,39 @@ def test_internal_create_program_no_waveform(self) -> None: measurement_mapping = {'M': 'N'} channel_mapping = {'B': 'A'} program_builder = LoopBuilder() - - expected_program = Loop() + program_builder.override( + scope=scope, + measurement_mapping=measurement_mapping, + channel_mapping=channel_mapping, + global_transformation=None, + to_single_waveform=set() + ) with mock.patch.object(template, 'build_waveform', return_value=None) as build_waveform: with mock.patch.object(template, 'get_measurement_windows', wraps=template.get_measurement_windows) as get_meas_windows: - template._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - program_builder=program_builder, - to_single_waveform=set(), - global_transformation=None) + template._internal_build_program(program_builder=program_builder) build_waveform.assert_called_once_with(parameters=scope, channel_mapping=channel_mapping) get_meas_windows.assert_not_called() self.assertIsNone(program_builder.to_program()) - def test_internal_create_program_volatile(self): + def test_internal_build_program_volatile(self): template = AtomicPulseTemplateStub(parameter_names={'foo'}) scope = DictScope.from_kwargs(foo=3.5, bar=3, volatile={'foo'}) measurement_mapping = {'M': 'N'} channel_mapping = {'B': 'A'} program_builder = LoopBuilder() + program_builder.override( + scope=scope, + measurement_mapping=measurement_mapping, + channel_mapping=channel_mapping, + global_transformation=None, + to_single_waveform=set() + ) with self.assertRaisesRegex(AssertionError, "volatile"): - template._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - program_builder=program_builder, - to_single_waveform=set(), - global_transformation=None) + template._internal_build_program(program_builder=program_builder) self.assertIsNone(program_builder.to_program()) diff --git a/tests/pulses/repetition_pulse_template_tests.py b/tests/pulses/repetition_pulse_template_tests.py index 459f80c0..8ee170e7 100644 --- a/tests/pulses/repetition_pulse_template_tests.py +++ b/tests/pulses/repetition_pulse_template_tests.py @@ -3,7 +3,7 @@ from unittest import mock from qupulse.parameter_scope import Scope, DictScope -from qupulse.program.waveforms import RepetitionWaveform +from qupulse.program.waveforms import RepetitionWaveform, TransformingWaveform from qupulse.utils.types import FrozenDict from qupulse.program import default_program_builder @@ -17,7 +17,8 @@ from tests.serialization_dummies import DummySerializer from tests.serialization_tests import SerializableTests from tests._program.transformation_tests import TransformationStub -from tests.pulses.pulse_template_tests import PulseTemplateStub, get_appending_internal_create_program +from tests.pulses.pulse_template_tests import PulseTemplateStub, get_appending_internal_create_program, \ + get_appending_internal_build_program class RepetitionPulseTemplateTest(unittest.TestCase): @@ -119,7 +120,7 @@ def test_with_repetition(self): class RepetitionPulseTemplateSequencingTests(MeasurementWindowTestCase): - def test_internal_create_program(self): + def test_internal_build_program(self): wf = DummyWaveform(duration=2.) body = PulseTemplateStub() @@ -136,36 +137,32 @@ def test_internal_create_program(self): to_single_waveform = {'to', 'single', 'waveform'} program_builder = LoopBuilder() - expected_program = Loop(children=[Loop(children=[Loop(waveform=wf)], repetition_count=6)], + expected_wf = TransformingWaveform(wf, global_transformation) + expected_program = Loop(children=[Loop(children=[Loop(waveform=expected_wf)], repetition_count=6)], measurements=[('l', .1, .2)]) - - real_relevant_parameters = dict(n_rep=3, mul=2, a=0.1, b=0.2) - - with mock.patch.object(body, '_create_program', - wraps=get_appending_internal_create_program(wf, always_append=True)) as body_create_program: - with mock.patch.object(rpt, 'validate_scope') as validate_scope: - with mock.patch.object(rpt, 'get_repetition_count_value', return_value=6) as get_repetition_count_value: - with mock.patch.object(rpt, 'get_measurement_windows', return_value=[('l', .1, .2)]) as get_meas: - rpt._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - global_transformation=global_transformation, - to_single_waveform=to_single_waveform, - program_builder=program_builder) - program = program_builder.to_program() - - self.assertEqual(program, expected_program) - body_create_program.assert_called_once_with(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - global_transformation=global_transformation, - to_single_waveform=to_single_waveform, - program_builder=program_builder) - validate_scope.assert_called_once_with(scope) - get_repetition_count_value.assert_called_once_with(scope) - get_meas.assert_called_once_with(scope, measurement_mapping) - - def test_create_program_constant_success_measurements(self) -> None: + program_builder.override( + scope=scope, + measurement_mapping=measurement_mapping, + channel_mapping=channel_mapping, + global_transformation=global_transformation, + to_single_waveform=to_single_waveform, + ) + program_builder.with_repetition = mock.Mock(wraps=program_builder.with_repetition) + + with mock.patch.object(body, '_build_program', + wraps=get_appending_internal_build_program(wf, always_append=True)) as body_build_program: + with mock.patch.object(rpt, 'get_repetition_count_value', return_value=6) as get_repetition_count_value: + with mock.patch.object(rpt, 'get_measurement_windows', return_value=[('l', .1, .2)]) as get_meas: + rpt._internal_build_program(program_builder=program_builder) + program = program_builder.to_program() + + self.assertEqual(program, expected_program) + body_build_program.assert_called_once_with(program_builder=program_builder) + get_repetition_count_value.assert_called_once_with(scope) + get_meas.assert_called_once_with(scope, measurement_mapping) + program_builder.with_repetition.assert_called_once_with(6, measurements=[('l', .1, .2)]) + + def test_build_program_constant_success_measurements(self) -> None: repetitions = 3 body = DummyPulseTemplate(duration=2.0, waveform=DummyWaveform(duration=2, defined_channels={'A'}), measurements=[('b', 0, 1)]) t = RepetitionPulseTemplate(body, repetitions, parameter_constraints=['foo<9'], measurements=[('my', 2, 2)]) @@ -173,12 +170,13 @@ def test_create_program_constant_success_measurements(self) -> None: measurement_mapping = {'my': 'thy', 'b': 'b'} channel_mapping = {} program_builder = LoopBuilder() - t._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - to_single_waveform=set(), - global_transformation=None, - program_builder=program_builder) + program_builder.override( + scope=scope, + measurement_mapping=measurement_mapping, + channel_mapping=channel_mapping, + ) + + t._internal_build_program(program_builder=program_builder) program = program_builder.to_program() @@ -187,9 +185,7 @@ def test_create_program_constant_success_measurements(self) -> None: self.assertEqual(repetitions, internal_loop.repetition_count) self.assertEqual(1, len(internal_loop)) - self.assertEqual((scope, measurement_mapping, channel_mapping, program_builder), body.create_program_calls[-1]) self.assertEqual(body.waveform, internal_loop[0].waveform) - self.assert_measurement_windows_equal({'b': ([0, 2, 4], [1, 1, 1]), 'thy': ([2], [2])}, program.get_measurement_windows()) # done in MultiChannelProgram @@ -198,39 +194,7 @@ def test_create_program_constant_success_measurements(self) -> None: self.assert_measurement_windows_equal({'b': ([0, 2, 4], [1, 1, 1]), 'thy': ([2], [2])}, program.get_measurement_windows()) - def test_create_program_declaration_success(self) -> None: - repetitions = "foo" - body = DummyPulseTemplate(duration=2.0, waveform=DummyWaveform(duration=2, defined_channels={'A'})) - t = RepetitionPulseTemplate(body, repetitions, parameter_constraints=['foo<9']) - scope = DictScope.from_kwargs(foo=3) - measurement_mapping = dict(moth='fire') - channel_mapping = dict(asd='f') - program_builder = LoopBuilder() - t._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - to_single_waveform=set(), - global_transformation=None, - program_builder=program_builder) - program = program_builder.to_program() - - self.assertEqual(1, program.repetition_count) - self.assertEqual(1, len(program.children)) - internal_loop = program.children[0] # type: Loop - self.assertEqual(scope[repetitions], internal_loop.repetition_count) - - self.assertEqual(1, len(internal_loop)) - self.assertEqual((scope, measurement_mapping, channel_mapping, program_builder), - body.create_program_calls[-1]) - self.assertEqual(body.waveform, internal_loop[0].waveform) - - self.assert_measurement_windows_equal({}, program.get_measurement_windows()) - - # ensure same result as from Sequencer - ## not the same as from Sequencer. Sequencer simplifies the whole thing to a single loop executing the waveform 3 times - ## due to absence of non-repeated measurements. create_program currently does no such optimization - - def test_create_program_declaration_success_appended_measurements(self) -> None: + def test_build_program_declaration_success_appended_measurements(self) -> None: repetitions = "foo" body = DummyPulseTemplate(duration=2.0, waveform=DummyWaveform(duration=2), measurements=[('b', 0, 1)]) t = RepetitionPulseTemplate(body, repetitions, parameter_constraints=['foo<9'], @@ -241,14 +205,9 @@ def test_create_program_declaration_success_appended_measurements(self) -> None: children = [Loop(waveform=DummyWaveform(duration=0))] program = Loop(children=children, measurements=[('a', [0], [1])], repetition_count=2) program_builder = LoopBuilder._testing_dummy([program]) + program_builder.override(scope=scope, measurement_mapping=measurement_mapping, channel_mapping=channel_mapping) - t._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - to_single_waveform=set(), - global_transformation=None, - program_builder=program_builder) - + t._internal_build_program(program_builder=program_builder) self.assertEqual(2, program.repetition_count) self.assertEqual(2, len(program.children)) self.assertIs(program.children[0], children[0]) @@ -256,17 +215,12 @@ def test_create_program_declaration_success_appended_measurements(self) -> None: self.assertEqual(scope[repetitions], internal_loop.repetition_count) self.assertEqual(1, len(internal_loop)) - self.assertEqual((scope, measurement_mapping, channel_mapping, program_builder), body.create_program_calls[-1]) self.assertEqual(body.waveform, internal_loop[0].waveform) - self.assert_measurement_windows_equal({'fire': ([0, 6], [7.1, 7.1]), 'b': ([0, 2, 4, 6, 8, 10], [1, 1, 1, 1, 1, 1]), 'a': ([0], [1])}, program.get_measurement_windows()) - # not ensure same result as from Sequencer here - we're testing appending to an already existing parent loop - # which is a use case that does not immediately arise from using Sequencer - - def test_create_program_declaration_success_measurements(self) -> None: + def test_build_program_declaration_success_measurements(self) -> None: repetitions = "foo" body = DummyPulseTemplate(duration=2.0, waveform=DummyWaveform(duration=2), measurements=[('b', 0, 1)]) t = RepetitionPulseTemplate(body, repetitions, parameter_constraints=['foo<9'], measurements=[('moth', 0, 'meas_end')]) @@ -274,12 +228,8 @@ def test_create_program_declaration_success_measurements(self) -> None: measurement_mapping = dict(moth='fire', b='b') channel_mapping = dict(asd='f') program_builder = LoopBuilder() - t._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - to_single_waveform=set(), - global_transformation=None, - program_builder=program_builder) + program_builder.override(scope=scope, measurement_mapping=measurement_mapping, channel_mapping=channel_mapping) + t._internal_build_program(program_builder=program_builder) program = program_builder.to_program() self.assertEqual(1, program.repetition_count) @@ -288,12 +238,11 @@ def test_create_program_declaration_success_measurements(self) -> None: self.assertEqual(scope[repetitions], internal_loop.repetition_count) self.assertEqual(1, len(internal_loop)) - self.assertEqual((scope, measurement_mapping, channel_mapping, program_builder), body.create_program_calls[-1]) self.assertEqual(body.waveform, internal_loop[0].waveform) self.assert_measurement_windows_equal({'fire': ([0], [7.1]), 'b': ([0, 2, 4], [1, 1, 1])}, program.get_measurement_windows()) - def test_create_program_declaration_exceeds_bounds(self) -> None: + def test_build_program_declaration_exceeds_bounds(self) -> None: repetitions = "foo" body_program = Loop(waveform=DummyWaveform(duration=1.0)) body = DummyPulseTemplate(duration=2.0, program=body_program) @@ -305,20 +254,16 @@ def test_create_program_declaration_exceeds_bounds(self) -> None: children = [Loop(waveform=DummyWaveform(duration=0))] program = Loop(children=children) program_builder = LoopBuilder._testing_dummy([program]) + program_builder.override(scope=scope, measurement_mapping=measurement_mapping, channel_mapping=channel_mapping) with self.assertRaises(ParameterConstraintViolation): - t._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - to_single_waveform=set(), - global_transformation=None, - program_builder=program_builder) + t._build_program(program_builder=program_builder) self.assertFalse(body.create_program_calls) self.assertEqual(1, program.repetition_count) self.assertEqual(children, list(program.children)) self.assertIsNone(program.waveform) self.assert_measurement_windows_equal({}, program.get_measurement_windows()) - def test_create_program_declaration_parameter_not_provided(self) -> None: + def test_build_program_declaration_parameter_not_provided(self) -> None: repetitions = "foo" body = DummyPulseTemplate(waveform=DummyWaveform(duration=2.0)) t = RepetitionPulseTemplate(body, repetitions, parameter_constraints=['foo<9'], measurements=[('a', 'd', 1)]) @@ -328,21 +273,9 @@ def test_create_program_declaration_parameter_not_provided(self) -> None: children = [Loop(waveform=DummyWaveform(duration=0))] program = Loop(children=children) program_builder = LoopBuilder._testing_dummy([program]) + program_builder.override(scope=scope, measurement_mapping=measurement_mapping, channel_mapping=channel_mapping) with self.assertRaises(ParameterNotProvidedException): - t._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - to_single_waveform=set(), - global_transformation=None, - program_builder=program_builder) - - with self.assertRaises(ParameterNotProvidedException): - t._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - to_single_waveform=set(), - global_transformation=None, - program_builder=program_builder) + t._internal_build_program(program_builder=program_builder) self.assertFalse(body.create_program_calls) self.assertEqual(1, program.repetition_count) @@ -350,7 +283,7 @@ def test_create_program_declaration_parameter_not_provided(self) -> None: self.assertIsNone(program.waveform) self.assert_measurement_windows_equal({}, program.get_measurement_windows()) - def test_create_program_declaration_parameter_value_not_whole(self) -> None: + def test_build_program_declaration_parameter_value_not_whole(self) -> None: repetitions = "foo" body = DummyPulseTemplate(duration=2.0, waveform=DummyWaveform(duration=2.0)) t = RepetitionPulseTemplate(body, repetitions, parameter_constraints=['foo<9']) @@ -360,21 +293,17 @@ def test_create_program_declaration_parameter_value_not_whole(self) -> None: children = [Loop(waveform=DummyWaveform(duration=0))] program = Loop(children=children) program_builder = LoopBuilder._testing_dummy([program]) + program_builder.override(scope=scope, measurement_mapping=measurement_mapping, channel_mapping=channel_mapping) with self.assertRaises(ParameterNotIntegerException): - t._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - to_single_waveform=set(), - global_transformation=None, - program_builder=program_builder) + t._internal_build_program(program_builder=program_builder) self.assertFalse(body.create_program_calls) self.assertEqual(1, program.repetition_count) self.assertEqual(children, list(program.children)) self.assertIsNone(program.waveform) self.assert_measurement_windows_equal({}, program.get_measurement_windows()) - def test_create_program_constant_measurement_mapping_failure(self) -> None: + def test_build_program_constant_measurement_mapping_failure(self) -> None: repetitions = "foo" body = DummyPulseTemplate(duration=2.0, waveform=DummyWaveform(duration=2.0), measurements=[('b', 0, 1)]) t = RepetitionPulseTemplate(body, repetitions, parameter_constraints=['foo<9'], measurements=[('a', 0, 1)]) @@ -384,31 +313,26 @@ def test_create_program_constant_measurement_mapping_failure(self) -> None: children = [Loop(waveform=DummyWaveform(duration=0))] program = Loop(children=children) program_builder = LoopBuilder._testing_dummy([program]) + program_builder.override( + scope=scope, + channel_mapping=channel_mapping, + measurement_mapping=measurement_mapping,) with self.assertRaises(KeyError): - t._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - to_single_waveform=set(), - global_transformation=None, - program_builder=program_builder) + t._internal_build_program(program_builder=program_builder) # test for failure on child level measurement_mapping = dict(a='a') + program_builder.override(measurement_mapping=measurement_mapping) with self.assertRaises(KeyError): - t._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - to_single_waveform=set(), - global_transformation=None, - program_builder=program_builder) + t._internal_build_program(program_builder=program_builder) self.assertFalse(body.create_program_calls) self.assertEqual(1, program.repetition_count) self.assertEqual(children, list(program.children)) self.assertIsNone(program.waveform) self.assert_measurement_windows_equal({}, program.get_measurement_windows()) - def test_create_program_rep_count_zero_constant(self) -> None: + def test_build_program_rep_count_zero_constant(self) -> None: repetitions = 0 body_program = Loop(waveform=DummyWaveform(duration=1.0)) body = DummyPulseTemplate(duration=2.0, program=body_program) @@ -422,17 +346,13 @@ def test_create_program_rep_count_zero_constant(self) -> None: channel_mapping = dict(asd='f') program_builder = LoopBuilder() - t._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - to_single_waveform=set(), - global_transformation=None, - program_builder=program_builder) + program_builder.override(scope=scope, measurement_mapping=measurement_mapping, channel_mapping=channel_mapping) + t._internal_build_program(program_builder=program_builder) program = program_builder.to_program() self.assertFalse(body.create_program_calls) self.assertIsNone(program) - def test_create_program_rep_count_zero_constant_with_measurement(self) -> None: + def test_build_program_rep_count_zero_constant_with_measurement(self) -> None: repetitions = 0 body_program = Loop(waveform=DummyWaveform(duration=1.0)) body = DummyPulseTemplate(duration=2.0, program=body_program) @@ -446,17 +366,12 @@ def test_create_program_rep_count_zero_constant_with_measurement(self) -> None: channel_mapping = dict(asd='f') program_builder = default_program_builder() - t._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - to_single_waveform=set(), - global_transformation=None, - program_builder=program_builder) + program_builder.override(scope=scope, measurement_mapping=measurement_mapping, channel_mapping=channel_mapping) + t._internal_build_program(program_builder=program_builder) program = program_builder.to_program() - self.assertFalse(body.create_program_calls) self.assertIsNone(program) - def test_create_program_rep_count_zero_declaration(self) -> None: + def test_build_program_rep_count_zero_declaration(self) -> None: repetitions = "foo" body_program = Loop(waveform=DummyWaveform(duration=1.0)) body = DummyPulseTemplate(duration=2.0, program=body_program) @@ -470,17 +385,13 @@ def test_create_program_rep_count_zero_declaration(self) -> None: channel_mapping = dict(asd='f') program_builder = default_program_builder() - t._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - to_single_waveform=set(), - global_transformation=None, - program_builder=program_builder) + program_builder.override(scope=scope, measurement_mapping=measurement_mapping, channel_mapping=channel_mapping) + t._internal_build_program(program_builder=program_builder) program = program_builder.to_program() self.assertFalse(body.create_program_calls) self.assertIsNone(program) - def test_create_program_rep_count_zero_declaration_with_measurement(self) -> None: + def test_build_program_rep_count_zero_declaration_with_measurement(self) -> None: repetitions = "foo" body_program = Loop(waveform=DummyWaveform(duration=1.0)) body = DummyPulseTemplate(duration=2.0, program=body_program) @@ -494,22 +405,19 @@ def test_create_program_rep_count_zero_declaration_with_measurement(self) -> Non channel_mapping = dict(asd='f') program_builder = default_program_builder() - t._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - to_single_waveform=set(), - global_transformation=None, - program_builder=program_builder) + program_builder.override(scope=scope, measurement_mapping=measurement_mapping, channel_mapping=channel_mapping) + t._internal_build_program(program_builder=program_builder) program = program_builder.to_program() self.assertFalse(body.create_program_calls) self.assertIsNone(program) - def test_create_program_rep_count_neg_declaration(self) -> None: + def test_build_program_rep_count_neg_declaration(self) -> None: repetitions = "foo" body_program = Loop(waveform=DummyWaveform(duration=1.0)) body = DummyPulseTemplate(duration=2.0, program=body_program) - # suppress warning about 0 repetitions on construction here, we are only interested in correct behavior during sequencing (i.e., do nothing) + # suppress warning about 0 repetitions on construction here. + # We are only interested in correct behavior during sequencing (i.e., do nothing) with warnings.catch_warnings(record=True): t = RepetitionPulseTemplate(body, repetitions) @@ -518,17 +426,15 @@ def test_create_program_rep_count_neg_declaration(self) -> None: channel_mapping = dict(asd='f') program_builder = default_program_builder() - t._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - to_single_waveform=set(), - global_transformation=None, - program_builder=program_builder) + program_builder.override( + scope=scope, measurement_mapping=measurement_mapping, channel_mapping=channel_mapping, + ) + t._internal_build_program(program_builder=program_builder) program = program_builder.to_program() self.assertFalse(body.create_program_calls) self.assertIsNone(program) - def test_create_program_rep_count_neg_declaration_with_measurements(self) -> None: + def test_build_program_rep_count_neg_declaration_with_measurements(self) -> None: repetitions = "foo" body_program = Loop(waveform=DummyWaveform(duration=1.0)) body = DummyPulseTemplate(duration=2.0, program=body_program) @@ -542,17 +448,13 @@ def test_create_program_rep_count_neg_declaration_with_measurements(self) -> Non channel_mapping = dict(asd='f') program_builder = default_program_builder() - t._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - to_single_waveform=set(), - global_transformation=None, - program_builder=program_builder) + program_builder.override(scope=scope, measurement_mapping=measurement_mapping, channel_mapping=channel_mapping) + t._internal_build_program(program_builder=program_builder) program = program_builder.to_program() self.assertFalse(body.create_program_calls) self.assertIsNone(program) - def test_create_program_none_subprogram(self) -> None: + def test_build_program_none_subprogram(self) -> None: repetitions = "foo" body = DummyPulseTemplate(duration=0.0, waveform=None) t = RepetitionPulseTemplate(body, repetitions, parameter_constraints=['foo<9']) @@ -560,15 +462,11 @@ def test_create_program_none_subprogram(self) -> None: measurement_mapping = dict(moth='fire') channel_mapping = dict(asd='f') program_builder = default_program_builder() - t._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - to_single_waveform=set(), - global_transformation=None, - program_builder=program_builder) + program_builder.override(scope=scope, measurement_mapping=measurement_mapping, channel_mapping=channel_mapping) + t._internal_build_program(program_builder=program_builder) self.assertIsNone(program_builder.to_program()) - def test_create_program_none_subprogram_with_measurement(self) -> None: + def test_build_program_none_subprogram_with_measurement(self) -> None: repetitions = "foo" body = DummyPulseTemplate(duration=2.0, waveform=None, measurements=[('b', 2, 3)]) t = RepetitionPulseTemplate(body, repetitions, parameter_constraints=['foo<9'], @@ -577,13 +475,9 @@ def test_create_program_none_subprogram_with_measurement(self) -> None: measurement_mapping = dict(moth='fire', b='b') channel_mapping = dict(asd='f') program_builder = default_program_builder() + program_builder.override(scope=scope, measurement_mapping=measurement_mapping, channel_mapping=channel_mapping) - t._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - to_single_waveform=set(), - global_transformation=None, - program_builder=program_builder) + t._internal_build_program(program_builder=program_builder) self.assertIsNone(program_builder.to_program()) def test_single_waveform(self): diff --git a/tests/pulses/sequence_pulse_template_tests.py b/tests/pulses/sequence_pulse_template_tests.py index 65f283d5..24791bdc 100644 --- a/tests/pulses/sequence_pulse_template_tests.py +++ b/tests/pulses/sequence_pulse_template_tests.py @@ -3,6 +3,7 @@ from qupulse.parameter_scope import DictScope from qupulse.expressions import Expression, ExpressionScalar +from qupulse.program.waveforms import TransformingWaveform from qupulse.pulses.table_pulse_template import TablePulseTemplate from qupulse.pulses.sequence_pulse_template import SequencePulseTemplate, SequenceWaveform from qupulse.pulses.mapping_pulse_template import MappingPulseTemplate @@ -14,7 +15,8 @@ from tests.serialization_dummies import DummySerializer from tests.serialization_tests import SerializableTests from tests._program.transformation_tests import TransformationStub -from tests.pulses.pulse_template_tests import get_appending_internal_create_program, PulseTemplateStub +from tests.pulses.pulse_template_tests import get_appending_internal_create_program, PulseTemplateStub, \ + get_appending_internal_build_program class SequencePulseTemplateTest(unittest.TestCase): @@ -234,7 +236,7 @@ def test_deserialize_old(self) -> None: class SequencePulseTemplateSequencingTests(MeasurementWindowTestCase): - def test_internal_create_program(self): + def test_build_program(self): sub_templates = PulseTemplateStub(defined_channels={'a'}, duration=ExpressionScalar('t1')),\ PulseTemplateStub(defined_channels={'a'}, duration=ExpressionScalar('t2')) @@ -250,22 +252,24 @@ def test_internal_create_program(self): channel_mapping={'g': 'h'}, global_transformation=TransformationStub(), to_single_waveform={'to', 'single', 'waveform'}) + expected_wfs = TransformingWaveform(wfs[0], kwargs['global_transformation']), TransformingWaveform(wfs[1], kwargs['global_transformation']) program_builder = LoopBuilder() + program_builder.override(**kwargs) - expected_program = Loop(children=[Loop(waveform=wfs[0]), - Loop(waveform=wfs[1])], + expected_program = Loop(children=[Loop(waveform=expected_wfs[0]), + Loop(waveform=expected_wfs[1])], measurements=[('l', .1, .2)]) with mock.patch.object(spt, 'validate_scope') as validate_scope: with mock.patch.object(spt, 'get_measurement_windows', return_value=[('l', .1, .2)]) as get_measurement_windows: - with mock.patch.object(sub_templates[0], '_create_program', - wraps=get_appending_internal_create_program(wfs[0], True)) as create_0,\ - mock.patch.object(sub_templates[1], '_create_program', - wraps=get_appending_internal_create_program(wfs[1], True)) as create_1: + with mock.patch.object(sub_templates[0], '_build_program', + wraps=get_appending_internal_build_program(wfs[0], True)) as create_0,\ + mock.patch.object(sub_templates[1], '_build_program', + wraps=get_appending_internal_build_program(wfs[1], True)) as create_1: - spt._internal_create_program(**kwargs, program_builder=program_builder) + spt._build_program(program_builder=program_builder) program = program_builder.to_program() @@ -273,10 +277,10 @@ def test_internal_create_program(self): validate_scope.assert_called_once_with(kwargs['scope']) get_measurement_windows.assert_called_once_with(kwargs['scope'], kwargs['measurement_mapping']) - create_0.assert_called_once_with(**kwargs, program_builder=program_builder) - create_1.assert_called_once_with(**kwargs, program_builder=program_builder) + create_0.assert_called_once_with(program_builder=program_builder) + create_1.assert_called_once_with(program_builder=program_builder) - def test_create_program_internal(self) -> None: + def test_internal_build_program_internal(self) -> None: sub1 = DummyPulseTemplate(duration=3, waveform=DummyWaveform(duration=3), measurements=[('b', 1, 2)], defined_channels={'A'}) sub2 = DummyPulseTemplate(duration=2, waveform=DummyWaveform(duration=2), parameter_names={'foo'}, defined_channels={'A'}) scope = DictScope.from_kwargs() @@ -284,13 +288,9 @@ def test_create_program_internal(self) -> None: channel_mapping = dict() seq = SequencePulseTemplate(sub1, sub2, measurements=[('a', 0, 1)]) program_builder = LoopBuilder() + program_builder.override(measurement_mapping=measurement_mapping, channel_mapping=channel_mapping, scope=scope) - seq._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - global_transformation=None, - to_single_waveform=set(), - program_builder=program_builder) + seq._internal_build_program(program_builder=program_builder) loop = program_builder.to_program() self.assertEqual(1, loop.repetition_count) self.assertIsNone(loop.waveform) @@ -302,12 +302,8 @@ def test_create_program_internal(self) -> None: ### test again with inverted sequence seq = SequencePulseTemplate(sub2, sub1, measurements=[('a', 0, 1)]) program_builder = LoopBuilder() - seq._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - global_transformation=None, - to_single_waveform=set(), - program_builder=program_builder) + program_builder.override(measurement_mapping=measurement_mapping, channel_mapping=channel_mapping, scope=scope) + seq._internal_build_program(program_builder=program_builder) loop = program_builder.to_program() self.assertEqual(1, loop.repetition_count) self.assertIsNone(loop.waveform) @@ -316,7 +312,7 @@ def test_create_program_internal(self) -> None: list(loop.children)) self.assert_measurement_windows_equal({'a': ([0], [1]), 'b': ([3], [2])}, loop.get_measurement_windows()) - def test_internal_create_program_no_measurement_mapping(self) -> None: + def test_internal_build_program_no_measurement_mapping(self) -> None: sub1 = DummyPulseTemplate(duration=3, waveform=DummyWaveform(duration=3), measurements=[('b', 1, 2)]) sub2 = DummyPulseTemplate(duration=2, waveform=DummyWaveform(duration=2), parameter_names={'foo'}) scope = DictScope.from_kwargs() @@ -324,14 +320,10 @@ def test_internal_create_program_no_measurement_mapping(self) -> None: children = [Loop(waveform=DummyWaveform())] loop = Loop(measurements=[], children=children) program_builder = LoopBuilder._testing_dummy([loop]) + program_builder.override(scope=scope, measurement_mapping={}) with self.assertRaises(KeyError): - seq._internal_create_program(scope=scope, - measurement_mapping=dict(), - channel_mapping=dict(), - global_transformation=None, - to_single_waveform=set(), - program_builder=program_builder) + seq._build_program(program_builder=program_builder) self.assertFalse(sub1.create_program_calls) self.assertFalse(sub2.create_program_calls) self.assertEqual(children, list(loop.children)) @@ -340,15 +332,14 @@ def test_internal_create_program_no_measurement_mapping(self) -> None: self.assert_measurement_windows_equal({}, loop.get_measurement_windows()) # test for child level measurements (does not guarantee to leave parent_loop unchanged in this case) + program_builder.override( + scope=scope, + measurement_mapping=dict(a='a'), + ) with self.assertRaises(KeyError): - seq._internal_create_program(scope=scope, - measurement_mapping=dict(a='a'), - channel_mapping=dict(), - global_transformation=None, - to_single_waveform=set(), - program_builder=program_builder) - - def test_internal_create_program_one_child_no_duration(self) -> None: + seq._internal_build_program(program_builder=program_builder) + + def test_internal_build_program_one_child_no_duration(self) -> None: sub1 = DummyPulseTemplate(duration=0, waveform=None, measurements=[('b', 1, 2)], defined_channels={'A'}) sub2 = DummyPulseTemplate(duration=2, waveform=DummyWaveform(duration=2), parameter_names={'foo'}, defined_channels={'A'}) scope = DictScope.from_kwargs() @@ -356,12 +347,8 @@ def test_internal_create_program_one_child_no_duration(self) -> None: channel_mapping = dict() seq = SequencePulseTemplate(sub1, sub2, measurements=[('a', 0, 1)]) program_builder = LoopBuilder() - seq._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - global_transformation=None, - to_single_waveform=set(), - program_builder=program_builder) + program_builder.override(measurement_mapping=measurement_mapping, channel_mapping=channel_mapping, scope=scope) + seq._internal_build_program(program_builder=program_builder) loop = program_builder.to_program() self.assertEqual(1, loop.repetition_count) self.assertIsNone(loop.waveform) @@ -376,12 +363,8 @@ def test_internal_create_program_one_child_no_duration(self) -> None: ### test again with inverted sequence seq = SequencePulseTemplate(sub2, sub1, measurements=[('a', 0, 1)]) program_builder = LoopBuilder() - seq._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - global_transformation=None, - to_single_waveform=set(), - program_builder=program_builder) + program_builder.override(measurement_mapping=measurement_mapping, channel_mapping=channel_mapping, scope=scope) + seq._internal_build_program(program_builder=program_builder) loop = program_builder.to_program() self.assertEqual(1, loop.repetition_count) self.assertIsNone(loop.waveform) @@ -393,7 +376,7 @@ def test_internal_create_program_one_child_no_duration(self) -> None: loop.cleanup() self.assert_measurement_windows_equal({'a': ([0], [1])}, loop.get_measurement_windows()) - def test_internal_create_program_both_children_no_duration(self) -> None: + def test_internal_build_program_both_children_no_duration(self) -> None: sub1 = DummyPulseTemplate(duration=0, waveform=None, measurements=[('b', 1, 2)], defined_channels={'A'}) sub2 = DummyPulseTemplate(duration=0, waveform=None, parameter_names={'foo'}, defined_channels={'A'}) scope = DictScope.from_kwargs() @@ -402,30 +385,24 @@ def test_internal_create_program_both_children_no_duration(self) -> None: seq = SequencePulseTemplate(sub1, sub2, measurements=[('a', 0, 1)]) program_builder = default_program_builder() - seq._internal_create_program(scope=scope, - measurement_mapping=measurement_mapping, - channel_mapping=channel_mapping, - global_transformation=None, - to_single_waveform=set(), - program_builder=program_builder) + program_builder.override(measurement_mapping=measurement_mapping, channel_mapping=channel_mapping, scope=scope) + seq._internal_build_program(program_builder=program_builder) self.assertIsNone(program_builder.to_program()) - def test_internal_create_program_parameter_constraint_violations(self) -> None: + def test_build_program_parameter_constraint_violations(self) -> None: sub1 = DummyPulseTemplate(duration=3, waveform=DummyWaveform(duration=3), measurements=[('b', 1, 2)]) sub2 = DummyPulseTemplate(duration=2, waveform=DummyWaveform(duration=2), parameter_names={'foo'}) scope = DictScope.from_kwargs(foo=7) seq = SequencePulseTemplate(sub1, sub2, measurements=[('a', 0, 1)], parameter_constraints={'foo < 2'}) program_builder = default_program_builder() + program_builder.override( + scope=scope, + ) with self.assertRaises(ParameterConstraintViolation): - seq._internal_create_program(scope=scope, - measurement_mapping={'a': 'a', 'b': 'b'}, - channel_mapping=dict(), - global_transformation=None, - to_single_waveform=set(), - program_builder=program_builder) + seq._build_program(program_builder=program_builder) self.assertIsNone(program_builder.to_program()) - def test_internal_create_program_parameter_missing(self) -> None: + def test_internal_build_program_parameter_missing(self) -> None: sub1 = DummyPulseTemplate(duration=3, waveform=DummyWaveform(duration=3), measurements=[('b', 1, 2)]) sub2 = DummyPulseTemplate(duration=2, waveform=DummyWaveform(duration=2), parameter_names={'foo'}) seq = SequencePulseTemplate(sub1, sub2, measurements=[('a', 'bar', 1)], parameter_constraints={'foo < 2'}) @@ -433,25 +410,17 @@ def test_internal_create_program_parameter_missing(self) -> None: # test parameter from constraints scope = DictScope.from_kwargs() program_builder = default_program_builder() + program_builder.override(scope=scope, measurement_mapping={'a': 'a', 'b': 'b'}) with self.assertRaises(ParameterNotProvidedException): - seq._internal_create_program(scope=scope, - measurement_mapping={'a': 'a', 'b': 'b'}, - channel_mapping=dict(), - global_transformation=None, - to_single_waveform=set(), - program_builder=program_builder) + seq._build_program(program_builder=program_builder) self.assertIsNone(program_builder.to_program()) # test parameter from measurements scope = DictScope.from_mapping({'foo': 1}) program_builder = default_program_builder() + program_builder.override(scope=scope, measurement_mapping={'a': 'a', 'b': 'b'}) with self.assertRaises(ParameterNotProvidedException): - seq._internal_create_program(scope=scope, - measurement_mapping={'a': 'a', 'b': 'b'}, - channel_mapping=dict(), - global_transformation=None, - to_single_waveform=set(), - program_builder=program_builder) + seq._build_program(program_builder=program_builder) self.assertIsNone(program_builder.to_program()) diff --git a/tests/pulses/sequencing_dummies.py b/tests/pulses/sequencing_dummies.py index c70cd8d8..5518c023 100644 --- a/tests/pulses/sequencing_dummies.py +++ b/tests/pulses/sequencing_dummies.py @@ -3,6 +3,7 @@ import typing from typing import Tuple, List, Dict, Optional, Set, Any, Union, Mapping import copy +from unittest import mock import numpy import unittest @@ -189,10 +190,25 @@ def __init__(self, if integrals is not None: assert isinstance(integrals, Mapping) + self._internal_build_program = mock.MagicMock(wraps=self._internal_build_program) + @property def duration(self): return self._duration + def _internal_build_program(self, program_builder: ProgramBuilder): + measurements = self.get_measurement_windows(program_builder.build_context.scope, + measurement_mapping=program_builder.build_context.measurement_mapping) + if self._program: + program_builder = typing.cast(program_builder, qupulse.program.loop.LoopBuilder) + parent_loop = program_builder._top + + parent_loop.add_measurements(measurements) + parent_loop.append_child(waveform=self._program.waveform, children=self._program.children) + elif self.waveform: + program_builder.measure(measurements) + program_builder.play_arbitrary_waveform(waveform=self.waveform) + @property def parameter_names(self) -> Set[str]: return set(self.parameter_names_)