# Repository: https://gitlab.com/quantify-os/quantify-scheduler
# Licensed according to the LICENCE file on the main branch
"""Schedule helper functions."""
from __future__ import annotations
from itertools import chain
from typing import TYPE_CHECKING, Any, Dict, List, Tuple
import numpy as np
from quantify_scheduler.helpers.collections import make_hash, without
from quantify_scheduler.schedules.schedule import (
AcquisitionMetadata,
CompiledSchedule,
ScheduleBase,
)
if TYPE_CHECKING:
from quantify_scheduler import Operation
[docs]def get_pulse_uuid(pulse_info: Dict[str, Any], excludes: List[str] = None) -> int:
"""
Returns an unique identifier for a pulse.
Parameters
----------
pulse_info
The pulse information dictionary.
Returns
-------
:
The uuid hash.
"""
if excludes is None:
excludes = ["t0"]
return make_hash(without(pulse_info, excludes))
[docs]def get_acq_uuid(acq_info: Dict[str, Any]) -> int:
"""
Returns an unique identifier for a acquisition protocol.
Parameters
----------
acq_info
The acquisition information dictionary.
Returns
-------
:
The uuid hash.
"""
return make_hash(without(acq_info, ["t0", "waveforms"]))
[docs]def get_total_duration(schedule: CompiledSchedule) -> float:
"""
Returns the total schedule duration in seconds.
Parameters
----------
schedule
The schedule.
Returns
-------
:
Duration in seconds.
"""
if len(schedule.schedulables) == 0:
return 0.0
def _get_operation_end(pair: Tuple[int, dict]) -> float:
"""Returns the operations end time in seconds."""
(timeslot_index, _) = pair
return get_operation_end(
schedule,
timeslot_index,
)
operations_ends = map(
_get_operation_end,
enumerate(schedule.schedulables.values()),
)
return max(
operations_ends,
default=0,
)
[docs]def get_operation_start(
schedule: CompiledSchedule,
timeslot_index: int,
) -> float:
"""
Returns the start of an operation in seconds.
Parameters
----------
schedule
timeslot_index
Returns
-------
:
The Operation start time in Seconds.
"""
if len(schedule.schedulables) == 0:
return 0.0
schedulable = list(schedule.schedulables.values())[timeslot_index]
operation = schedule.operations[schedulable["operation_repr"]]
t0: float = schedulable["abs_time"]
pulse_info: dict = (
operation["pulse_info"][0]
if len(operation["pulse_info"]) > 0
else {"t0": -1, "duration": 0}
)
acq_info: dict = (
operation["acquisition_info"][0]
if len(operation["acquisition_info"]) > 0
else {"t0": -1, "duration": 0}
)
if acq_info["t0"] != -1 and acq_info["t0"] < pulse_info["t0"]:
t0 += acq_info["t0"]
elif pulse_info["t0"] >= 0:
t0 += pulse_info["t0"]
return t0
[docs]def get_operation_end(
schedule: CompiledSchedule,
timeslot_index: int,
) -> float:
"""
Returns the end of an operation in seconds.
Parameters
----------
schedule
timeslot_index
Returns
-------
:
The Operation end time in Seconds.
"""
if len(schedule.schedulables) == 0:
return 0.0
schedulable = list(schedule.schedulables.values())[timeslot_index]
operation: Operation = schedule.operations[schedulable["operation_repr"]]
t0: float = schedulable["abs_time"]
return t0 + operation.duration
[docs]def get_port_timeline(
schedule: CompiledSchedule,
) -> Dict[str, Dict[int, List[int]]]:
"""
Returns a new dictionary containing the timeline of
pulses, readout- and acquisition pulses of a port.
Using iterators on this collection enables sorting.
.. code-block::
print(port_timeline_dict)
# { {'q0:mw', {0, [123456789]}},
# ... }
# Sorted items.
print(port_timeline_dict.items())
Parameters
----------
schedule
The schedule.
"""
port_timeline_dict: Dict[str, Dict[int, List[int]]] = {}
# Sort timing constraints based on abs_time and keep the original index.
schedulables_map = dict(
sorted(
map(
lambda pair: (pair[0], pair[1]),
enumerate(schedule.schedulables.values()),
),
key=lambda pair: pair[1]["abs_time"],
)
)
for timeslot_index, schedulable in schedulables_map.items():
operation = schedule.operations[schedulable["operation_repr"]]
abs_time = schedulable["abs_time"]
pulse_info_iter = map(
lambda pulse_info: (get_pulse_uuid(pulse_info), pulse_info),
operation["pulse_info"],
)
acq_info_iter = map(
lambda acq_info: (get_acq_uuid(acq_info), acq_info),
operation["acquisition_info"],
)
# Sort pulses and acquisitions within an operation.
for uuid, info in sorted(
chain(pulse_info_iter, acq_info_iter),
key=lambda pair: abs_time # pylint: disable=cell-var-from-loop
+ pair[1]["t0"],
):
port = str(info["port"])
if port not in port_timeline_dict:
port_timeline_dict[port] = {}
if timeslot_index not in port_timeline_dict[port]:
port_timeline_dict[port][timeslot_index] = []
port_timeline_dict[port][timeslot_index].append(uuid)
return port_timeline_dict
[docs]def get_schedule_time_offset(
schedule: CompiledSchedule,
port_timeline_dict: Dict[str, Dict[int, List[int]]],
) -> float:
"""
Returns the start time in seconds of the first pulse
in the CompiledSchedule. The "None" port containing the Reset
Operation will be ignored.
Parameters
----------
schedule
port_timeline_dict
Returns
-------
:
The operation t0 in seconds.
"""
return min(
map(
lambda port: get_operation_start(
schedule,
timeslot_index=next(iter(port_timeline_dict[port])),
)
if port != "None"
else np.inf,
port_timeline_dict.keys(),
),
default=0,
)
[docs]def get_pulse_info_by_uuid(
schedule: CompiledSchedule,
) -> Dict[int, Dict[str, Any]]:
"""
Returns a lookup dictionary of pulses with its
hash as unique identifiers.
Parameters
----------
schedule
The schedule.
"""
pulseid_pulseinfo_dict: Dict[int, Dict[str, Any]] = {}
for schedulable in schedule.schedulables.values():
operation = schedule.operations[schedulable["operation_repr"]]
for pulse_info in operation["pulse_info"]:
pulse_id = get_pulse_uuid(pulse_info)
if pulse_id in pulseid_pulseinfo_dict:
# Unique pulse info already populated in the dictionary.
continue
pulseid_pulseinfo_dict[pulse_id] = pulse_info
for acq_info in operation["acquisition_info"]:
for pulse_info in acq_info["waveforms"]:
pulse_id = get_pulse_uuid(pulse_info)
if pulse_id in pulseid_pulseinfo_dict:
# Unique pulse info already populated in the dictionary.
continue
pulseid_pulseinfo_dict[pulse_id] = pulse_info
return pulseid_pulseinfo_dict
[docs]def get_acq_info_by_uuid(schedule: CompiledSchedule) -> Dict[int, Dict[str, Any]]:
"""
Returns a lookup dictionary of unique identifiers
of acquisition information.
Parameters
----------
schedule
The schedule.
"""
acqid_acqinfo_dict: Dict[int, Dict[str, Any]] = {}
for schedulable in schedule.schedulables.values():
operation = schedule.operations[schedulable["operation_repr"]]
for acq_info in operation["acquisition_info"]:
acq_id = get_acq_uuid(acq_info)
if acq_id in acqid_acqinfo_dict:
# Unique acquisition info already populated in the dictionary.
continue
acqid_acqinfo_dict[acq_id] = acq_info
return acqid_acqinfo_dict