feat: Use QPY as exchange format (#104)

* Use QPY as exchange format

* Fix tests

* Fix tests for QuantumVolume

* Add test to check wrong input circuits

* Convert last tests that were using QASM

* Add ecr test for qpy

* Add fix to avoid random test errors

* Revert "Add fix to avoid random test errors"

This reverts commit 505fea3.

* Add release notes

* Accept only QuantumCircuit as input

* Fix tests

* Use QASM as fallback serialization format if QPY fails

* Fix test_layout_construction_no_service

* Fix code for Python 3.8

* Add testcase for qpy fallback

* Code review suggestions

qiskit_ibm_transpiler/transpiler_service.py

@@ -96,7 +96,7 @@ def __init__(
 
     def run(
         self,
-        circuits: Union[List[Union[str, QuantumCircuit]], Union[str, QuantumCircuit]],
+        circuits: Union[List[QuantumCircuit], QuantumCircuit],
     ):
         """Transpile the circuit(s) by calling the service /transpile endpoint.
 

qiskit_ibm_transpiler/utils.py

@@ -26,14 +26,16 @@
 
 """
 
-from typing import Dict, Union
+from typing import Dict, Union, List, Tuple
 
 import numpy as np
-from qiskit import QuantumCircuit, qasm2, qasm3
+import base64
+import io
+import struct
+from qiskit import QuantumCircuit, qasm2, qasm3, qpy
 from qiskit.circuit.library import LinearFunction
 from qiskit.quantum_info import Clifford
 from qiskit.synthesis.linear.linear_matrix_utils import random_invertible_binary_matrix
-from qiskit import QuantumCircuit, qasm2, qasm3
 from qiskit.circuit import QuantumCircuit, library
 from qiskit.transpiler.basepasses import TransformationPass
 
@@ -127,3 +129,64 @@ def get_circuit_from_qasm(qasm_string: str) -> QuantumCircuit:
     qasm2.CustomInstruction("ecr", 0, 2, library.ECRGate)
 )
 get_circuit_from_qasm.fix_ecr = FixECR()
+
+
+def get_qpy_from_circuit(
+    input_circ: Union[QuantumCircuit, List[QuantumCircuit]]
+) -> str:
+    if isinstance(input_circ, QuantumCircuit) or isinstance(input_circ, list):
+        output_b = io.BytesIO()
+        qpy.dump(input_circ, output_b)
+        qpy_string = base64.b64encode(output_b.getvalue()).decode("utf-8")
+    else:
+        raise TypeError(
+            f"Input circuits must be QuantumCircuit or list of QuantumCircuits. Type provided: {type(input_circ)}"
+        )
+    return qpy_string
+
+
+def get_circuit_from_qpy(qpy_string: str) -> QuantumCircuit:
+    return qpy.load(io.BytesIO(base64.b64decode(qpy_string.encode("utf-8"))))[0]
+
+
+def get_circuits_from_qpy(qpy_string: str) -> List[QuantumCircuit]:
+    return qpy.load(io.BytesIO(base64.b64decode(qpy_string.encode("utf-8"))))
+
+
+def serialize_circuit_to_qpy_or_qasm(
+    input_circuit: QuantumCircuit,
+) -> Tuple[Union[str, None], Union[str, None]]:
+    qpy_result = None
+    qasm_result = None
+    try:
+        qpy_result = get_qpy_from_circuit(input_circuit)
+    except struct.error:
+        qasm_result = input_to_qasm(input_circuit).replace("\n", " ")
+
+    return qpy_result, qasm_result
+
+
+def serialize_circuits_to_qpy_or_qasm(
+    input_circuits: List[QuantumCircuit],
+) -> Tuple[Union[str, None], Union[List[str], None]]:
+    qpy_result = None
+    qasm_result = None
+    try:
+        qpy_result = get_qpy_from_circuit(input_circuits)
+    except struct.error:
+        qasm_result = [
+            input_to_qasm(the_circ).replace("\n", " ") for the_circ in input_circuits
+        ]
+
+    return qpy_result, qasm_result
+
+
+def deserialize_circuit_from_qpy_or_qasm(
+    qpy_input: Union[str, None], qasm_input: Union[str, None]
+) -> Union[QuantumCircuit, None]:
+    if qpy_input:
+        return get_circuit_from_qpy(qpy_input)
+    elif qasm_input:
+        return get_circuit_from_qasm(qasm_input)
+    else:
+        return None

qiskit_ibm_transpiler/wrappers/ai_routing.py

@@ -11,7 +11,12 @@
 # that they have been altered from the originals.
 
 from qiskit import QuantumCircuit
-from qiskit_ibm_transpiler.utils import get_circuit_from_qasm, input_to_qasm
+from qiskit_ibm_transpiler.utils import (
+    deserialize_circuit_from_qpy_or_qasm,
+    get_circuit_from_qpy,
+    get_qpy_from_circuit,
+    serialize_circuit_to_qpy_or_qasm,
+)
 from .base import QiskitTranspilerService
 from typing import List, Union, Literal
 
@@ -37,10 +42,10 @@ def routing(
             OptimizationOptions, List[OptimizationOptions], None
         ] = None,
     ):
-        qasm = input_to_qasm(circuit)
-
+        qpy, qasm = serialize_circuit_to_qpy_or_qasm(circuit)
         body_params = {
-            "qasm": qasm.replace("\n", " "),
+            "qasm": qasm,
+            "qpy": qpy,
             "coupling_map": coupling_map,
             "optimization_preferences": optimization_preferences,
         }
@@ -56,7 +61,9 @@ def routing(
         )
 
         if routing_resp.get("success"):
-            routed_circuit = get_circuit_from_qasm(routing_resp["qasm"])
+            routed_circuit = deserialize_circuit_from_qpy_or_qasm(
+                routing_resp["qpy"], routing_resp["qasm"]
+            )
             return (
                 routed_circuit,
                 routing_resp["layout"]["initial"],

qiskit_ibm_transpiler/wrappers/ai_synthesis.py

@@ -18,6 +18,7 @@
 from qiskit.quantum_info import Clifford
 
 from .base import QiskitTranspilerService
+from ..utils import deserialize_circuit_from_qpy_or_qasm
 
 logging.basicConfig()
 logging.getLogger(__name__).setLevel(logging.INFO)
@@ -66,8 +67,12 @@ def transpile(
 
         results = []
         for transpile_resp in transpile_resps:
-            if transpile_resp.get("success") and transpile_resp.get("qasm") is not None:
-                results.append(QuantumCircuit.from_qasm_str(transpile_resp.get("qasm")))
+            if transpile_resp.get("success"):
+                results.append(
+                    deserialize_circuit_from_qpy_or_qasm(
+                        transpile_resp.get("qpy"), transpile_resp.get("qasm")
+                    )
+                )
             else:
                 results.append(None)
         return results
@@ -116,8 +121,12 @@ def transpile(
 
         results = []
         for transpile_resp in transpile_resps:
-            if transpile_resp.get("success") and transpile_resp.get("qasm") is not None:
-                results.append(QuantumCircuit.from_qasm_str(transpile_resp.get("qasm")))
+            if transpile_resp.get("success"):
+                results.append(
+                    deserialize_circuit_from_qpy_or_qasm(
+                        transpile_resp.get("qpy"), transpile_resp.get("qasm")
+                    )
+                )
             else:
                 results.append(None)
         return results
@@ -163,8 +172,12 @@ def transpile(
 
         results = []
         for transpile_resp in transpile_resps:
-            if transpile_resp.get("success") and transpile_resp.get("qasm") is not None:
-                results.append(QuantumCircuit.from_qasm_str(transpile_resp.get("qasm")))
+            if transpile_resp.get("success"):
+                results.append(
+                    deserialize_circuit_from_qpy_or_qasm(
+                        transpile_resp.get("qpy"), transpile_resp.get("qasm")
+                    )
+                )
             else:
                 results.append(None)
         return results

qiskit_ibm_transpiler/wrappers/transpile.py

@@ -18,7 +18,13 @@
 from qiskit.circuit import QuantumCircuit, QuantumRegister, Qubit
 from qiskit.transpiler import TranspileLayout
 from qiskit.transpiler.layout import Layout
-from qiskit_ibm_transpiler.utils import get_circuit_from_qasm, input_to_qasm
+from qiskit_ibm_transpiler.utils import (
+    deserialize_circuit_from_qpy_or_qasm,
+    get_circuit_from_qpy,
+    get_circuits_from_qpy,
+    get_qpy_from_circuit,
+    serialize_circuits_to_qpy_or_qasm,
+)
 from qiskit_ibm_transpiler.wrappers import QiskitTranspilerService
 
 # setting backoff logger to error level to avoid too much logging
@@ -37,9 +43,7 @@ def __init__(self, **kwargs):
 
     def transpile(
         self,
-        circuits: Union[
-            Union[List[str], str], Union[List[QuantumCircuit], QuantumCircuit]
-        ],
+        circuits: Union[List[QuantumCircuit], QuantumCircuit],
         optimization_level: int = 1,
         optimization_preferences: Union[
             OptimizationOptions, List[OptimizationOptions], None
@@ -51,12 +55,12 @@ def transpile(
         ai_layout_mode: str = None,
         use_fractional_gates: bool = False,
     ):
-        circuits = circuits if isinstance(circuits, list) else [circuits]
-
-        qasm_circuits = [input_to_qasm(circ) for circ in circuits]
+        circuits = [circuits] if isinstance(circuits, QuantumCircuit) else circuits
+        qpy_circuits, qasm_circuits = serialize_circuits_to_qpy_or_qasm(circuits)
 
         body_params = {
             "qasm_circuits": qasm_circuits,
+            "qpy_circuits": qpy_circuits,
             "optimization_preferences": optimization_preferences,
         }
 
@@ -100,17 +104,13 @@ def transpile(
 
 
 def _get_circuit_from_result(transpile_resp, orig_circuit):
-    transpiled_circuit = get_circuit_from_qasm(transpile_resp["qasm"])
+    transpiled_circuit = deserialize_circuit_from_qpy_or_qasm(
+        transpile_resp["qpy"], transpile_resp["qasm"]
+    )
 
     init_layout = transpile_resp["layout"]["initial"]
     final_layout = transpile_resp["layout"]["final"]
 
-    orig_circuit = (
-        get_circuit_from_qasm(orig_circuit)
-        if isinstance(orig_circuit, str)
-        else orig_circuit
-    )
-
     transpiled_circuit = QuantumCircuit(len(init_layout)).compose(transpiled_circuit)
     transpiled_circuit._layout = _create_transpile_layout(
         init_layout, final_layout, transpiled_circuit, orig_circuit

release-notes/unreleased/104.feat.rst

@@ -0,0 +1 @@
+Use QPY as exchange format with service