Skip to content

Commit

Permalink
pad_to as single waveform
Browse files Browse the repository at this point in the history
  • Loading branch information
Nomos11 committed Sep 21, 2024
1 parent 4325ea3 commit 74d5dc8
Show file tree
Hide file tree
Showing 3 changed files with 143 additions and 9 deletions.
16 changes: 16 additions & 0 deletions qupulse/program/linspace.py
Original file line number Diff line number Diff line change
Expand Up @@ -743,6 +743,22 @@ def with_sequence(self,
yield self

def new_subprogram(self, global_transformation: 'Transformation' = None) -> ContextManager['ProgramBuilder']:

inner_builder = LinSpaceBuilder(self._to_stepping_repeat,self._play_marker_when_constant)
yield inner_builder
inner_program = inner_builder.to_program()

# if inner_program is not None:

# # measurements = [(name, begin, length)
# # for name, (begins, lengths) in inner_program.get_measurement_windows().items()
# # for begin, length in zip(begins, lengths)]
# # self._top.add_measurements(measurements)
# waveform = to_waveform(inner_program,self._idx_to_name)
# if global_transformation is not None:
# waveform = TransformingWaveform.from_transformation(waveform, global_transformation)
# self.play_arbitrary_waveform(waveform)

raise NotImplementedError('Not implemented yet (postponed)')

def with_iteration(self, index_name: str, rng: range,
Expand Down
34 changes: 25 additions & 9 deletions qupulse/pulses/pulse_template.py
Original file line number Diff line number Diff line change
Expand Up @@ -376,7 +376,7 @@ def with_appended(self, *appended: 'PulseTemplate'):
return self

def pad_to(self, to_new_duration: Union[ExpressionLike, Callable[[Expression], ExpressionLike]],
pt_kwargs: Mapping[str, Any] = None) -> 'PulseTemplate':
pt_kwargs: Mapping[str, Any] = {}) -> 'PulseTemplate':
"""Pad this pulse template to the given duration.
The target duration can be numeric, symbolic or a callable that returns a new duration from the current
duration.
Expand All @@ -398,23 +398,39 @@ def pad_to(self, to_new_duration: Union[ExpressionLike, Callable[[Expression], E
pt_kwargs: Keyword arguments for the newly created sequence pulse template.
Returns:
A pulse template that has the duration given by ``to_new_duration``. It can be ``self`` if the duration is
already as required. It is never ``self`` if ``pt_kwargs`` is non-empty.
A pulse template that has the duration given by ``to_new_duration``.
self if ConstantPT,
SingleWFTimeExtensionPulseTemplate otherwise.
"""
from qupulse.pulses import ConstantPT, SequencePT
from qupulse.pulses.time_extension_pulse_template import SingleWFTimeExtensionPulseTemplate
current_duration = self.duration
if callable(to_new_duration):
new_duration = to_new_duration(current_duration)
else:
new_duration = ExpressionScalar(to_new_duration)
pad_duration = new_duration - current_duration
if not pt_kwargs and pad_duration == 0:

# if not pt_kwargs and pad_duration == 0:
# return self

#shortcut
if isinstance(self,ConstantPT):
if pt_kwargs:
raise NotImplementedError()
self._duration = new_duration
return self
pad_pt = ConstantPT(pad_duration, self.final_values)
if pt_kwargs:
return SequencePT(self, pad_pt, **pt_kwargs)
else:
return self @ pad_pt

return SingleWFTimeExtensionPulseTemplate(self, new_duration, **pt_kwargs)

# pad_duration = new_duration - current_duration
# if not pt_kwargs and pad_duration == 0:
# return self
# pad_pt = ConstantPT(pad_duration, self.final_values)
# if pt_kwargs:
# return SequencePT(self, pad_pt, **pt_kwargs)
# else:
# return self @ pad_pt

def __format__(self, format_spec: str):
if format_spec == '':
Expand Down
102 changes: 102 additions & 0 deletions qupulse/pulses/time_extension_pulse_template.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,102 @@
from numbers import Real
from typing import Dict, Optional, Set, Union, List, Iterable, Any

from qupulse import ChannelID
from qupulse.parameter_scope import Scope
from qupulse.pulses.pulse_template import PulseTemplate, AtomicPulseTemplate
from qupulse.pulses.constant_pulse_template import ConstantPulseTemplate as ConstantPT
from qupulse.expressions import ExpressionLike, ExpressionScalar
from qupulse._program.waveforms import ConstantWaveform
from qupulse.program import ProgramBuilder
from qupulse.pulses.parameters import ConstraintLike
from qupulse.pulses.measurement import MeasurementDeclaration
from qupulse.serialization import Serializer, PulseRegistryType
from qupulse.program.waveforms import SequenceWaveform


def _evaluate_expression_dict(expression_dict: Dict[str, ExpressionScalar], scope: Scope) -> Dict[str, float]:
return {ch: value.evaluate_in_scope(scope)
for ch, value in expression_dict.items()}


class SingleWFTimeExtensionPulseTemplate(AtomicPulseTemplate):
"""Extend the given pulse template with a constant suffix.
"""

def __init__(self,
main_pt: PulseTemplate,
new_duration: Union[str, ExpressionScalar],
identifier: Optional[str] = None,
*,
measurements: Optional[List[MeasurementDeclaration]]=None,
registry: PulseRegistryType=None) -> None:

AtomicPulseTemplate.__init__(self, identifier=identifier, measurements=measurements)

self.__main_pt = main_pt
self.__pad_pt = ConstantPT(new_duration-main_pt.duration, self.final_values)
self._duration = ExpressionScalar.make(new_duration)

self._register(registry=registry)

@property
def parameter_names(self) -> Set[str]:
return self.__main_pt.parameter_names

@property
def duration(self) -> ExpressionScalar:
"""An expression for the duration of this PulseTemplate."""
return self._duration

@property
def defined_channels(self) -> Set[ChannelID]:
return self.__main_pt.defined_channels

@property
def integral(self) -> Dict[ChannelID, ExpressionScalar]:

unextended = self.__main_pt.integral

return {ch: unextended_ch + (self.duration-self.__main_pt.duration)*self.__main_pt.final_values[ch] \
for ch,unextended_ch in unextended.items()}

@property
def initial_values(self) -> Dict[ChannelID, ExpressionScalar]:
return self.__main_pt.initial_values

@property
def final_values(self) -> Dict[ChannelID, ExpressionScalar]:
return self.__main_pt.final_values

def get_serialization_data(self, serializer: Optional[Serializer]=None) -> Dict[str, Any]:
if serializer is not None:
raise NotImplementedError("SingleWFTimeExtensionPulseTemplate does not implement legacy serialization.")
data = super().get_serialization_data(serializer)
data['main_pt'] = self.__main_pt
data['new_duration'] = self.duration
data['measurements']: self.measurement_declarations

return data

@classmethod
def deserialize(cls,
serializer: Optional[Serializer]=None, # compatibility to old serialization routines, deprecated
**kwargs) -> 'SingleWFTimeExtensionPulseTemplate':
main_pt = kwargs['main_pt']
new_duration = kwargs['new_duration']
del kwargs['main_pt']
del kwargs['new_duration']

if serializer: # compatibility to old serialization routines, deprecated
raise NotImplementedError()

return cls(main_pt,new_duration,**kwargs)

def build_waveform(self,
parameters: Dict[str, Real],
channel_mapping: Dict[ChannelID, ChannelID]) -> SequenceWaveform:
return SequenceWaveform.from_sequence(
[wf for sub_template in [self.__main_pt,self.__pad_pt]
if (wf:=sub_template.build_waveform(parameters, channel_mapping=channel_mapping)) is not None])


0 comments on commit 74d5dc8

Please sign in to comment.