16672493391

Committed 01 Aug 2025 10:15AM UTC coverage: 88.761%. First build

Build # 16672493391

Build Type

Pull #917

github

Committed by

web-flow

Commit Message

Merge 19e71ade7 into a8a829612

Pull Request Pull Request #917: Experiment followup dlv

Run Details

134 of 141 new or added lines in 2 files covered. (95.04%)

18915 of 21310 relevant lines covered (88.76%)

5.33 hits per line

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

91.45

/qupulse/program/values.py

"""Runtime variable value implementations."""

from dataclasses import dataclass
from numbers import Real
from typing import TypeVar, Generic, Mapping, Union, List, Tuple, Optional
from types import NotImplementedType
import operator

import numpy as np

from qupulse.program.volatile import VolatileRepetitionCount
from qupulse.utils.types import TimeType
from qupulse.expressions import sympy as sym_expr
from qupulse.utils.sympy import _lambdify_modules


NumVal = TypeVar('NumVal', bound=Real)


@dataclass
class DynamicLinearValue(Generic[NumVal]):
    """This is a potential runtime-evaluable expression of the form

    C + C1*R1 + C2*R2 + ...
    where R1, R2, ... are potential runtime parameters.

    The main use case is the expression of for loop-dependent variables where the Rs are loop indices. There the
    expressions can be calculated via simple increments.

    This class tries to pass a number and a :py:class:`sympy.expr.Expr` on best effort basis.
    """

    #: The part of this expression which is not runtime parameter-dependent
    base: NumVal

    #: A mapping of inner parameter names to the factor with which they contribute to the final value.
    factors: Mapping[str, NumVal]

    def __post_init__(self):
        assert isinstance(self.factors, Mapping)

    def value(self, scope: Mapping[str, NumVal]) -> NumVal:
        """Numeric value of the expression with the given scope.
        Args:
            scope: Scope in which the expression is evaluated.
        Returns:
            The numeric value.
        """
        value = self.base
        for name, factor in self.factors.items():
            value += scope[name] * factor
        return value
    
    def __abs__(self):
        # The deifnition of an absolute value is ambiguous, but there is a case
        # to define it as sum_i abs(f_i) + abs(base) for certain conveniences.
        # return abs(self.base)+sum([abs(o) for o in self.factors.values()])
        raise NotImplementedError(f'abs({self.__class__.__name__}) is ambiguous')
    
    def __eq__(self, other):
        if isinstance(other, type(self)):
            return self.base == other.base and self.factors == other.factors

        if (base_eq := self.base.__eq__(other)) is NotImplemented:
            return NotImplemented

        return base_eq and not self.factors
    
    def __add__(self, other):
        if isinstance(other, (float, int, TimeType)):
            return DynamicLinearValue(self.base + other, self.factors)

        if type(other) == type(self):
            factors = dict(self.factors)
            for name, value in other.factors.items():
                factors[name] = value + factors.get(name, 0)
            return DynamicLinearValue(self.base + other.base, factors)

        # this defers evaluation when other is still a symbolic expression
        return NotImplemented

    def __radd__(self, other):
        return self.__add__(other)

    def __sub__(self, other):
        return self.__add__(-other)

    def __rsub__(self, other):
        return (-self).__add__(other)

    def __neg__(self):
        return DynamicLinearValue(-self.base, {name: -value for name, value in self.factors.items()})

    def __mul__(self, other: NumVal):
        if isinstance(other, (float, int, TimeType)):
            return DynamicLinearValue(self.base * other, {name: other * value for name, value in self.factors.items()})

        # this defers evaluation when other is still a symbolic expression
        return NotImplemented

    def __rmul__(self, other):
        return self.__mul__(other)

    def __truediv__(self, other):
        inv = 1 / other
        return self.__mul__(inv)
    
    def __hash__(self):
        return hash((self.base,frozenset(sorted(self.factors.items()))))
    
    @property
    def free_symbols(self):
        """This is required for the :py:class:`sympy.expr.Expr` interface compliance. Since the keys of
        :py:attr:`.offsets` are internal parameters we do not have free symbols.

        Returns:
            An empty tuple
        """
        return ()

    def _sympy_(self):
        """This method is used by :py:`sympy.sympify`. This class tries to "just work" in the sympy evaluation pipelines.

        Returns:
            self
        """
        return self

    def replace(self, r, s):
        """We mock :class:`sympy.Expr.replace` here. This class does not support inner parameters so there is nothing
        to replace. Importantly, the keys of the offsets are no runtime variables!

        Returns:
            self
        """
        return self


# is there any way to cast the numpy cumprod to int?
int_type = Union[np.int64,np.int32,int]

class ResolutionDependentValue(Generic[NumVal]):
    """This is a potential runtime-evaluable expression of the form
    
    o + sum_i  b_i*m_i
    
    with (potential) float o, b_i and integers m_i. o and b_i are rounded(gridded)
    to a resolution given upon __call__.
    
    The main use case is the correct rounding of increments used in command-based
    voltage scans on some hardware devices, where an imprecise numeric value is
    looped over m_i times and could, if not rounded, accumulate a proportional
    error leading to unintended drift in output voltages when jump-back commands
    afterwards do not account for the deviations.
    Rounding the value preemptively and supplying corrected values to jump-back
    commands prevents this.
    """
    def __init__(self,
                 bases: Tuple[NumVal],
                 multiplicities: Tuple[int],
                 offset: NumVal):

        self.bases = tuple(bases)
        self.multiplicities = tuple(multiplicities)
        self.offset = offset
        self.__is_time_or_int = all(isinstance(b,(TimeType,int_type)) for b in bases) and isinstance(offset,(TimeType,int_type))

    #this is not to circumvent float errors in python, but rounding errors from awg-increment commands.
    #python float are thereby accurate enough
    def __call__(self, resolution: Optional[float]) -> Union[NumVal,TimeType]:
        #with resolution = None handle TimeType/int case?
        if resolution is None:
            assert self.__is_time_or_int
            return sum(b*m for b,m in zip(self.bases,self.multiplicities)) + self.offset
        #resolution as float value of granularity of base val.
        #to avoid conflicts between positive and negative vals from casting
        #half to even, use abs val
        return sum(np.sign(b) * round(abs(b) / resolution) * m * resolution for b,m in zip(self.bases,self.multiplicities))\
             + np.sign(self.offset) * round(abs(self.offset) / resolution) * resolution

    def __bool__(self):
        #if any value is not zero - this helps for some checks
        return any(bool(b) for b in self.bases) or bool(self.offset)

    def __add__(self, other):
        # this should happen in the context of an offset being added to it, not the bases being modified.
        if isinstance(other, (float, int, TimeType)):
            return ResolutionDependentValue(self.bases,self.multiplicities,self.offset+other)
        return NotImplemented

    def __radd__(self, other):
        return self.__add__(other)

    def __sub__(self, other):
        return self.__add__(-other)

    def __mul__(self, other):
        # this should happen when the amplitude is being scaled
        # multiplicities are not affected
        if isinstance(other, (float, int, TimeType)):
            return ResolutionDependentValue(tuple(b*other for b in self.bases),self.multiplicities,self.offset*other)
        return NotImplemented

    def __rmul__(self,other):
        return self.__mul__(other)

    def __truediv__(self,other):
        return self.__mul__(1/other)

    def __float__(self):
        return float(self(resolution=None))

    def __str__(self):
        return f"RDP of {sum(b*m for b,m in zip(self.bases,self.multiplicities)) + self.offset}"

    def __repr__(self):
        return "RDP("+",".join([f"{k}="+v.__str__() for k,v in vars(self).items()])+")"

    def __eq__(self,o):
        if not isinstance(o,ResolutionDependentValue):
            return False
        return self.__dict__ == o.__dict__

    def __hash__(self):
        return hash((self.bases,self.offset,self.multiplicities,self.__is_time_or_int))


#This is a simple dervide class to allow better isinstance checks in the HDAWG driver
@dataclass
class DynamicLinearValueStepped(DynamicLinearValue):
    step_nesting_level: int
    rng: range
    reverse: int|bool


# TODO: hackedy, hackedy
sym_expr.ALLOWED_NUMERIC_SCALAR_TYPES = sym_expr.ALLOWED_NUMERIC_SCALAR_TYPES + (DynamicLinearValue,)

# this keeps the simple expression in lambdified results
_lambdify_modules.append({'DynamicLinearValue': DynamicLinearValue,
                          'DynamicLinearValueStepped': DynamicLinearValueStepped})

RepetitionCount = Union[int, VolatileRepetitionCount, DynamicLinearValue[int]]
HardwareTime = Union[TimeType, DynamicLinearValue[TimeType]]
HardwareVoltage = Union[float, DynamicLinearValue[float]]

1	"""Runtime variable value implementations."""
2
3	from dataclasses import dataclass	6✔
4	from numbers import Real	6✔
5	from typing import TypeVar, Generic, Mapping, Union, List, Tuple, Optional	6✔
6	from types import NotImplementedType	6✔
7	import operator	6✔
8
9	import numpy as np	6✔
10
11	from qupulse.program.volatile import VolatileRepetitionCount	6✔
12	from qupulse.utils.types import TimeType	6✔
13	from qupulse.expressions import sympy as sym_expr	6✔
14	from qupulse.utils.sympy import _lambdify_modules	6✔
15
16
17	NumVal = TypeVar('NumVal', bound=Real)	6✔
18
19
20	@dataclass	6✔
21	class DynamicLinearValue(Generic[NumVal]):	6✔
22	"""This is a potential runtime-evaluable expression of the form
23
24	C + C1R1 + C2R2 + ...
25	where R1, R2, ... are potential runtime parameters.
26
27	The main use case is the expression of for loop-dependent variables where the Rs are loop indices. There the
28	expressions can be calculated via simple increments.
29
30	This class tries to pass a number and a :py:class:`sympy.expr.Expr` on best effort basis.
31	"""
32
33	#: The part of this expression which is not runtime parameter-dependent
34	base: NumVal	6✔
35
36	#: A mapping of inner parameter names to the factor with which they contribute to the final value.
37	factors: Mapping[str, NumVal]	6✔
38
39	def __post_init__(self):	6✔
40	assert isinstance(self.factors, Mapping)	6✔
41
42	def value(self, scope: Mapping[str, NumVal]) -> NumVal:	6✔
43	"""Numeric value of the expression with the given scope.
44	Args:
45	scope: Scope in which the expression is evaluated.
46	Returns:
47	The numeric value.
48	"""
49	value = self.base	6✔
50	for name, factor in self.factors.items():	6✔
51	value += scope[name] * factor	6✔
52	return value	6✔
53
54	def __abs__(self):	6✔
55	# The deifnition of an absolute value is ambiguous, but there is a case
56	# to define it as sum_i abs(f_i) + abs(base) for certain conveniences.
57	# return abs(self.base)+sum([abs(o) for o in self.factors.values()])
NEW 58	raise NotImplementedError(f'abs({self.__class__.__name__}) is ambiguous')	×
59
60	def __eq__(self, other):	6✔
61	if isinstance(other, type(self)):	6✔
62	return self.base == other.base and self.factors == other.factors	6✔
63
64	if (base_eq := self.base.__eq__(other)) is NotImplemented:	6✔
65	return NotImplemented	6✔
66
67	return base_eq and not self.factors	6✔
68
69	def __add__(self, other):	6✔
70	if isinstance(other, (float, int, TimeType)):	6✔
71	return DynamicLinearValue(self.base + other, self.factors)	6✔
72
73	if type(other) == type(self):	6✔
74	factors = dict(self.factors)	6✔
75	for name, value in other.factors.items():	6✔
76	factors[name] = value + factors.get(name, 0)	6✔
77	return DynamicLinearValue(self.base + other.base, factors)	6✔
78
79	# this defers evaluation when other is still a symbolic expression
80	return NotImplemented	×
81
82	def __radd__(self, other):	6✔
83	return self.__add__(other)	×
84
85	def __sub__(self, other):	6✔
86	return self.__add__(-other)	6✔
87
88	def __rsub__(self, other):	6✔
89	return (-self).__add__(other)	×
90
91	def __neg__(self):	6✔
92	return DynamicLinearValue(-self.base, {name: -value for name, value in self.factors.items()})	6✔
93
94	def __mul__(self, other: NumVal):	6✔
95	if isinstance(other, (float, int, TimeType)):	6✔
96	return DynamicLinearValue(self.base * other, {name: other * value for name, value in self.factors.items()})	6✔
97
98	# this defers evaluation when other is still a symbolic expression
99	return NotImplemented	×
100
101	def __rmul__(self, other):	6✔
102	return self.__mul__(other)	6✔
103
104	def __truediv__(self, other):	6✔
105	inv = 1 / other	6✔
106	return self.__mul__(inv)	6✔
107
108	def __hash__(self):	6✔
109	return hash((self.base,frozenset(sorted(self.factors.items()))))	6✔
110
111	@property	6✔
112	def free_symbols(self):	6✔
113	"""This is required for the :py:class:`sympy.expr.Expr` interface compliance. Since the keys of
114	:py:attr:`.offsets` are internal parameters we do not have free symbols.
115
116	Returns:
117	An empty tuple
118	"""
119	return ()	6✔
120
121	def _sympy_(self):	6✔
122	"""This method is used by :py:`sympy.sympify`. This class tries to "just work" in the sympy evaluation pipelines.
123
124	Returns:
125	self
126	"""
127	return self	6✔
128
129	def replace(self, r, s):	6✔
130	"""We mock :class:`sympy.Expr.replace` here. This class does not support inner parameters so there is nothing
131	to replace. Importantly, the keys of the offsets are no runtime variables!
132
133	Returns:
134	self
135	"""
136	return self	6✔
137
138
139	# is there any way to cast the numpy cumprod to int?
140	int_type = Union[np.int64,np.int32,int]	6✔
141
142	class ResolutionDependentValue(Generic[NumVal]):	6✔
143	"""This is a potential runtime-evaluable expression of the form
144
145	o + sum_i b_i*m_i
146
147	with (potential) float o, b_i and integers m_i. o and b_i are rounded(gridded)
148	to a resolution given upon __call__.
149
150	The main use case is the correct rounding of increments used in command-based
151	voltage scans on some hardware devices, where an imprecise numeric value is
152	looped over m_i times and could, if not rounded, accumulate a proportional
153	error leading to unintended drift in output voltages when jump-back commands
154	afterwards do not account for the deviations.
155	Rounding the value preemptively and supplying corrected values to jump-back
156	commands prevents this.
157	"""
158	def __init__(self,	6✔
159	bases: Tuple[NumVal],
160	multiplicities: Tuple[int],
161	offset: NumVal):
162
163	self.bases = tuple(bases)	6✔
164	self.multiplicities = tuple(multiplicities)	6✔
165	self.offset = offset	6✔
166	self.__is_time_or_int = all(isinstance(b,(TimeType,int_type)) for b in bases) and isinstance(offset,(TimeType,int_type))	6✔
167
168	#this is not to circumvent float errors in python, but rounding errors from awg-increment commands.
169	#python float are thereby accurate enough
170	def __call__(self, resolution: Optional[float]) -> Union[NumVal,TimeType]:	6✔
171	#with resolution = None handle TimeType/int case?
172	if resolution is None:	6✔
173	assert self.__is_time_or_int	6✔
174	return sum(b*m for b,m in zip(self.bases,self.multiplicities)) + self.offset	6✔
175	#resolution as float value of granularity of base val.
176	#to avoid conflicts between positive and negative vals from casting
177	#half to even, use abs val
178	return sum(np.sign(b) * round(abs(b) / resolution) * m * resolution for b,m in zip(self.bases,self.multiplicities))\	6✔
179	+ np.sign(self.offset) * round(abs(self.offset) / resolution) * resolution
180
181	def __bool__(self):	6✔
182	#if any value is not zero - this helps for some checks
183	return any(bool(b) for b in self.bases) or bool(self.offset)	6✔
184
185	def __add__(self, other):	6✔
186	# this should happen in the context of an offset being added to it, not the bases being modified.
187	if isinstance(other, (float, int, TimeType)):	6✔
188	return ResolutionDependentValue(self.bases,self.multiplicities,self.offset+other)	6✔
189	return NotImplemented	6✔
190
191	def __radd__(self, other):	6✔
NEW 192	return self.__add__(other)	×
193
194	def __sub__(self, other):	6✔
195	return self.__add__(-other)	6✔
196
197	def __mul__(self, other):	6✔
198	# this should happen when the amplitude is being scaled
199	# multiplicities are not affected
200	if isinstance(other, (float, int, TimeType)):	6✔
201	return ResolutionDependentValue(tuple(bother for b in self.bases),self.multiplicities,self.offsetother)	6✔
NEW 202	return NotImplemented	×
203
204	def __rmul__(self,other):	6✔
NEW 205	return self.__mul__(other)	×
206
207	def __truediv__(self,other):	6✔
208	return self.__mul__(1/other)	6✔
209
210	def __float__(self):	6✔
211	return float(self(resolution=None))	6✔
212
213	def __str__(self):	6✔
214	return f"RDP of {sum(b*m for b,m in zip(self.bases,self.multiplicities)) + self.offset}"	6✔
215
216	def __repr__(self):	6✔
NEW 217	return "RDP("+",".join([f"{k}="+v.__str__() for k,v in vars(self).items()])+")"	×
218
219	def __eq__(self,o):	6✔
220	if not isinstance(o,ResolutionDependentValue):	6✔
NEW 221	return False	×
222	return self.__dict__ == o.__dict__	6✔
223
224	def __hash__(self):	6✔
225	return hash((self.bases,self.offset,self.multiplicities,self.__is_time_or_int))	6✔
226
227
228	#This is a simple dervide class to allow better isinstance checks in the HDAWG driver
229	@dataclass	6✔
230	class DynamicLinearValueStepped(DynamicLinearValue):	6✔
231	step_nesting_level: int	6✔
232	rng: range	6✔
233	reverse: int\|bool	6✔
234
235
236	# TODO: hackedy, hackedy
237	sym_expr.ALLOWED_NUMERIC_SCALAR_TYPES = sym_expr.ALLOWED_NUMERIC_SCALAR_TYPES + (DynamicLinearValue,)	6✔
238
239	# this keeps the simple expression in lambdified results
240	_lambdify_modules.append({'DynamicLinearValue': DynamicLinearValue,	6✔
241	'DynamicLinearValueStepped': DynamicLinearValueStepped})
242
243	RepetitionCount = Union[int, VolatileRepetitionCount, DynamicLinearValue[int]]	6✔
244	HardwareTime = Union[TimeType, DynamicLinearValue[TimeType]]	6✔
245	HardwareVoltage = Union[float, DynamicLinearValue[float]]	6✔

qutech / qupulse / 16672493391

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