Saving and loading calibrations data

qiskit_experiments/calibration_management/calibrations.py

@@ -12,14 +12,14 @@
 
 """Class to store and manage the results of calibration experiments."""
 
+import warnings
 import os
-from collections import defaultdict
+from collections import defaultdict, Counter
 from datetime import datetime, timezone
 from typing import Any, Dict, Set, Tuple, Union, List, Optional
 import csv
 import dataclasses
 import json
-import warnings
 import rustworkx as rx
 
 from qiskit.pulse import (
@@ -35,6 +35,7 @@
 from qiskit.pulse.channels import PulseChannel
 from qiskit.circuit import Parameter, ParameterExpression
 from qiskit.providers.backend import Backend
+from qiskit.utils.deprecation import deprecate_func, deprecate_arg
 
 from qiskit_experiments.exceptions import CalibrationError
 from qiskit_experiments.calibration_management.basis_gate_library import BasisGateLibrary
@@ -51,7 +52,7 @@
     ParameterValueType,
     ScheduleKey,
 )
-from qiskit_experiments.framework import BackendData
+from qiskit_experiments.framework import BackendData, ExperimentEncoder, ExperimentDecoder
 
 
 class Calibrations:
@@ -1352,30 +1353,31 @@ def _append_to_list(
         value_dict["date_time"] = value_dict["date_time"].strftime("%Y-%m-%d %H:%M:%S.%f%z")
         data.append(value_dict)
 
+    @deprecate_arg(
+        name="file_type",
+        since="0.6",
+        additional_msg="Full calibration saving is now supported in json format. csv is deprecated.",
+        package_name="qiskit-experiments",
+        predicate=lambda file_type: file_type == "csv",
+    )
     def save(
         self,
-        file_type: str = "csv",
+        file_type: str = "json",
         folder: str = None,
         overwrite: bool = False,
         file_prefix: str = "",
         most_recent_only: bool = False,
     ):
         """Save the parameterized schedules and parameter value.
 
-        The schedules and parameter values can be stored in csv files. This method creates
-        three files:
+        .. note::
 
-        * parameter_config.csv: This file stores a table of parameters which indicates
-          which parameters appear in which schedules.
-        * parameter_values.csv: This file stores the values of the calibrated parameters.
-        * schedules.csv: This file stores the parameterized schedules.
-
-        Warning:
-            Schedule blocks will only be saved in string format and can therefore not be
-            reloaded and must instead be rebuilt.
+            Full round-trip serialization of a :class:`.Calibrations` instance
+            is only supported in JSON format.
+            This may be extended to other file formats in future version.
 
         Args:
-            file_type: The type of file to which to save. By default this is a csv.
+            file_type: The type of file to which to save. By default, this is a json.
                 Other file types may be supported in the future.
             folder: The folder in which to save the calibrations.
             overwrite: If the files already exist then they will not be overwritten
@@ -1388,43 +1390,56 @@ def save(
         Raises:
             CalibrationError: If the files exist and overwrite is not set to True.
         """
-        warnings.warn("Schedules are only saved in text format. They cannot be re-loaded.")
-
         cwd = os.getcwd()
         if folder:
             os.chdir(folder)
 
-        parameter_config_file = file_prefix + "parameter_config.csv"
-        parameter_value_file = file_prefix + "parameter_values.csv"
-        schedule_file = file_prefix + "schedules.csv"
+        if file_type == "json":
+            from .save_utils import calibrations_to_dict
 
-        if os.path.isfile(parameter_config_file) and not overwrite:
-            raise CalibrationError(
-                f"{parameter_config_file} already exists. Set overwrite to True."
-            )
+            file_path = file_prefix + ".json"
+            if os.path.isfile(file_path) and not overwrite:
+                raise CalibrationError(f"{file_path} already exists. Set overwrite to True.")
 
-        if os.path.isfile(parameter_value_file) and not overwrite:
-            raise CalibrationError(f"{parameter_value_file} already exists. Set overwrite to True.")
+            canonical_data = calibrations_to_dict(self, most_recent_only=most_recent_only)
+            with open(file_path, "w", encoding="utf-8") as file:
+                json.dump(canonical_data, file, cls=ExperimentEncoder)
 
-        if os.path.isfile(schedule_file) and not overwrite:
-            raise CalibrationError(f"{schedule_file} already exists. Set overwrite to True.")
+        elif file_type == "csv":
+            warnings.warn("Schedules are only saved in text format. They cannot be re-loaded.")
 
-        # Write the parameter configuration.
-        header_keys = ["parameter.name", "parameter unique id", "schedule", "qubits"]
-        body = []
+            parameter_config_file = file_prefix + "parameter_config.csv"
+            parameter_value_file = file_prefix + "parameter_values.csv"
+            schedule_file = file_prefix + "schedules.csv"
 
-        for parameter, keys in self.parameters.items():
-            for key in keys:
-                body.append(
-                    {
-                        "parameter.name": parameter.name,
-                        "parameter unique id": self._hash_to_counter_map[parameter],
-                        "schedule": key.schedule,
-                        "qubits": key.qubits,
-                    }
+            if os.path.isfile(parameter_config_file) and not overwrite:
+                raise CalibrationError(
+                    f"{parameter_config_file} already exists. Set overwrite to True."
                 )
 
-        if file_type == "csv":
+            if os.path.isfile(parameter_value_file) and not overwrite:
+                raise CalibrationError(
+                    f"{parameter_value_file} already exists. Set overwrite to True."
+                )
+
+            if os.path.isfile(schedule_file) and not overwrite:
+                raise CalibrationError(f"{schedule_file} already exists. Set overwrite to True.")
+
+            # Write the parameter configuration.
+            header_keys = ["parameter.name", "parameter unique id", "schedule", "qubits"]
+            body = []
+
+            for parameter, keys in self.parameters.items():
+                for key in keys:
+                    body.append(
+                        {
+                            "parameter.name": parameter.name,
+                            "parameter unique id": self._hash_to_counter_map[parameter],
+                            "schedule": key.schedule,
+                            "qubits": key.qubits,
+                        }
+                    )
+
             with open(parameter_config_file, "w", newline="", encoding="utf-8") as output_file:
                 dict_writer = csv.DictWriter(output_file, header_keys)
                 dict_writer.writeheader()
@@ -1453,6 +1468,14 @@ def save(
 
         os.chdir(cwd)
 
+    @deprecate_func(
+        since="0.6",
+        additional_msg=(
+            "Saving calibration in csv format is deprecate "
+            "as well as functions that support this functionality."
+        ),
+        package_name="qiskit-experiments",
+    )
     def schedule_information(self) -> Tuple[List[str], List[Dict]]:
         """Get the information on the schedules stored in the calibrations.
 
@@ -1471,6 +1494,11 @@ def schedule_information(self) -> Tuple[List[str], List[Dict]]:
 
         return ["name", "qubits", "schedule"], schedules
 
+    @deprecate_func(
+        since="0.6",
+        additional_msg="Loading and saving calibrations in CSV format is deprecated.",
+        package_name="qiskit-experiments",
+    )
     def load_parameter_values(self, file_name: str = "parameter_values.csv"):
         """
         Load parameter values from a given file into self._params.
@@ -1514,6 +1542,8 @@ def _add_parameter_value_from_conf(
             parameter: The name of the parameter.
             qubits: The qubits on which the parameter acts.
         """
+        # TODO remove this after load_parameter_values method is removed.
+
         param_val = ParameterValue(value, date_time, valid, exp_id, group)
 
         if schedule == "":
@@ -1525,12 +1555,32 @@ def _add_parameter_value_from_conf(
         self.add_parameter_value(param_val, *key, update_inst_map=False)
 
     @classmethod
-    def load(cls, files: List[str]) -> "Calibrations":
+    @deprecate_arg(
+        name="files",
+        new_alias="file_path",
+        since="0.6",
+        package_name="qiskit-experiments",
+    )
+    def load(cls, file_path: str) -> "Calibrations":
         """
         Retrieves the parameterized schedules and pulse parameters from the
         given location.
+
+        Args:
+            file_path: Path to file location.
+
+        Returns:
+            Calibration instance restored from the file.
         """
-        raise CalibrationError("Full calibration loading is not implemented yet.")
+        from .save_utils import calibrations_from_dict
+
+        with open(file_path, "r", encoding="utf-8") as file:
+            # Do we really need branching for data types?
+            # Parsing data format and dispatching the loader seems an overkill,
+            # but save method intend to support multiple formats.
+            cal_data = json.load(file, cls=ExperimentDecoder)
+
+        return calibrations_from_dict(cal_data)
 
     @staticmethod
     def _to_tuple(qubits: Union[str, int, Tuple[int, ...]]) -> Tuple[int, ...]:
@@ -1595,14 +1645,22 @@ def __eq__(self, other: "Calibrations") -> bool:
         if self._schedules.keys() != other._schedules.keys():
             return False
 
-        def _hash(data: dict):
-            return hash(json.dumps(data))
-
-        sorted_params_a = sorted(self.parameters_table()["data"], key=_hash)
-        sorted_params_b = sorted(other.parameters_table()["data"], key=_hash)
-
-        return sorted_params_a == sorted_params_b
-
+        def _counting(table):
+            return Counter(map(lambda d: tuple(d.items()), table["data"]))
+
+        # Use counting sort algorithm to compare unordered sequences
+        # https://en.wikipedia.org/wiki/Counting_sort
+        return _counting(self.parameters_table()) == _counting(other.parameters_table())
+
+    @deprecate_func(
+        since="0.6",
+        additional_msg=(
+            "Configuration data for Calibrations instance is deprecate. "
+            "Please use ExperimentEncoder and ExperimentDecoder to "
+            "serialize and deserialize this instance with JSON format."
+        ),
+        package_name="qiskit-experiments",
+    )
     def config(self) -> Dict[str, Any]:
         """Return the settings used to initialize the calibrations.
 
@@ -1635,23 +1693,33 @@ def config(self) -> Dict[str, Any]:
         }
 
     @classmethod
+    @deprecate_func(
+        since="0.6",
+        additional_msg="This method will be removed and no alternative will be provided.",
+        package_name="qiskit-experiments",
+    )
     def from_config(cls, config: Dict) -> "Calibrations":
-        """Deserialize the calibrations given the input dictionary"""
-
-        config["kwargs"]["control_channel_map"] = config["kwargs"]["control_channel_map"].chan_map
+        """Restore Calibration from config data.
 
-        calibrations = cls(**config["kwargs"])
+        Args:
+            config: Configuration data.
 
-        for param_config in config["parameters"]:
-            calibrations._add_parameter_value_from_conf(**param_config)
+        Returns:
+            Calibration instance restored from configuration data.
+        """
+        from .save_utils import calibrations_from_dict
 
-        return calibrations
+        return calibrations_from_dict(config)
 
     def __json_encode__(self):
         """Convert to format that can be JSON serialized."""
-        return self.config()
+        from .save_utils import calibrations_to_dict
+
+        return calibrations_to_dict(self, most_recent_only=False)
 
     @classmethod
     def __json_decode__(cls, value: Dict[str, Any]) -> "Calibrations":
         """Load from JSON compatible format."""
-        return cls.from_config(value)
+        from .save_utils import calibrations_from_dict
+
+        return calibrations_from_dict(value)