# Repository: https://gitlab.com/quantify-os/quantify-scheduler
# Licensed according to the LICENCE file on the main branch
"""Module containing quantify JSON utilities."""
from __future__ import annotations
import functools
import json
import pathlib
import sys
import warnings
from types import ModuleType
from typing import Any, Callable, Dict, List, Type, Union
import fastjsonschema
import numpy as np
from qcodes import Instrument
from quantify_scheduler import enums
from quantify_scheduler.helpers import inspect as inspect_helpers
from quantify_scheduler.helpers.importers import import_python_object_from_string
[docs]
current_python_version = sys.version_info
[docs]
lru_cache = functools.lru_cache(maxsize=200)
[docs]
DEFAULT_TYPES = [
complex,
float,
int,
bool,
str,
np.ndarray,
np.complex128,
np.int32,
np.uint32,
np.int64,
]
[docs]
def validate_json(data, schema):
"""Validate schema using jsonschema-rs."""
return fastjsonschema.validate(schema, data)
[docs]
def load_json_schema(relative_to: Union[str, pathlib.Path], filename: str):
"""
Load a JSON schema from file. Expects a 'schemas' directory in the same directory
as ``relative_to``.
.. tip::
Typical usage of the form
``schema = load_json_schema(__file__, 'definition.json')``
Parameters
----------
relative_to
the file to begin searching from
filename
the JSON file to load
Returns
-------
dict
the schema
"""
path = pathlib.Path(relative_to).resolve().parent.joinpath("schemas", filename)
with path.open(mode="r", encoding="utf-8") as file:
return json.load(file)
@lru_cache
[docs]
def load_json_validator(
relative_to: Union[str, pathlib.Path], filename: str
) -> Callable:
"""
Load a JSON validator from file. Expects a 'schemas' directory in the same directory
as ``relative_to``.
Parameters
----------
relative_to
the file to begin searching from
filename
the JSON file to load
Returns
-------
Callable
The validator
"""
definition = load_json_schema(relative_to, filename)
validator = fastjsonschema.compile(definition, handlers={}, formats={})
return validator
[docs]
class UnknownDeserializationTypeError(Exception):
"""Raised when an unknown deserialization type is encountered."""
[docs]
class JSONSchemaValMixin: # pylint: disable=too-few-public-methods
"""
A mixin that adds validation utilities to classes that have
a data attribute like a :class:`UserDict` based on JSONSchema.
This requires the class to have a class variable "schema_filename"
"""
@classmethod
[docs]
def is_valid(cls, object_to_be_validated) -> bool:
"""
Checks if the object is valid according to its schema.
Raises
------
fastjsonschema.JsonSchemaException
if the data is invalid
Returns
-------
:
"""
validator_method = load_json_validator(__file__, cls.schema_filename)
validator_method(object_to_be_validated.data)
return True # if no exception was raised during validation
[docs]
class SchedulerJSONDecoder(json.JSONDecoder):
"""
The Quantify Scheduler JSONDecoder.
The SchedulerJSONDecoder is used to convert a string with JSON content into
instances of classes in quantify-scheduler.
For a few types, :data:`~.DEFAULT_TYPES` contains the mapping from type name to the
python object. This dictionary can be expanded with classes from modules specified
in the keyword argument ``modules``.
Classes not contained in :data:`~.DEFAULT_TYPES` by default must implement
``__getstate__``, such that it returns a dictionary containing at least the keys
``"deserialization_type"`` and ``"data"``, and ``__setstate__``, which should be
able to parse the data from ``__getstate__``.
The value of ``"deserialization_type"`` must be either the name of the class
specified in :data:`~.DEFAULT_TYPES` or the fully qualified name of the class, which
can be obtained from
:func:`~quantify_scheduler.helpers.importers.export_python_object_to_path_string`.
Keyword Arguments
-----------------
modules : List[ModuleType], *optional*
A list of custom modules containing serializable classes, by default []
""" # noqa: D416
[docs]
_classes: Dict[str, Type[Any]]
def __init__(self, *args, **kwargs) -> None:
extended_modules: List[ModuleType] = kwargs.pop("modules", [])
invalid_modules = list(
filter(lambda o: not isinstance(o, ModuleType), extended_modules)
)
if invalid_modules:
raise ValueError(
f"Attempting to create a Schedule decoder class SchedulerJSONDecoder. "
f"The following modules provided are not an instance of the ModuleType:"
f" {invalid_modules} ."
)
super().__init__(
object_hook=self.custom_object_hook,
*args,
**kwargs,
)
self._classes = inspect_helpers.get_classes(*[enums, *extended_modules])
self._classes.update({t.__name__: t for t in DEFAULT_TYPES})
[docs]
def decode_dict(
self, obj: Dict[str, Any]
) -> Union[Dict[str, Any], np.ndarray, type]:
"""
Returns the deserialized JSON dictionary.
Parameters
----------
obj
The dictionary to deserialize.
Returns
-------
:
The deserialized result.
"""
# If "deserialization_type" is present in `obj` it means the object was
# serialized using `__getstate__` and should be deserialized using
# `__setstate__`.
if "deserialization_type" in obj:
try:
class_type = self._get_type_from_string(obj["deserialization_type"])
except UnknownDeserializationTypeError as exc:
raise UnknownDeserializationTypeError(
f"Object '{obj}' cannot be deserialized to type '{obj['deserialization_type']}'"
) from exc
if "mode" in obj and obj["mode"] == "__init__":
if class_type == np.ndarray:
return np.array(obj["data"])
if issubclass(class_type, Instrument):
return class_type(**obj["data"])
return class_type(obj["data"])
if "mode" in obj and obj["mode"] == "type":
return class_type
new_obj = class_type.__new__(class_type)
new_obj.__setstate__(obj)
return new_obj
return obj
[docs]
def custom_object_hook(self, obj: object) -> object:
"""
The ``object_hook`` hook will be called with the result of every JSON object
decoded and its return value will be used in place of the given ``dict``.
Parameters
----------
obj
A pair of JSON objects.
Returns
-------
:
The deserialized result.
"""
if isinstance(obj, dict):
return self.decode_dict(obj)
return obj
[docs]
def _get_type_from_string(self, deserialization_type: str) -> Type:
"""
Get the python type based on the description string.
The following methods are tried, in order:
1. Try to find the string in :data:`~.DEFAULT_TYPES` or the extended modules
passed to this class' initializer.
2. Try to import the type. This works only if ``deserialization_type`` is
formatted as a dot-separated path to the type. E.g.
``quantify_scheduler.json_utils.SchedulerJSONDecoder``.
3. (deprecated) Try to find the class by its ``__name__`` in a predefined
selection of types present in ``quantify_scheduler``.
Parameters
----------
deserialization_type
Description of a type.
Raises
------
UnknownDeserializationTypeError
If the type cannot be found by any of the methods described.
Returns
-------
Type
The ``Type`` found.
"""
try:
return self._classes[deserialization_type]
except KeyError:
pass
try:
return import_python_object_from_string(deserialization_type)
except (AttributeError, ModuleNotFoundError, ValueError):
pass
try:
return _get_type_from_string_deprecated(deserialization_type)
except KeyError:
raise UnknownDeserializationTypeError(
f"Type '{deserialization_type}' is not a known deserialization type."
)
[docs]
def _get_type_from_string_deprecated(deserialization_type: str) -> Type:
# Use local import to void Error('Operation' from partially initialized module
# 'quantify_scheduler')
# pylint: disable=import-outside-toplevel
from quantify_scheduler import resources
from quantify_scheduler.backends.qblox.operations.stitched_pulse import (
StitchedPulse as QbloxStitchedPulse,
)
from quantify_scheduler.device_under_test import transmon_element
from quantify_scheduler.device_under_test.composite_square_edge import (
CompositeSquareEdge,
)
from quantify_scheduler.device_under_test.quantum_device import QuantumDevice
from quantify_scheduler.operations import (
acquisition_library,
gate_library,
nv_native_library,
operation,
pulse_library,
shared_native_library,
)
from quantify_scheduler.schedules.schedule import AcquisitionMetadata, Schedulable
classes = inspect_helpers.get_classes(
operation,
transmon_element,
acquisition_library,
gate_library,
pulse_library,
nv_native_library,
shared_native_library,
resources,
)
classes.update(
{
c.__name__: c
for c in [
AcquisitionMetadata,
Schedulable,
QuantumDevice,
CompositeSquareEdge,
]
}
)
classes.update({"StitchedPulse": QbloxStitchedPulse})
class_type = classes[deserialization_type]
# Only warn if we succeed
warnings.warn(
"Having only the class name as the deserialization type is deprecated "
"and this feature will be removed in quantify-scheduler >= 0.20.0. "
"Please re-serialize the object to use the fully qualified class name.",
FutureWarning,
)
return class_type
[docs]
class SchedulerJSONEncoder(json.JSONEncoder):
"""
Custom JSONEncoder which encodes a Quantify Scheduler object into a JSON file format
string.
"""
[docs]
def default(self, o):
"""
Overloads the json.JSONEncoder default method that returns a serializable
object. It will try 3 different serialization methods which are, in order,
check if the object is to be serialized to a string using repr. If not, try
to use ``__getstate__``. Finally, try to serialize the ``__dict__`` property.
"""
if hasattr(o, "__getstate__"):
return o.__getstate__()
if isinstance(o, (complex, np.int32, np.complex128, np.int64, enums.BinMode)):
return {
"deserialization_type": type(o).__name__,
"mode": "__init__",
"data": str(o),
}
if isinstance(o, (np.ndarray,)):
return {
"deserialization_type": type(o).__name__,
"mode": "__init__",
"data": list(o),
}
if o in DEFAULT_TYPES:
return {"deserialization_type": o.__name__, "mode": "type"}
if hasattr(o, "__dict__"):
return o.__dict__
# Let the base class default method raise the TypeError
return json.JSONEncoder.default(self, o)
