# Repository: https://gitlab.com/qblox/packages/software/qblox-scheduler
# Licensed according to the LICENSE file on the main branch
#
# Copyright 2020-2025, Quantify Consortium
# Copyright 2025, Qblox B.V.
"""Helper functions for debugging experiments."""
from __future__ import annotations
import json
import os
import sys
import tempfile
import traceback
import zipfile
from datetime import datetime, timezone
from importlib.metadata import distribution
from importlib.metadata import version as get_version
from pathlib import Path
from typing import TYPE_CHECKING, Any
import numpy as np
from packaging.requirements import Requirement
from qcodes.utils.json_utils import NumpyJSONEncoder
from qblox_scheduler._version import __version__ as __scheduler_version__
from qblox_scheduler.analysis.data_handling import OutputDirectoryManager
from qblox_scheduler.backends.qblox import data
from qblox_scheduler.quantify_utils import snapshot
from qblox_scheduler.schedules.schedule import TimeableSchedule
if TYPE_CHECKING:
from types import TracebackType
from xarray import Dataset
from qblox_scheduler.device_under_test.quantum_device import QuantumDevice
from qblox_scheduler.instrument_coordinator.instrument_coordinator import (
InstrumentCoordinator,
)
from qblox_scheduler.schedule import Schedule
from qblox_scheduler.schedules.schedule import CompiledSchedule
[docs]
def generate_diagnostics_report( # noqa: C901, PLR0915
directory: str | Path | None = None,
quantum_device: QuantumDevice | None = None,
gettable_config: dict[str, Any] | None = None,
schedule: Schedule | TimeableSchedule | None = None,
instrument_coordinator: InstrumentCoordinator | None = None,
compiled_schedule: CompiledSchedule | None | None = None,
acquisition_data: tuple[np.ndarray, ...] | None | None = None,
dataset: Dataset | None = None,
experiment_exception: (
tuple[type[BaseException], BaseException, TracebackType] | tuple[None, None, None] | None
) = None,
initialized: bool = False,
) -> Path:
"""
Generate a report with the current state of an experiment for debugging.
Returns
-------
The Path to the generated zipfile report.
"""
def _flatten_hardware_logs_dict(
hw_logs: dict,
extracted_hw_logs: dict,
prefix: str | None = None,
) -> dict:
for key, value in hw_logs.items():
if isinstance(value, dict):
_flatten_hardware_logs_dict(
hw_logs=value, extracted_hw_logs=extracted_hw_logs, prefix=key
)
else:
extracted_hw_logs[f"{prefix}_{key}" if prefix is not None else f"{key}"] = value
return extracted_hw_logs
def _get_dependency_versions() -> list:
dependencies = [
dependency
for dependency in (distribution("qblox-scheduler").requires or [])
if "extra" not in dependency
]
all_dependency_versions = []
for line in dependencies:
req = Requirement(line)
if req.marker and not req.marker.evaluate():
# marker.evaluate() compares the marker (e.g. the python version) with
# the local environment. If the result is False, the requirement does
# not apply.
continue
dependency = req.name
if dependency == "quantify_core":
dependency = dependency.replace("_", "-")
version = get_version(dependency)
all_dependency_versions.append(f"{dependency}: {version}")
all_dependency_versions.append(f"qblox-scheduler: {__scheduler_version__}")
all_dependency_versions.append(f"python: {sys.version.split(' ')[0]}")
return sorted(all_dependency_versions)
connection_exception = None
hardware_logs = {}
if instrument_coordinator is None:
status = "no instrument coordinator"
elif initialized:
try:
# TODO uncomment when we have a way to retrieve hardware logs qblox os instruments,
# internal jira ticket: BM-155
# OR when it raises a timeout error
# hardware_logs = instrument_coordinator.retrieve_hardware_logs()
pass
except Exception: # noqa: BLE001
connection_exception = sys.exc_info()
if experiment_exception is not None:
status = (
"failed to connect to the hardware"
if connection_exception is not None
else "schedule failed to run"
)
else:
status = (
"failed to retrieve hardware logs"
if connection_exception is not None
else "schedule completed"
)
else:
status = "schedule not yet run"
if directory is None:
directory = OutputDirectoryManager.get_datadir()
timestamp = (
datetime.now(timezone.utc).isoformat(sep="T", timespec="milliseconds").replace("+00:00", "")
)
filename_timestamp = timestamp.replace(":", "").replace("-", "")
report_zipfile = os.path.join(
directory, f"diagnostics_report_scheduler_{filename_timestamp}.zip"
)
with zipfile.ZipFile(
report_zipfile, mode="w", compression=zipfile.ZIP_DEFLATED, compresslevel=9
) as zip_file:
info_json = {"timestamp": timestamp, "status": status}
zip_file.writestr("info.json", json.dumps(info_json))
zip_file.writestr("dependency_versions.json", json.dumps(_get_dependency_versions()))
if gettable_config:
zip_file.writestr(
"gettable.json", json.dumps(gettable_config, cls=NumpyJSONEncoder, indent=4)
)
if quantum_device:
zip_file.writestr("quantum_device.json", quantum_device.to_json())
zip_file.writestr(
"hardware_configuration.json",
json.dumps(
quantum_device.generate_hardware_config(),
cls=NumpyJSONEncoder,
indent=4,
),
)
if schedule is not None:
if isinstance(schedule, TimeableSchedule):
timeable_schedule = schedule
else:
timeable_schedule = schedule._timeable_schedule
if timeable_schedule is not None:
zip_file.writestr("schedule.json", timeable_schedule.to_json())
if compiled_schedule is not None:
zip_file.writestr("compiled_schedule.json", compiled_schedule.to_json())
if dataset:
with tempfile.TemporaryDirectory() as tmpdir:
tmp_path = Path(tmpdir) / "dataset.hdf5"
dataset = data.convert_int_keys_to_strings(dataset)
dataset.to_netcdf(tmp_path, engine="h5netcdf")
zip_file.write(tmp_path, "dataset.hdf5")
if initialized:
zip_file.writestr(
"snapshot.json",
json.dumps(snapshot(), cls=NumpyJSONEncoder, indent=4),
)
if acquisition_data is not None:
data_str = ""
for line in acquisition_data:
data_str += f"{np.array2string(line)},\n"
data_str.rstrip(",\n")
zip_file.writestr("acquisition_data.txt", data_str)
if hardware_logs:
for (
component_name,
component_hardware_logs,
) in hardware_logs.items():
for log_name, hardware_log in _flatten_hardware_logs_dict(
hw_logs=component_hardware_logs, extracted_hw_logs={}
).items():
zip_file.writestr(
os.path.join(component_name, f"{log_name}.txt"),
hardware_log,
)
if experiment_exception is not None:
exception_str = " ".join(traceback.format_exception(*experiment_exception))
zip_file.writestr("error_trace.txt", exception_str)
elif connection_exception is not None:
# No need to print this if there is already an experiment_exception
exception_str = " ".join(traceback.format_exception(*connection_exception))
zip_file.writestr("connection_error_trace.txt", exception_str)
return Path(report_zipfile)
