16671212783

Committed 01 Aug 2025 09:13AM UTC coverage: 88.765%. First build

Build # 16671212783

Build Type

Pull #917

github

Committed by

web-flow

Commit Message

Merge 5ffee0f7b into a8a829612

Pull Request Pull Request #917: Experiment followup dlv

Run Details

119 of 123 new or added lines in 2 files covered. (96.75%)

18922 of 21317 relevant lines covered (88.76%)

5.33 hits per line

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

91.94

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

_COMPARATORS = {"__lt__": operator.lt, "__le__": operator.le, "__gt__": operator.gt,
    "__ge__": operator.ge, "__eq__": operator.eq, "__ne__": operator.ne,}


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

qutech / qupulse / 16671212783

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