16686887835

Committed 01 Aug 2025 11:06PM UTC coverage: 88.764%. First build

Build # 16686887835

Build Type

Pull #917

github

Committed by

web-flow

Commit Message

Merge acf70991c into a8a829612

Pull Request Pull Request #917: Experiment followup dlv

Run Details

139 of 146 new or added lines in 2 files covered. (95.21%)

18920 of 21315 relevant lines covered (88.76%)

5.33 hits per line

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

91.8

/qupulse/program/values.py

"""Runtime variable value implementations."""

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

import numpy as np

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


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


@dataclass(frozen=True)
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)
        immutable = MappingProxyType(dict(self.factors))
        object.__setattr__(self, 'factors', immutable)
        
    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]

@dataclass(frozen=True)
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.
    """
    
    bases: Tuple[NumVal, ...]
    multiplicities: Tuple[int, ...]
    offset: NumVal
    __is_time_or_int: bool = field(init=False, repr=False)
    
    def __post_init__(self):

        flag = all(isinstance(b,(TimeType,int_type)) for b in self.bases)\
            and isinstance(self.offset,(TimeType,int_type))
        object.__setattr__(self, '_ResolutionDependentValue__is_time_or_int', flag)

    #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(frozen=True)
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,
                          'mappingproxy': MappingProxyType})

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

qutech / qupulse / 16686887835

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