• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

qutech / qupulse / 21436745373

28 Jan 2026 11:39AM UTC coverage: 88.726%. First build
21436745373

Pull #936

github

web-flow
Merge f2cbaf6ea into fa1c64c00
Pull Request #936: Change to_single_waveform handling

16 of 20 new or added lines in 3 files covered. (80.0%)

19179 of 21616 relevant lines covered (88.73%)

5.32 hits per line

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

99.3
/qupulse/program/protocol.py
1
"""Definition of the program builder protocol."""
2
import contextlib
6✔
3
import copy
6✔
4
import dataclasses
6✔
5
from abc import abstractmethod, ABC
6✔
6
from contextlib import contextmanager
6✔
7
from typing import runtime_checkable, Protocol, Mapping, Optional, Sequence, Iterable, ContextManager, AbstractSet, \
6✔
8
    Union
9

10
from qupulse import MeasurementWindow
6✔
11
from qupulse.expressions import Expression
6✔
12
from qupulse.parameter_scope import Scope, MappedScope
6✔
13
from qupulse.program.waveforms import Waveform, ConstantWaveform, TransformingWaveform
6✔
14
from qupulse.program.transformation import Transformation, chain_transformations
6✔
15
from qupulse.program.values import RepetitionCount, HardwareTime, HardwareVoltage
6✔
16
from qupulse.pulses.metadata import TemplateMetadata
6✔
17

18
from qupulse.utils.types import TimeType, ChannelID
6✔
19

20

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

26
    @property
6✔
27
    @abstractmethod
6✔
28
    def duration(self) -> TimeType:
6✔
29
        """The duration of the program in nanoseconds."""
30

31
    @abstractmethod
6✔
32
    def get_defined_channels(self) -> AbstractSet[ChannelID]:
6✔
33
        """Get the set of channels that are used in this program."""
34

35

36
@dataclasses.dataclass
6✔
37
class BuildContext:
6✔
38
    """This dataclass bundles the mutable context information during the build."""
39

40
    scope: Scope = None
6✔
41
    measurement_mapping: Mapping[str, Optional[str]] = None
6✔
42
    channel_mapping: Mapping[ChannelID, Optional[ChannelID]] = None
6✔
43
    transformation: Optional[Transformation] = None
6✔
44
    minimal_sample_rate: Optional[TimeType] = None
6✔
45

46
    def apply_mappings(self,
6✔
47
                       parameter_mapping: Mapping[str, Expression] = None,
48
                       measurement_mapping: Mapping[str, Optional[str]] = None,
49
                       channel_mapping: Mapping[ChannelID, Optional[ChannelID]] = None,
50
                       ) -> "BuildContext":
51
        scope = self.scope
6✔
52
        if parameter_mapping is not None:
6✔
53
            scope = MappedScope(scope=scope, mapping=parameter_mapping)
6✔
54
        mapped_measurement_mapping = self.measurement_mapping
6✔
55
        if measurement_mapping is not None:
6✔
56
            mapped_measurement_mapping = {k: mapped_measurement_mapping[v] for k, v in measurement_mapping.items()}
6✔
57
        mapped_channel_mapping = self.channel_mapping
6✔
58
        if channel_mapping is not None:
6✔
59
            mapped_channel_mapping = {inner_ch: None if outer_ch is None else mapped_channel_mapping[outer_ch]
6✔
60
                                      for inner_ch, outer_ch in channel_mapping.items()}
61
        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✔
62

63

64
@dataclasses.dataclass(frozen=True)
6✔
65
class BuildSettings:
6✔
66
    """This dataclass bundles the immutable settings."""
67
    to_single_waveform: AbstractSet[str]
6✔
68

69

