21369040529

Committed 26 Jan 2026 06:23PM UTC coverage: 88.721%. First build

Build # 21369040529

Build Type

Pull #933

github

Committed by

web-flow

Commit Message

Merge c6f2e02ff into 2d86c016d

Pull Request Pull Request #933: Refactor context management for program builders

Run Details

453 of 484 new or added lines in 26 files covered. (93.6%)

19170 of 21607 relevant lines covered (88.72%)

5.32 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.1

/qupulse/program/protocol.py

"""Definition of the program builder protocol."""
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.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, ChannelID


@runtime_checkable
class Program(Protocol):
    """This protocol is used to inspect and or manipulate programs. As you can see the functionality is very limited
    because most of a program class' capability are specific to the implementation."""

    @property
    @abstractmethod
    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):
    """This protocol is used by :py:meth:`.PulseTemplate.create_program` to build a program via a variation of the
    visitor pattern.

    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
    :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
    cases. However, this is not relevant for HDAWG specific implementations because these are expected to ignore
    :py:meth:`.ProgramBuilder.measure` calls.

    This interface makes heavy use of context managers and generators/iterators which allows for flexible iteration
    and repetition implementation.
    """

    @property
    @abstractmethod
    def build_context(self) -> BuildContext:
        """Get the current build context."""

    @property
    @abstractmethod
    def build_settings(self) -> BuildSettings:
        """Get the current build settings"""

    @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
        hardware dependent type which are inserted into the parameter scope via :py:meth:`.ProgramBuilder.with_iteration`.

        Args:
            duration: Duration of voltage hold
            voltages: Voltages for each channel
        """

    # further specialized commandos like play_harmonic might be added here

    @abstractmethod
    def play_arbitrary_waveform(self, waveform: Waveform):
        """Insert the playback of an arbitrary waveform. If possible pulse templates should use more specific commands
        like :py:meth:`.ProgramBuilder.hold_voltage` (the only more specific command at the time of this writing).

        Args:
            waveform: The waveform to play
        """

    @abstractmethod
    def measure(self, measurements: Optional[Sequence[MeasurementWindow]]):
        """Unconditionally add given measurements relative to the current position.

        Args:
            measurements: Measurements to add.
        """

    @abstractmethod
    def with_repetition(self, repetition_count: RepetitionCount,
                        measurements: Optional[Sequence[MeasurementWindow]] = None) -> Iterable['ProgramBuilder']:
        """Start a new repetition context with given repetition count. The caller has to iterate over the return value
        and call `:py:meth:`.ProgramBuilder.inner_scope` inside the iteration context.

        Args:
            repetition_count: Repetition count
            measurements: These measurements are added relative to the position at the start of the iteration iff the
                          iteration is not empty.

        Returns:
            An iterable of :py:class:`ProgramBuilder` instances.
        """

    @abstractmethod
    def with_sequence(self,
                      measurements: Optional[Sequence[MeasurementWindow]] = None) -> ContextManager['ProgramBuilder']:
        """Start a new sequence context. The caller has to enter the returned context manager and add the sequenced
        elements there.

        Measurements that are added in to the returned program builder are discarded if the sequence is empty on exit.

        Args:
            measurements: These measurements are added relative to the position at the start of the sequence iff the
            sequence is not empty.

        Returns:
            A context manager that returns a :py:class:`ProgramBuilder` on entering.
        """

    @abstractmethod
    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.

        Returns:
            A context manager that returns a :py:class:`ProgramBuilder` on entering.
        """

    @abstractmethod
    def with_iteration(self, index_name: str, rng: range,
                       measurements: Optional[Sequence[MeasurementWindow]] = None) -> Iterable['ProgramBuilder']:
        """Create an iterable that represent the body of the iteration. This can be an iterable with an element for each
        step in the iteration or a single object that represents the complete iteration.

        Args:
            index_name: The name of index
            rng: The range if the index
            measurements: Measurements to add iff the iteration body is not empty.
        """

    @abstractmethod
    def time_reversed(self) -> ContextManager['ProgramBuilder']:
        """This returns a new context manager that will reverse everything added to it in time upon exit.

        Returns:
            A context manager that returns a :py:class:`ProgramBuilder` on entering.
        """

    @abstractmethod
    def to_program(self) -> Optional[Program]:
        """Generate the final program. This is allowed to invalidate the program builder.

        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)

1	"""Definition of the program builder protocol."""
2	import copy	6✔
3	import dataclasses	6✔
4	from abc import abstractmethod, ABC	6✔
5	from contextlib import contextmanager	6✔
6	from typing import runtime_checkable, Protocol, Mapping, Optional, Sequence, Iterable, ContextManager, AbstractSet, \	6✔
7	Union
8
9	from qupulse import MeasurementWindow	6✔
10	from qupulse.expressions import Expression	6✔
11	from qupulse.parameter_scope import Scope, MappedScope	6✔
12	from qupulse.program.waveforms import Waveform, ConstantWaveform, TransformingWaveform	6✔
13	from qupulse.program.transformation import Transformation, chain_transformations	6✔
14	from qupulse.program.values import RepetitionCount, HardwareTime, HardwareVoltage	6✔
15	from qupulse.pulses.metadata import TemplateMetadata	6✔
16
17	from qupulse.utils.types import TimeType, ChannelID	6✔
18
19
20	@runtime_checkable	6✔
21	class Program(Protocol):	6✔
22	"""This protocol is used to inspect and or manipulate programs. As you can see the functionality is very limited
23	because most of a program class' capability are specific to the implementation."""
24
25	@property	6✔
26	@abstractmethod	6✔
27	def duration(self) -> TimeType:	6✔
28	"""The duration of the program in nanoseconds."""
29
30	@abstractmethod	6✔
31	def get_defined_channels(self) -> AbstractSet[ChannelID]:	6✔
32	"""Get the set of channels that are used in this program."""
33
34
35	@dataclasses.dataclass	6✔
36	class BuildContext:	6✔
37	"""This dataclass bundles the mutable context information during the build."""
38
39	scope: Scope = None	6✔
40	measurement_mapping: Mapping[str, Optional[str]] = None	6✔
41	channel_mapping: Mapping[ChannelID, Optional[ChannelID]] = None	6✔
42	transformation: Optional[Transformation] = None	6✔
43	minimal_sample_rate: Optional[TimeType] = None	6✔
44
45	def apply_mappings(self,	6✔
46	parameter_mapping: Mapping[str, Expression] = None,
47	measurement_mapping: Mapping[str, Optional[str]] = None,
48	channel_mapping: Mapping[ChannelID, Optional[ChannelID]] = None,
49	) -> "BuildContext":
50	scope = self.scope	6✔
51	if parameter_mapping is not None:	6✔
52	scope = MappedScope(scope=scope, mapping=parameter_mapping)	6✔
53	mapped_measurement_mapping = self.measurement_mapping	6✔
54	if measurement_mapping is not None:	6✔
55	mapped_measurement_mapping = {k: mapped_measurement_mapping[v] for k, v in measurement_mapping.items()}	6✔
56	mapped_channel_mapping = self.channel_mapping	6✔
57	if channel_mapping is not None:	6✔
58	mapped_channel_mapping = {inner_ch: None if outer_ch is None else mapped_channel_mapping[outer_ch]	6✔
59	for inner_ch, outer_ch in channel_mapping.items()}
60	return BuildContext(scope=scope, measurement_mapping=mapped_measurement_mapping, channel_mapping=mapped_channel_mapping, transformation=self.transformation, minimal_sample_rate=self.minimal_sample_rate)	6✔
61
62
63	@dataclasses.dataclass(frozen=True)	6✔
64	class BuildSettings:	6✔
65	"""This dataclass bundles the immutable settings."""
66	to_single_waveform: AbstractSet[str \| object]	6✔
67
68
69	@runtime_checkable	6✔
70	class ProgramBuilder(Protocol):	6✔
71	"""This protocol is used by :py:meth:`.PulseTemplate.create_program` to build a program via a variation of the
72	visitor pattern.
73
74	The pulse templates call the methods that correspond to their functionality on the program builder. For example,
75	:py:class:`.ConstantPulseTemplate` translates itself into a simple :py:meth:`.ProgramBuilder.hold_voltage` call while
76	:py:class:`SequencePulseTemplate` uses :py:meth:`.ProgramBuilder.with_sequence` to signify a logical unit with
77	attached measurements and passes the resulting object to the sequenced sub-templates.
78
79	Due to backward compatibility, the handling of measurements is a bit weird since they have to be omitted in certain
80	cases. However, this is not relevant for HDAWG specific implementations because these are expected to ignore
81	:py:meth:`.ProgramBuilder.measure` calls.
82
83	This interface makes heavy use of context managers and generators/iterators which allows for flexible iteration
84	and repetition implementation.
85	"""
86
87	@property	6✔
88	@abstractmethod	6✔
89	def build_context(self) -> BuildContext:	6✔
90	"""Get the current build context."""
91
92	@property	6✔
93	@abstractmethod	6✔
94	def build_settings(self) -> BuildSettings:	6✔
95	"""Get the current build settings"""
96
97	@abstractmethod	6✔
98	def override(self,	6✔
99	scope: Scope = None,
100	measurement_mapping: Optional[Mapping[str, Optional[str]]] = None,
101	channel_mapping: Optional[Mapping[ChannelID, Optional[ChannelID]]] = None,
102	global_transformation: Optional[Transformation] = None,
103	to_single_waveform: AbstractSet[str \| object] = None,):
104	"""Override the non-None values in context and settings"""
105
106	@abstractmethod	6✔
107	def with_mappings(self, *,	6✔
108	parameter_mapping: Mapping[str, Expression],
109	measurement_mapping: Mapping[str, Optional[str]],
110	channel_mapping: Mapping[ChannelID, Optional[ChannelID]],
111	) -> ContextManager['ProgramBuilder']:
112	"""Modify the build context for the duration of the context manager.
113
114	Args:
115	parameter_mapping: A mapping of parameter names to expressions.
116	measurement_mapping: A mapping of measurement names to measurement names or None.
117	channel_mapping: A mapping of channel IDs to channel IDs or None.
118	"""
119
120	@abstractmethod	6✔
121	def with_transformation(self, transformation: Transformation) -> ContextManager['ProgramBuilder']:	6✔
122	"""Modify the build context for the duration of the context manager."""
123
124	@abstractmethod	6✔
125	def with_metadata(self, metadata: TemplateMetadata) -> ContextManager['ProgramBuilder']:	6✔
126	"""Modify the build context for the duration of the context manager."""
127
128	@abstractmethod	6✔
129	def hold_voltage(self, duration: HardwareTime, voltages: Mapping[str, HardwareVoltage]):	6✔
130	"""Hold the specified voltage for a given time. Advances the current time by ``duration``. The values are
131	hardware dependent type which are inserted into the parameter scope via :py:meth:`.ProgramBuilder.with_iteration`.
132
133	Args:
134	duration: Duration of voltage hold
135	voltages: Voltages for each channel
136	"""
137
138	# further specialized commandos like play_harmonic might be added here
139
140	@abstractmethod	6✔
141	def play_arbitrary_waveform(self, waveform: Waveform):	6✔
142	"""Insert the playback of an arbitrary waveform. If possible pulse templates should use more specific commands
143	like :py:meth:`.ProgramBuilder.hold_voltage` (the only more specific command at the time of this writing).
144
145	Args:
146	waveform: The waveform to play
147	"""
148
149	@abstractmethod	6✔
150	def measure(self, measurements: Optional[Sequence[MeasurementWindow]]):	6✔
151	"""Unconditionally add given measurements relative to the current position.
152
153	Args:
154	measurements: Measurements to add.
155	"""
156
157	@abstractmethod	6✔
158	def with_repetition(self, repetition_count: RepetitionCount,	6✔
159	measurements: Optional[Sequence[MeasurementWindow]] = None) -> Iterable['ProgramBuilder']:
160	"""Start a new repetition context with given repetition count. The caller has to iterate over the return value
161	and call `:py:meth:`.ProgramBuilder.inner_scope` inside the iteration context.
162
163	Args:
164	repetition_count: Repetition count
165	measurements: These measurements are added relative to the position at the start of the iteration iff the
166	iteration is not empty.
167
168	Returns:
169	An iterable of :py:class:`ProgramBuilder` instances.
170	"""
171
172	@abstractmethod	6✔
173	def with_sequence(self,	6✔
174	measurements: Optional[Sequence[MeasurementWindow]] = None) -> ContextManager['ProgramBuilder']:
175	"""Start a new sequence context. The caller has to enter the returned context manager and add the sequenced
176	elements there.
177
178	Measurements that are added in to the returned program builder are discarded if the sequence is empty on exit.
179
180	Args:
181	measurements: These measurements are added relative to the position at the start of the sequence iff the
182	sequence is not empty.
183
184	Returns:
185	A context manager that returns a :py:class:`ProgramBuilder` on entering.
186	"""
187
188	@abstractmethod	6✔
189	def new_subprogram(self) -> ContextManager['ProgramBuilder']:	6✔
190	"""Create a context managed program builder whose contents are translated into a single waveform upon exit if
191	it is not empty.
192
193	Returns:
194	A context manager that returns a :py:class:`ProgramBuilder` on entering.
195	"""
196
197	@abstractmethod	6✔
198	def with_iteration(self, index_name: str, rng: range,	6✔
199	measurements: Optional[Sequence[MeasurementWindow]] = None) -> Iterable['ProgramBuilder']:
200	"""Create an iterable that represent the body of the iteration. This can be an iterable with an element for each
201	step in the iteration or a single object that represents the complete iteration.
202
203	Args:
204	index_name: The name of index
205	rng: The range if the index
206	measurements: Measurements to add iff the iteration body is not empty.
207	"""
208
209	@abstractmethod	6✔
210	def time_reversed(self) -> ContextManager['ProgramBuilder']:	6✔
211	"""This returns a new context manager that will reverse everything added to it in time upon exit.
212
213	Returns:
214	A context manager that returns a :py:class:`ProgramBuilder` on entering.
215	"""
216
217	@abstractmethod	6✔
218	def to_program(self) -> Optional[Program]:	6✔
219	"""Generate the final program. This is allowed to invalidate the program builder.
220
221	Returns:
222	A program implementation. None if nothing was added to this program builder.
223	"""
224
225
226	class BaseProgramBuilder(ProgramBuilder, ABC):	6✔
227	"""Helper base class for program builder to reduce code duplication. The interface is defined by :py:class:`ProgramBuilder`.
228
229	This class provides shared functionality for context and settings and correct transformation handling.
230	"""
231
232	def __init__(self, initial_context: BuildContext = None, initial_settings: BuildSettings = None):	6✔
233	self._build_context_stack: list[BuildContext] = [BuildContext() if initial_context is None else initial_context]	6✔
234	self._build_settings_stack: list[BuildSettings] = [BuildSettings(set()) if initial_settings is None else initial_settings]	6✔
235
236	@property	6✔
237	def build_context(self) -> BuildContext:	6✔
238	return self._build_context_stack[-1]	6✔
239
240	@property	6✔
241	def build_settings(self) -> BuildSettings:	6✔
242	return self._build_settings_stack[-1]	6✔
243
244	def override(self,	6✔
245	scope: Scope = None,
246	measurement_mapping: Optional[Mapping[str, Optional[str]]] = None,
247	channel_mapping: Optional[Mapping[ChannelID, Optional[ChannelID]]] = None,
248	global_transformation: Optional[Transformation] = None,
249	to_single_waveform: AbstractSet[Union[str, 'PulseTemplate']] = None):
250	old_context = self._build_context_stack[-1]	6✔
251	context = BuildContext(	6✔
252	scope=old_context.scope if scope is None else scope,
253	measurement_mapping=old_context.measurement_mapping if measurement_mapping is None else measurement_mapping,
254	channel_mapping=old_context.channel_mapping if channel_mapping is None else channel_mapping,
255	transformation=old_context.transformation if global_transformation is None else global_transformation,
256	)
257	old_settings = self._build_settings_stack[-1]	6✔
258	settings = BuildSettings(	6✔
259	to_single_waveform=old_settings.to_single_waveform if to_single_waveform is None else to_single_waveform,
260	)
261
262	self._build_context_stack.append(context)	6✔
263	self._build_settings_stack.append(settings)	6✔
264
265	@contextmanager	6✔
266	def _with_patched_context(self, **kwargs):	6✔
267	context = copy.copy(self._build_context_stack[-1])	6✔
268	for name, value in kwargs.items():	6✔
269	setattr(context, name, value)	6✔
270	self._build_context_stack.append(context)	6✔
271	yield	6✔
272	self._build_context_stack.pop()	6✔
273
274	@contextmanager	6✔
275	def with_metadata(self, metadata: TemplateMetadata):	6✔
276	# metadata.to_single_waveform == "always" is handled in PulseTemplate._build_program
NEW 277	if metadata.minimal_sample_rate is not None:	×
NEW 278	with self._with_patched_context(minimal_sample_rate=metadata.minimal_sample_rate) as builder:	×
NEW 279	yield builder	×
280	else:
NEW 281	yield self	×
282
283	@contextmanager	6✔
284	def with_transformation(self, transformation: Transformation):	6✔
285	context = copy.copy(self.build_context)	6✔
286	context.transformation = chain_transformations(context.transformation, transformation)	6✔
287	self._build_context_stack.append(context)	6✔
288	yield self	6✔
289	self._build_context_stack.pop()	6✔
290
291	@contextmanager	6✔
292	def with_mappings(self, *,	6✔
293	parameter_mapping: Mapping[str, Expression],
294	measurement_mapping: Mapping[str, Optional[str]],
295	channel_mapping: Mapping[ChannelID, Optional[ChannelID]],
296	):
297	context = self.build_context.apply_mappings(parameter_mapping, measurement_mapping, channel_mapping)	6✔
298	self._build_context_stack.append(context)	6✔
299	yield self	6✔
300	self._build_context_stack.pop()	6✔
301
302	@abstractmethod	6✔
303	def _transformed_hold_voltage(self, duration: HardwareTime, voltages: Mapping[str, HardwareVoltage]):	6✔
304	"""This internal function gets the constant voltage values transformed by the current built context's transformation.
305	"""
306
307	@abstractmethod	6✔
308	def _transformed_play_arbitrary_waveform(self, waveform: Waveform):	6✔
309	"""This internal function gets the waveform transformed by the current built context's transformation."""
310
311	def play_arbitrary_waveform(self, waveform: Waveform):	6✔
312	transformation = self.build_context.transformation	6✔
313	if transformation:	6✔
314	transformed_waveform = TransformingWaveform(waveform, transformation)	6✔
315	self._transformed_play_arbitrary_waveform(transformed_waveform)	6✔
316	else:
317	self._transformed_play_arbitrary_waveform(waveform)	6✔
318
319	def hold_voltage(self, duration: HardwareTime, voltages: Mapping[str, HardwareVoltage]):	6✔
320	transformation = self.build_context.transformation	6✔
321	if transformation:	6✔
322	if transformation.get_constant_output_channels(voltages.keys()) != transformation.get_output_channels(voltages.keys()):	6✔
323	waveform = TransformingWaveform(ConstantWaveform.from_mapping(duration, voltages), transformation)	6✔
324	self._transformed_play_arbitrary_waveform(waveform)	6✔
325	else:
326	transformed_voltages = transformation(0.0, voltages)	6✔
327	self._transformed_hold_voltage(duration, transformed_voltages)	6✔
328	else:
329	self._transformed_hold_voltage(duration, voltages)	6✔

qutech / qupulse / 21369040529

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous