# Repository: https://gitlab.com/quantify-os/quantify-scheduler
# Licensed according to the LICENCE file on the main branch
r"""
This module represents the Q-Profile quantum control electronics profiler.
Profiling of the control electronics is enabled by using the
:class:`ProfiledScheduleGettable` in place of
:class:`~.ScheduleGettable`.
.. note::
The :class:`ProfiledScheduleGettable` is currently only tested to support Qblox hardware.
"""
import json
import os
import time
import matplotlib.pyplot as plt
import numpy as np
from qcodes.instrument.instrument import Instrument
from xarray import Dataset
from quantify_scheduler.gettables import ScheduleGettable
from quantify_scheduler.instrument_coordinator.instrument_coordinator import (
InstrumentCoordinator,
)
[docs]
def profiler(func):
"""
Decorator that reports the execution time of the decorated function
and stores this in ``ProfiledInstrumentCoordinator.profile``.
Parameters
----------
func: Callable
Target function to be profiled.
"""
def wrap(self, *args, **kwargs):
start = time.time()
result = func(self, *args, **kwargs)
end = time.time()
if func.__name__ not in self.profile:
self.profile[func.__name__] = []
self.profile[func.__name__].append(end - start)
return result
return wrap
[docs]
class ProfiledInstrumentCoordinator(InstrumentCoordinator):
"""
Subclass of :class:`~.InstrumentCoordinator` that implements a profiling tool to log
timing results. Time results are stored in ``ProfiledInstrumentCoordinator.profile``.
:class:`ProfiledInstrumentCoordinator` is set up to be used when using
:class:`ProfiledScheduleGettable`, code example:
.. code-block:: python
ic = InstrumentCoordinator(name="instrument_coordinator")
quantum_device = QuantumDevice(name="quantum_device")
quantum_device.instr_instrument_coordinator(ic.name)
profiled_gettable = ProfiledScheduleGettable(
quantum_device=quantum_device,
schedule_function=...,
schedule_kwargs=...,
)
profiled_gettable.initialize()
profiled_ic = (
profiled_gettable.quantum_device.instr_instrument_coordinator.get_instr()
)
Parameters
----------
name: str
Name of :class:`ProfiledInstrumentCoordinator` instance.
parent_ic: InstrumentCoordinator
Original :class:`~.InstrumentCoordinator`.
"""
def __init__(self, name: str, parent_ic: InstrumentCoordinator):
[docs]
self.profile = {"schedule": []}
super().__init__(name, add_default_generic_icc=False)
[docs]
self.parent_ic = parent_ic
@profiler
[docs]
def add_component(
self,
component,
) -> None:
self.parent_ic.add_component(component)
@profiler
[docs]
def prepare(
self,
compiled_schedule,
) -> None:
self.profile["schedule"].append(compiled_schedule.get_schedule_duration())
self.parent_ic.prepare(compiled_schedule)
@profiler
[docs]
def start(self):
self.parent_ic.start()
@profiler
[docs]
def stop(self, allow_failure=False):
self.parent_ic.stop()
@profiler
[docs]
def retrieve_acquisition(self) -> Dataset:
return self.parent_ic.retrieve_acquisition()
@profiler
[docs]
def wait_done(self, timeout_sec: int = 10):
self.parent_ic.wait_done(timeout_sec)
[docs]
class ProfiledScheduleGettable(ScheduleGettable):
"""
To be used in place of :class:`~.ScheduleGettable` to enable profiling of the
compilation. Logged execution times can be read from ``self.profile``, and plotted
via :func:`plot_profile`.
"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# Overwrite linked IC to a profiled IC
[docs]
self.instr_coordinator: InstrumentCoordinator = ( # type: ignore
self.quantum_device.instr_instrument_coordinator.get_instr()
)
[docs]
self.profiled_instr_coordinator = ProfiledInstrumentCoordinator(
name="profiled_ic", parent_ic=self.instr_coordinator
)
self.quantum_device.instr_instrument_coordinator(
self.profiled_instr_coordinator.name
)
@profiler
[docs]
def _compile(self, sched):
"""Overwrite compile step for profiling."""
super()._compile(sched)
[docs]
def close(self):
"""Cleanup new profiling instruments to avoid future conflicts."""
self.profile.update(self.profiled_instr_coordinator.profile)
self.quantum_device.instr_instrument_coordinator(self.instr_coordinator.name)
prof_ic = Instrument.find_instrument("profiled_ic")
Instrument.close(prof_ic)
[docs]
def log_profile(
self,
obj=None,
path="profiling_logs",
filename=None,
indent: int = 4,
separators=None,
):
"""Store profiling logs to json file."""
if not obj:
obj = self.profile
if not separators:
separators = (",", ": ")
if filename:
if not os.path.exists(path):
os.makedirs(path)
file_path = os.path.join(path, filename)
with open(file_path, "w", encoding="utf-8") as file:
json.dump(obj, file, indent=indent, separators=separators)
return self.profile
[docs]
def plot_profile(self, path=None, filename="average_runtimes.pdf"):
"""Create barplot of accumulated profiling data."""
profile = self.profile
time_ax = list(profile.keys())
num_keys = len(time_ax)
x_pos = np.arange(num_keys)
means = [np.mean(x) for x in profile.values()]
error = [np.std(x) for x in profile.values()]
fig, ax = plt.subplots(figsize=(9, 6))
color = ["r", "b", "c", "m", "k", "g", "y"][:num_keys]
ax.bar(x_pos, means, yerr=error, color=color)
ax.bar(num_keys, means[0], color=color[0])
for i in range(1, num_keys):
ax.bar(num_keys, means[i], color=color[i], bottom=sum(means[:i]))
time_ax.append("total")
ax.set_xticks(np.append(x_pos, num_keys))
ax.set_xticklabels(time_ax)
self.plot = (fig, ax)
plt.ylabel("runtime [s]")
plt.title("Average runtimes")
file_path = os.path.join(path, filename) if path else filename
fig.savefig(file_path)