70
@runtime_checkable
6✔
71
class ProgramBuilder(Protocol):
6✔
72
    """This protocol is used by :py:meth:`.PulseTemplate.create_program` to build a program via a variation of the
73
    visitor pattern.
74

75
    The pulse templates call the methods that correspond to their functionality on the program builder. For example,
76
    :py:class:`.ConstantPulseTemplate` translates itself into a simple :py:meth:`.ProgramBuilder.hold_voltage` call while
77
    :py:class:`SequencePulseTemplate` uses :py:meth:`.ProgramBuilder.with_sequence` to signify a logical unit with
78
    attached measurements and passes the resulting object to the sequenced sub-templates.
79

80
    Due to backward compatibility, the handling of measurements is a bit weird since they have to be omitted in certain
81
    cases. However, this is not relevant for HDAWG specific implementations because these are expected to ignore
82
    :py:meth:`.ProgramBuilder.measure` calls.
83

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

88
    @property
6✔
89
    @abstractmethod
6✔
90
    def build_context(self) -> BuildContext:
6✔
91
        """Get the current build context."""
92

93
    @property
6✔
94
    @abstractmethod
6✔
95
    def build_settings(self) -> BuildSettings:
6✔
96
        """Get the current build settings"""
97

98
    @abstractmethod
6✔
99
    def override(self,
6✔
100
                 scope: Scope = None,
101
                 measurement_mapping: Optional[Mapping[str, Optional[str]]] = None,
102
                 channel_mapping: Optional[Mapping[ChannelID, Optional[ChannelID]]] = None,
103
                 global_transformation: Optional[Transformation] = None,
104
                 to_single_waveform: AbstractSet[str | object] = None,):
105
        """Override the non-None values in context and settings"""
106

107
    @abstractmethod
6✔
108
    def with_mappings(self, *,
6✔
109
                      parameter_mapping: Mapping[str, Expression],
110
                      measurement_mapping: Mapping[str, Optional[str]],
111
                      channel_mapping: Mapping[ChannelID, Optional[ChannelID]],
112
                      ) -> ContextManager['ProgramBuilder']:
113
        """Modify the build context for the duration of the context manager.
114

115
        Args:
116
            parameter_mapping: A mapping of parameter names to expressions.
117
            measurement_mapping: A mapping of measurement names to measurement names or None.
118
            channel_mapping: A mapping of channel IDs to channel IDs or None.
119
        """
120

121
    @abstractmethod
6✔
122
    def with_transformation(self, transformation: Transformation) -> ContextManager['ProgramBuilder']:
6✔
123
        """Modify the build context for the duration of the context manager."""
124

125
    @abstractmethod
6✔
126
    def with_metadata(self, metadata: TemplateMetadata, identifier: str | None) -> ContextManager['ProgramBuilder']:
6✔
127
        """Modify the build context for the duration of the context manager."""
128

129
    @abstractmethod
6✔
130
    def hold_voltage(self, duration: HardwareTime, voltages: Mapping[str, HardwareVoltage]):
6✔
131
        """Hold the specified voltage for a given time. Advances the current time by ``duration``. The values are
132
        hardware dependent type which are inserted into the parameter scope via :py:meth:`.ProgramBuilder.with_iteration`.
133

134
        Args:
135
            duration: Duration of voltage hold
136
            voltages: Voltages for each channel
137
        """
138

139
    # further specialized commandos like play_harmonic might be added here
140

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

146
        Args:
147
            waveform: The waveform to play
148
        """
149

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

154
        Args:
155
            measurements: Measurements to add.
156
        """
157

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

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

169
        Returns:
170
            An iterable of :py:class:`ProgramBuilder` instances.
171
        """
172

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

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

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

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

189
    @abstractmethod
6✔
190
    def new_subprogram(self) -> ContextManager['ProgramBuilder']:
6✔
191
        """Create a context managed program builder whose contents are translated into a single waveform upon exit if
192
        it is not empty.
193

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

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

204
        Args:
205
            index_name: The name of index
206
            rng: The range if the index
207
            measurements: Measurements to add iff the iteration body is not empty.
208
        """
209

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

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

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

222
        Returns:
223
            A program implementation. None if nothing was added to this program builder.
224
        """
225

226

227
class BaseProgramBuilder(ProgramBuilder, ABC):
6✔
228
    """Helper base class for program builder to reduce code duplication. The interface is defined by :py:class:`ProgramBuilder`.
229

230
    This class provides shared functionality for context and settings and correct transformation handling.
