Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft for parameterized measurement window definition #674

Draft
wants to merge 2 commits into
base: master
Choose a base branch
from
Draft

Draft for parameterized measurement window definition #674

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0794478
Move range stuff in separate module
shumpohl Sep 5, 2022
c007a11
First draft of range base measurement window parameterization
shumpohl Sep 5, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
147 changes: 4 additions & 143 deletions qupulse/pulses/loop_pulse_template.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
from qupulse.pulses.pulse_template import PulseTemplate, ChannelID, AtomicPulseTemplate
from qupulse._program.waveforms import SequenceWaveform as ForLoopWaveform
from qupulse.pulses.measurement import MeasurementDefiner, MeasurementDeclaration
from qupulse.pulses.range import ParametrizedRange, RangeScope

__all__ = ['ForLoopPulseTemplate', 'LoopPulseTemplate', 'LoopIndexNotUsedException']

Expand All @@ -45,54 +46,6 @@ def measurement_names(self) -> Set[str]:
return self.__body.measurement_names


class ParametrizedRange:
"""Like the builtin python range but with parameters."""
def __init__(self, *args, **kwargs):
"""Positional and keyword arguments cannot be mixed.

Args:
*args: Interpreted as ``(start, )`` or ``(start, stop[, step])``
**kwargs: Expected to contain ``start``, ``stop`` and ``step``
Raises:
TypeError: If positional and keyword arguments are mixed
KeyError: If keyword arguments but one of ``start``, ``stop`` or ``step`` is missing
"""
if args and kwargs:
raise TypeError('ParametrizedRange only takes either positional or keyword arguments')
elif kwargs:
start = kwargs['start']
stop = kwargs['stop']
step = kwargs['step']
elif len(args) in (1, 2, 3):
if len(args) == 3:
start, stop, step = args
elif len(args) == 2:
(start, stop), step = args, 1
elif len(args) == 1:
start, (stop,), step = 0, args, 1
else:
raise TypeError('ParametrizedRange expected 1 to 3 arguments, got {}'.format(len(args)))

self.start = ExpressionScalar.make(start)
self.stop = ExpressionScalar.make(stop)
self.step = ExpressionScalar.make(step)

def to_tuple(self) -> Tuple[Any, Any, Any]:
"""Return a simple representation of the range which is useful for comparison and serialization"""
return (self.start.get_serialization_data(),
self.stop.get_serialization_data(),
self.step.get_serialization_data())

def to_range(self, parameters: Mapping[str, Number]) -> range:
return range(checked_int_cast(self.start.evaluate_in_scope(parameters)),
checked_int_cast(self.stop.evaluate_in_scope(parameters)),
checked_int_cast(self.step.evaluate_in_scope(parameters)))

@property
def parameter_names(self) -> Set[str]:
return set(self.start.variables) | set(self.stop.variables) | set(self.step.variables)


