fix: Align QPY version with service (#136)

* Force qpy to use version 12

* Refactor

* Serialize QPY circuits according to service qiskit version

* Fix qpy version wrong typing

* Fix json access when key doesn't exist

* Fix corner cases when requesting qiskit version to service

* Refactor

* Update qiskit_ibm_transpiler/utils.py

Co-authored-by: victor-villar <[email protected]>

* Fix typo

* Improve logic in case qiskit version is not mapped

* Add release notes

---------

Co-authored-by: victor-villar <[email protected]>

qiskit_ibm_transpiler/utils.py

@@ -37,13 +37,19 @@
 
 from qiskit import QuantumCircuit, qasm2, qasm3, qpy
 from qiskit.circuit.library import LinearFunction
+from qiskit.qpy import common
 from qiskit.quantum_info import Clifford
 from qiskit.synthesis.linear.linear_matrix_utils import random_invertible_binary_matrix
 from qiskit.circuit import QuantumCircuit, library
 from qiskit.transpiler.basepasses import TransformationPass
 
 logger = logging.getLogger(__name__)
 
+QPY_QISKIT_VERSION_MAPPING = {
+    "1.3.0": 13,
+    "1.2.4": 12,
+}
+
 
 def get_metrics(qc: QuantumCircuit) -> Dict[str, int]:
     """Returns a dict with metrics from a QuantumCircuit"""
@@ -232,11 +238,16 @@ def check_topology_synthesized_circuit(
 
 
 def get_qpy_from_circuit(
-    input_circ: Union[QuantumCircuit, List[QuantumCircuit]]
+    input_circ: Union[QuantumCircuit, List[QuantumCircuit]],
+    qiskit_version: Union[str, None] = None,
 ) -> str:
     if isinstance(input_circ, QuantumCircuit) or isinstance(input_circ, list):
         output_b = io.BytesIO()
-        qpy.dump(input_circ, output_b)
+        qpy.dump(
+            input_circ,
+            output_b,
+            version=QPY_QISKIT_VERSION_MAPPING.get(qiskit_version, common.QPY_VERSION),
+        )
         qpy_string = base64.b64encode(output_b.getvalue()).decode("utf-8")
     else:
         raise TypeError(
@@ -255,24 +266,25 @@ def get_circuits_from_qpy(qpy_string: str) -> List[QuantumCircuit]:
 
 def serialize_circuit_to_qpy_or_qasm(
     input_circuit: QuantumCircuit,
+    qiskit_version: Union[str, None] = None,
 ) -> Tuple[Union[str, None], Union[str, None]]:
     qpy_result = None
     qasm_result = None
     try:
-        qpy_result = get_qpy_from_circuit(input_circuit)
+        qpy_result = get_qpy_from_circuit(input_circuit, qiskit_version)
     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],
+    input_circuits: List[QuantumCircuit], qiskit_version: Union[str, None] = None
 ) -> Tuple[Union[str, None], Union[List[str], None]]:
     qpy_result = None
     qasm_result = None
     try:
-        qpy_result = get_qpy_from_circuit(input_circuits)
+        qpy_result = get_qpy_from_circuit(input_circuits, qiskit_version)
     except struct.error:
         qasm_result = [
             input_to_qasm(the_circ).replace("\n", " ") for the_circ in input_circuits

qiskit_ibm_transpiler/wrappers/ai_api_routing.py

@@ -43,7 +43,7 @@ def routing(
             OptimizationOptions, List[OptimizationOptions], None
         ] = None,
     ):
-        qpy, qasm = serialize_circuit_to_qpy_or_qasm(circuit)
+        qpy, qasm = serialize_circuit_to_qpy_or_qasm(circuit, self.get_qiskit_version())
         body_params = {
             "qasm": qasm,
             "qpy": qpy,

qiskit_ibm_transpiler/wrappers/ai_api_synthesis.py

@@ -224,7 +224,9 @@ def transpile(
         # backend is not used yet, but probably it will replace backend_name
         backend: Union[Backend, None] = None,
     ):
-        qpy, qasm = serialize_circuits_to_qpy_or_qasm(circuits)
+        qpy, qasm = serialize_circuits_to_qpy_or_qasm(
+            circuits, self.get_qiskit_version()
+        )
         if coupling_map is not None:
             logger.info("Running synthesis against the Qiskit Transpiler Service")
             transpile_resps = self.request_and_wait(

qiskit_ibm_transpiler/wrappers/base.py

@@ -101,6 +101,13 @@ def get_versions(self):
 
         return resp
 
+    def get_qiskit_version(self):
+        try:
+            return self.get_versions().get("qiskit")
+        except Exception as exc:
+            logger.warning(f"Exception requesting qiskit version: {exc}")
+            return None
+
     def get_supported_backends(self):
         url = f"{self.url}/backends"
         resp = requests.get(

qiskit_ibm_transpiler/wrappers/transpile.py

@@ -56,7 +56,9 @@ def transpile(
         use_fractional_gates: bool = False,
     ):
         circuits = [circuits] if isinstance(circuits, QuantumCircuit) else circuits
-        qpy_circuits, qasm_circuits = serialize_circuits_to_qpy_or_qasm(circuits)
+        qpy_circuits, qasm_circuits = serialize_circuits_to_qpy_or_qasm(
+            circuits, self.get_qiskit_version()
+        )
 
         body_params = {
             "qasm_circuits": qasm_circuits,

release-notes/unreleased/136.bug.rst

@@ -0,0 +1 @@
+Align QPY version with service.