231
    """
232

233
    def __init__(self, initial_context: BuildContext = None, initial_settings: BuildSettings = None):
6✔
234
        self._build_context_stack: list[BuildContext] = [BuildContext() if initial_context is None else initial_context]
6✔
235
        self._build_settings_stack: list[BuildSettings] = [BuildSettings(set()) if initial_settings is None else initial_settings]
6✔
236

237
    @property
6✔
238
    def build_context(self) -> BuildContext:
6✔
239
        return self._build_context_stack[-1]
6✔
240

241
    @property
6✔
242
    def build_settings(self) -> BuildSettings:
6✔
243
        return self._build_settings_stack[-1]
6✔
244

245
    def override(self,
6✔
246
                 scope: Scope = None,
247
                 measurement_mapping: Optional[Mapping[str, Optional[str]]] = None,
248
                 channel_mapping: Optional[Mapping[ChannelID, Optional[ChannelID]]] = None,
249
                 global_transformation: Optional[Transformation] = None,
250
                 to_single_waveform: AbstractSet[Union[str, 'PulseTemplate']] = None):
251
        old_context = self._build_context_stack[-1]
6✔
252
        context = BuildContext(
6✔
253
            scope=old_context.scope if scope is None else scope,
254
            measurement_mapping=old_context.measurement_mapping if measurement_mapping is None else measurement_mapping,
255
            channel_mapping=old_context.channel_mapping if channel_mapping is None else channel_mapping,
256
            transformation=old_context.transformation if global_transformation is None else global_transformation,
257
        )
258
        old_settings = self._build_settings_stack[-1]
6✔
259
        settings = BuildSettings(
6✔
260
            to_single_waveform=old_settings.to_single_waveform if to_single_waveform is None else to_single_waveform,
261
        )
262

263
        self._build_context_stack.append(context)
6✔
264
        self._build_settings_stack.append(settings)
6✔
265

266
    @contextmanager
6✔
267
    def _with_patched_context(self, **kwargs):
6✔
268
        context = copy.copy(self._build_context_stack[-1])
6✔
269
        for name, value in kwargs.items():
6✔
270
            setattr(context, name, value)
6✔
271
        self._build_context_stack.append(context)
6✔
272
        yield
6✔
273
        self._build_context_stack.pop()
6✔
274

275
    @contextmanager
6✔
276
    def with_metadata(self, metadata: TemplateMetadata, identifier: str | None):
6✔
277
        stack = contextlib.ExitStack()
6✔
278

279
        builder = self
6✔
280

281
        if metadata.minimal_sample_rate is not None:
6✔
NEW
282
            builder = stack.enter_context(builder._with_patched_context(minimal_sample_rate=metadata.minimal_sample_rate))
×
283

284
        if metadata.to_single_waveform == "always" or identifier in self.build_settings.to_single_waveform:
6✔
285
            builder = stack.enter_context(builder.new_subprogram())
6✔
286

287
        with stack:
6✔
288
            yield builder
6✔
289

290
    @contextmanager
6✔
291
    def with_transformation(self, transformation: Transformation):
6✔
292
        context = copy.copy(self.build_context)
6✔
293
        context.transformation = chain_transformations(context.transformation, transformation)
6✔
294
        self._build_context_stack.append(context)
6✔
295
        yield self
6✔
296
        self._build_context_stack.pop()
6✔
297

298
    @contextmanager
6✔
299
    def with_mappings(self, *,
6✔
300
                      parameter_mapping: Mapping[str, Expression],
301
                      measurement_mapping: Mapping[str, Optional[str]],
302
                      channel_mapping: Mapping[ChannelID, Optional[ChannelID]],
303
                      ):
304
        context = self.build_context.apply_mappings(parameter_mapping, measurement_mapping, channel_mapping)
6✔
305
        self._build_context_stack.append(context)
6✔
306
        yield self
6✔
307
        self._build_context_stack.pop()
6✔
308

309
    @abstractmethod
6✔
310
    def _transformed_hold_voltage(self, duration: HardwareTime, voltages: Mapping[str, HardwareVoltage]):
6✔
311
        """This internal function gets the constant voltage values transformed by the current built context's transformation.
312
        """
313

314
    @abstractmethod
6✔
315
    def _transformed_play_arbitrary_waveform(self, waveform: Waveform):
6✔
316
        """This internal function gets the waveform transformed by the current built context's transformation."""
317

318
    def play_arbitrary_waveform(self, waveform: Waveform):
6✔
319
        transformation = self.build_context.transformation
6✔
320
        if transformation:
6✔
321
            transformed_waveform = TransformingWaveform(waveform, transformation)
6✔
322
            self._transformed_play_arbitrary_waveform(transformed_waveform)
6✔
323
        else:
324
            self._transformed_play_arbitrary_waveform(waveform)
6✔
325

326
    def hold_voltage(self, duration: HardwareTime, voltages: Mapping[str, HardwareVoltage]):
6✔
327
        transformation = self.build_context.transformation
6✔
328
        if transformation:
6✔
329
            if transformation.get_constant_output_channels(voltages.keys()) != transformation.get_output_channels(voltages.keys()):
6✔
330
                waveform = TransformingWaveform(ConstantWaveform.from_mapping(duration, voltages), transformation)
6✔
331
                self._transformed_play_arbitrary_waveform(waveform)
6✔
332
            else:
333
                transformed_voltages = transformation(0.0, voltages)
6✔
334
                self._transformed_hold_voltage(duration, transformed_voltages)
6✔
335
        else:
336
            self._transformed_hold_voltage(duration, voltages)
6✔
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc