Refactor context management for program builders

changes.d/933.feature

@@ -0,0 +1 @@
+Replace PulseTemplate._create_program(**a_lot_of__kwargs) with PulseTemplate._build_program(program_builder).

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):

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'],

qupulse/program/loop.py

@@ -3,26 +3,30 @@
 # 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
 from contextlib import contextmanager
 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,31 +772,25 @@ 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:
      - This program builder does not use the Loop class to generate the measurements
 
     """
 
-    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):