class ForLoopPulseTemplate(LoopPulseTemplate, MeasurementDefiner, ParameterConstrainer):
"""This pulse template allows looping through an parametrized integer range and provides the loop index as a
parameter to the body. If you do not need the index in the pulse template, consider using
Expand Down Expand Up @@ -122,18 +75,7 @@ def __init__(self,
MeasurementDefiner.__init__(self, measurements=measurements)
ParameterConstrainer.__init__(self, parameter_constraints=parameter_constraints)

if isinstance(loop_range, ParametrizedRange):
self._loop_range = loop_range
elif isinstance(loop_range, (int, str)):
self._loop_range = ParametrizedRange(loop_range)
elif isinstance(loop_range, (tuple, list)):
self._loop_range = ParametrizedRange(*loop_range)
elif isinstance(loop_range, range):
self._loop_range = ParametrizedRange(start=loop_range.start,
stop=loop_range.stop,
step=loop_range.step)
else:
raise ValueError('loop_range is not valid')
self._loop_range = ParametrizedRange.from_range_like(loop_range)

if not loop_index.isidentifier():
raise InvalidParameterNameException(loop_index)
Expand Down Expand Up @@ -198,15 +140,8 @@ def _body_scope_generator(self, scope: Scope, forward=True) -> Iterator[Scope]:
loop_range = loop_range if forward else reversed(loop_range)
loop_index_name = self._loop_index

get_for_loop_scope = _get_for_loop_scope

for loop_index_value in loop_range:
try:
yield get_for_loop_scope(scope, loop_index_name, loop_index_value)
except TypeError:
# we cannot hash the scope so we will not try anymore
get_for_loop_scope = _ForLoopScope
yield get_for_loop_scope(scope, loop_index_name, loop_index_value)
yield _ForLoopScope(scope, loop_index_name, loop_index_value)

def _internal_create_program(self, *,
scope: Scope,
Expand Down Expand Up @@ -301,78 +236,4 @@ def __str__(self) -> str:
self.body_parameter_names)


class _ForLoopScope(Scope):
__slots__ = ('_index_name', '_index_value', '_inner')

def __init__(self, inner: Scope, index_name: str, index_value: int):
super().__init__()
self._inner = inner
self._index_name = index_name
self._index_value = index_value

def get_volatile_parameters(self) -> FrozenMapping[str, Expression]:
inner_volatile = self._inner.get_volatile_parameters()

if self._index_name in inner_volatile:
# TODO: use delete method of frozendict
index_name = self._index_name
return FrozenDict((name, value) for name, value in inner_volatile.items() if name != index_name)
else:
return inner_volatile

def __hash__(self):
return hash((self._inner, self._index_name, self._index_value))

def __eq__(self, other: '_ForLoopScope'):
try:
return (self._index_name == other._index_name
and self._index_value == other._index_value
and self._inner == other._inner)
except AttributeError:
return False

def __contains__(self, item):
return item == self._index_name or item in self._inner

def get_parameter(self, parameter_name: str) -> Number:
if parameter_name == self._index_name:
return self._index_value
else:
return self._inner.get_parameter(parameter_name)

__getitem__ = get_parameter

def change_constants(self, new_constants: Mapping[str, Number]) -> 'Scope':
return _get_for_loop_scope(self._inner.change_constants(new_constants), self._index_name, self._index_value)

def __len__(self) -> int:
return len(self._inner) + int(self._index_name not in self._inner)

def __iter__(self) -> Iterator:
if self._index_name in self._inner:
return iter(self._inner)
else:
return itertools.chain(self._inner, (self._index_name,))

def as_dict(self) -> FrozenMapping[str, Number]:
if self._as_dict is None:
self._as_dict = FrozenDict({**self._inner.as_dict(), self._index_name: self._index_value})
return self._as_dict

def keys(self):
return self.as_dict().keys()

def items(self):
return self.as_dict().items()

def values(self):
return self.as_dict().values()

def __repr__(self):
return f'{type(self)}(inner={self._inner!r}, index_name={self._index_name!r}, ' \
f'index_value={self._index_value!r})'


@functools.lru_cache(maxsize=10**6)
def _get_for_loop_scope(inner: Scope, index_name: str, index_value: int) -> Scope:
return _ForLoopScope(inner, index_name, index_value)
_ForLoopScope = RangeScope
75 changes: 61 additions & 14 deletions qupulse/pulses/measurement.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,22 +2,39 @@
from numbers import Real
import itertools

from qupulse.expressions import Expression
from qupulse.expressions import ExpressionScalar, ExpressionLike
from qupulse.utils.types import MeasurementWindow
from qupulse.parameter_scope import Scope
from qupulse.pulses.range import ParametrizedRange, RangeLike, RangeScope

MeasurementDeclaration = Tuple[str, Union[Expression, str, Real], Union[Expression, str, Real]]
MeasurementRangeDeclaration = Tuple[str, ExpressionLike, ExpressionLike, Tuple[str, RangeLike]]

MeasurementDeclaration = Union[
Tuple[str, ExpressionLike, ExpressionLike],
MeasurementRangeDeclaration
]


class MeasurementDefiner:
def __init__(self, measurements: Optional[List[MeasurementDeclaration]]):
if measurements is None:
self._measurement_windows = []
else:
self._measurement_windows = [(name,
begin if isinstance(begin, Expression) else Expression(begin),
length if isinstance(length, Expression) else Expression(length))
for name, begin, length in measurements]
self._measurement_windows: Tuple[Tuple[str, ExpressionScalar, ExpressionScalar], ...] = ()
self._measurement_ranges: Tuple[Tuple[str, ExpressionScalar, ExpressionScalar, Tuple[str, RangeLike]], ...] = ()

if measurements is not None:
measurement_windows = []
measurement_ranges = []

for declaration in measurements:
if len(declaration) == 3:
name, begin, length = declaration
measurement_windows.append((name, ExpressionScalar(begin), ExpressionScalar(length)))
else:
name, begin, length, (idx_name, param_range) = declaration
measurement_ranges.append((name, ExpressionScalar(begin), ExpressionScalar(length),
(idx_name, ParametrizedRange.from_range_like(param_range))))

self._measurement_windows = tuple(measurement_windows)
self._measurement_ranges = tuple(measurement_ranges)
for _, _, length in self._measurement_windows:
if (length < 0) is True:
raise ValueError('Measurement window length may not be negative')
Expand Down Expand Up @@ -59,6 +76,24 @@ def get_measurement_windows(self,
begin_val,
length_val)
)

for name, begin, length, (idx_name, idx_range) in self._measurement_ranges:
name = measurement_mapping[name]
if name is None:
continue

for idx_val in idx_range.to_range(parameters):
scope = RangeScope(parameters, idx_name, idx_val)

begin_val = begin.evaluate_in_scope(scope)
length_val = length.evaluate_in_scope(scope)

resulting_windows.append(
(name,
begin_val,
length_val)
)

return resulting_windows

@property
Expand All @@ -72,13 +107,25 @@ def measurement_parameters(self) -> Set[str]:
@property
def measurement_declarations(self) -> List[MeasurementDeclaration]:
"""Return the measurements that are directly declared on `self`. Does _not_ visit eventual child objects."""
return [(name,
begin.original_expression,
length.original_expression)
for name, begin, length in self._measurement_windows]
measurements = []
measurements.extend((name,
begin.original_expression,
length.original_expression)
for name, begin, length in self._measurement_windows)
measurements.extend((name, begin.original_expression, length.original_expression,
(idx_name, idx_range.to_tuple()))
for name, begin, length, (idx_name, idx_range) in self._measurement_ranges)
return measurements

@property
def measurement_names(self) -> Set[str]:
"""Return the names of measurements that are directly declared on `self`.
Does _not_ visit eventual child objects."""
return {name for name, *_ in self._measurement_windows}
return {name for name, *_ in itertools.chain(self._measurement_windows, self._measurement_ranges)}

def __hash__(self):
return hash((self._measurement_windows, self._measurement_ranges))

def __eq__(self, other):
return (self._measurement_windows == getattr(other, '_measurement_windows', None) and
self._measurement_ranges == getattr(other, '_measurement_ranges', None))
Loading