feat(qio): update to qio

Author: Sonderfall

examples/run_aqt_estimator.py

@@ -63,9 +63,7 @@ def cost_function(
     Return the estimated expectation value of the Hamiltonian
     on the state prepared by the Ansatz circuit.
     """
-    return float(
-        estimator.run([(ansatz, hamiltonian, params)]).result().values[0]
-    )
+    return float(estimator.run([(ansatz, hamiltonian, params)]).result().values[0])
 
 
 # Run the VQE using the SciPy minimizer routine

qiskit_scaleway/backends/aqt/backend.py

@@ -11,7 +11,6 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-
 from qiskit.providers import Options
 
 from qiskit_scaleway.backends import BaseBackend

qiskit_scaleway/backends/base_job.py

@@ -1,4 +1,4 @@
-# Copyright 2024 Scaleway
+# Copyright 2025 Scaleway
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -12,32 +12,31 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import time
-import zlib
 import httpx
-import json
-import numpy as np
 import randomname
 
 from typing import List, Union, Optional, Dict
 
-from qiskit import qasm3, QuantumCircuit
+from qiskit import QuantumCircuit
 from qiskit.result import Result
 from qiskit.providers import JobV1
 from qiskit.providers import JobError, JobTimeoutError, JobStatus
 
 from qiskit_scaleway.versions import USER_AGENT
 
+from qio.core import (
+    QuantumProgram,
+    QuantumProgramResult,
+    QuantumComputationModel,
+    QuantumComputationParameters,
+    QuantumNoiseModel,
+    BackendData,
+    ClientData,
+)
+
 from scaleway_qaas_client.v1alpha1 import (
     QaaSClient,
     QaaSJobResult,
-    QaaSJobData,
-    QaaSJobClientData,
-    QaaSCircuitData,
-    QaaSJobRunData,
-    QaaSJobBackendData,
-    QaaSCircuitSerializationFormat,
-    QaaSNoiseModelData,
-    QaaSNoiseModelSerializationFormat,
 )
 
 
@@ -78,59 +77,39 @@ def submit(self, session_id: str) -> None:
         options = self._config.copy()
         shots = options.pop("shots")
 
-        run_data = QaaSJobRunData(
-            options={
-                "shots": shots,
-                "memory": options.pop("memory", False),
-            },
-            circuits=list(
-                map(
-                    lambda c: QaaSCircuitData(
-                        serialization_format=QaaSCircuitSerializationFormat.QASM_V3,
-                        circuit_serialization=qasm3.dumps(c),
-                    ),
-                    self._circuits,
-                )
-            ),
-        )
+        programs = map(lambda c: QuantumProgram.from_qiskit_circuit(c), self._circuits)
 
         noise_model = options.pop("noise_model", None)
         if noise_model:
-            noise_model_dict = _encode_numpy_complex(noise_model.to_dict(False))
-            noise_model = QaaSNoiseModelData(
-                serialization_format=QaaSNoiseModelSerializationFormat.AER_COMPRESSED_JSON,
-                noise_model_serialization=zlib.compress(
-                    json.dumps(noise_model_dict).encode()
-                ),
-            )
-            ### Uncomment to use standard JSON serialization, provided there is no more issue with AER deserialization logic
-            # noise_model = QaaSNoiseModelData(
-            #     serialization_format = QaaSNoiseModelSerializationFormat.JSON,
-            #     noise_model_serialization = json.dumps(noise_model.to_dict(True)).encode()
-            # )
-            ###
-
-        backend_data = QaaSJobBackendData(
+            noise_model = QuantumNoiseModel.from_qiskit_aer_noise_model(noise_model)
+
+        backend_data = BackendData(
             name=self.backend().name,
             version=self.backend().version,
             options=options,
+            frozenset=True,
         )
 
-        client_data = QaaSJobClientData(
+        client_data = ClientData(
             user_agent=USER_AGENT,
+            frozenset=True,
         )
 
-        data = QaaSJobData.schema().dumps(
-            QaaSJobData(
-                backend=backend_data,
-                run=run_data,
-                client=client_data,
-                noise_model=noise_model,
-            )
-        )
+        computation_model_dict = QuantumComputationModel(
+            programs=programs,
+            backend=backend_data,
+            client=client_data,
+            noise_model=noise_model,
+            frozenset=True,
+        ).to_dict()
+
+        computation_parameters_dict = QuantumComputationParameters(
+            shots=shots,
+            frozenset=True,
+        ).to_dict()
 
         model = self._client.create_model(
-            payload=data,
+            payload=computation_model_dict,
         )
 
         if not model:
@@ -140,7 +119,7 @@ def submit(self, session_id: str) -> None:
             name=self._name,
             session_id=session_id,
             model_id=model.id,
-            parameters={"shots": shots},
+            parameters=computation_parameters_dict,
         ).id
 
     def result(
@@ -151,37 +130,28 @@ def result(
 
         job_results = self._wait_for_result(timeout, fetch_interval)
 
-        def __make_result_from_payload(payload: str) -> Result:
-            payload_dict = json.loads(payload)
-
-            return Result.from_dict(
-                {
-                    "results": payload_dict["results"],
-                    "backend_name": self.backend().name,
-                    "backend_version": self.backend().version,
-                    "job_id": self._job_id,
-                    "qobj_id": ", ".join(x.name for x in self._circuits),
-                    "success": payload_dict["success"],
-                    "header": payload_dict.get("header"),
-                    "metadata": payload_dict.get("metadata"),
-                }
-            )
-
-        qiskit_results = list(
+        program_results = list(
             map(
-                lambda r: __make_result_from_payload(
-                    self._extract_payload_from_response(r)
+                lambda r: self._extract_payload_from_response(r).to_qiskit_result(
+                    **{
+                        "backend_name": self.backend().name,
+                        "backend_version": self.backend().version,
+                        "job_id": self._job_id,
+                        "qobj_id": ", ".join(x.name for x in self._circuits),
+                    }
                 ),
                 job_results,
             )
         )
 
-        if len(qiskit_results) == 1:
-            return qiskit_results[0]
+        if len(program_results) == 1:
+            return program_results[0]
 
-        return qiskit_results
+        return program_results
 
-    def _extract_payload_from_response(self, job_result: QaaSJobResult) -> str:
+    def _extract_payload_from_response(
+        self, job_result: QaaSJobResult
+    ) -> QuantumProgramResult:
         result = job_result.result
 
         if result is None or result == "":
@@ -190,12 +160,11 @@ def _extract_payload_from_response(self, job_result: QaaSJobResult) -> str:
             if url is not None:
                 resp = httpx.get(url)
                 resp.raise_for_status()
-
-                return resp.text
+                result = resp.text
             else:
-                raise Exception("Got result with empty data and url fields")
-        else:
-            return result
+                raise RuntimeError("Got result with empty data and url fields")
+
+        return QuantumProgramResult.from_json(result)
 
     def _wait_for_result(
         self, timeout: Optional[int], fetch_interval: int
@@ -217,25 +186,3 @@ def _wait_for_result(
                 raise JobError("Job error")
 
             time.sleep(fetch_interval)
-
-
-def _encode_numpy_complex(obj):
-    """
-    Recursively traverses a structure and converts numpy arrays and
-    complex numbers into a JSON-serializable format.
-    """
-    if isinstance(obj, np.ndarray):
-        return {
-            "__ndarray__": True,
-            "data": _encode_numpy_complex(obj.tolist()),  # Recursively encode data
-            "dtype": obj.dtype.name,
-            "shape": obj.shape,
-        }
-    elif isinstance(obj, (complex, np.complex128)):
-        return {"__complex__": True, "real": obj.real, "imag": obj.imag}
-    elif isinstance(obj, dict):
-        return {key: _encode_numpy_complex(value) for key, value in obj.items()}
-    elif isinstance(obj, (list, tuple)):
-        return [_encode_numpy_complex(item) for item in obj]
-    else:
-        return obj

requirements.txt

@@ -3,4 +3,5 @@ qiskit-aer~=0.17
 randomname>=0.2.1
 dataclasses-json>=0.6.4
 dataclasses>=0.6
-scaleway-qaas-client>=0.1.23
+scaleway-qaas-client>=0.1.23
+qio>=0.1.0