Refactors ai_api_synthesis (#140)

* Refactors AICliffordAPI * Refactors AIPermutationAPI * Extracts handle response functionality * Fixes qiskit_ibm_ai_local_transpiler version * Fixes linter errors * Organizes imports --------- Co-authored-by: Jesus Talavera <[email protected]>
Qiskit · Jan 9, 2025 · 2130b0e · 2130b0e
1 parent 0a831a1
commit 2130b0e
Show file tree

Hide file tree

Showing 2 changed files with 127 additions and 119 deletions.
diff --git a/qiskit_ibm_transpiler/wrappers/ai_api_synthesis.py b/qiskit_ibm_transpiler/wrappers/ai_api_synthesis.py
@@ -19,7 +19,6 @@
 from qiskit.quantum_info import Clifford
 
 from ..utils import (
-    deserialize_circuit_from_qpy_or_qasm,
     serialize_circuits_to_qpy_or_qasm,
 )
 from .base import QiskitTranspilerService
@@ -42,48 +41,51 @@ def transpile(
         backend_name: Union[str, None] = None,
         # backend is not used yet, but probably it will replace backend_name
         backend: Union[Backend, None] = None,
-    ):
-        if coupling_map is not None:
-            logger.info("Running synthesis against the Qiskit Transpiler Service")
-            transpile_resps = self.request_and_wait(
-                endpoint="transpile",
-                body={
-                    "clifford_dict": [
-                        Clifford(circuit).to_dict() for circuit in circuits
-                    ],
-                    "qargs": qargs,
-                    "backend_coupling_map": coupling_map,
-                },
-                params=dict(),
-            )
-        elif backend_name is not None:
-            logger.info("Running synthesis against the Qiskit Transpiler Service")
-            transpile_resps = self.request_and_wait(
-                endpoint="transpile",
-                body={
-                    "clifford_dict": [
-                        Clifford(circuit).to_dict() for circuit in circuits
-                    ],
-                    "qargs": qargs,
-                },
-                params={"backend": backend_name},
-            )
-        else:
+    ) -> List[Union[QuantumCircuit, None]]:
+        """Synthetize one or more quantum circuits into an optimized equivalent. It differs from a standard synthesis process in that it takes into account where the cliffords are (qargs)
+        and respects it on the synthesized circuit.
+
+        Args:
+            circuits (List[Union[QuantumCircuit, Clifford]]): A list of quantum circuits to be synthesized.
+            qargs (List[List[int]]): A list of lists of qubit indices for each circuit. Each list of qubits indices represent where the cliffors circuit is.
+            coupling_map (Union[List[List[int]], None]): A coupling map representing the connectivity of the quantum computer.
+            backend_name (Union[str, None]): The name of the backend to use for the synthesis.
+
+        Returns:
+            List[Union[QuantumCircuit, None]]: A list of synthesized quantum circuits. If the synthesis fails for any circuit, the corresponding element in the list will be None.
+        """
+
+        # Although this function is called `transpile`, it does a synthesis. It has this name because the synthesis
+        # is made as a pass on the Qiskit Pass Manager which is used in the transpilation process.
+
+        if not coupling_map and not backend_name:
             raise ValueError(
                 "ERROR. Either a 'coupling_map' or a 'backend_name' must be provided."
             )
 
-        results = []
-        for transpile_resp in transpile_resps:
-            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
+        body_params = {
+            "clifford_dict": [Clifford(circuit).to_dict() for circuit in circuits],
+            "qargs": qargs,
+        }
+
+        query_params = dict()
+
+        if coupling_map is not None:
+            body_params["backend_coupling_map"] = coupling_map
+        elif backend_name is not None:
+            query_params["backend"] = backend_name
+
+        logger.info("Running synthesis against the Qiskit Transpiler Service")
+
+        transpile_response = self.request_and_wait(
+            endpoint="transpile",
+            body=body_params,
+            params=query_params,
+        )
+
+        synthesized_circuits = self._handle_response(transpile_response)
+
+        return synthesized_circuits
 
 
 class AILinearFunctionAPI(QiskitTranspilerService):
@@ -142,14 +144,7 @@ def transpile(
             params=query_params,
         )
 
-        synthesized_circuits = []
-        for response_element in transpile_response:
-            synthesized_circuit = None
-            if response_element.get("success"):
-                synthesized_circuit = deserialize_circuit_from_qpy_or_qasm(
-                    response_element.get("qpy"), response_element.get("qasm")
-                )
-            synthesized_circuits.append(synthesized_circuit)
+        synthesized_circuits = self._handle_response(transpile_response)
 
         return synthesized_circuits
 
@@ -168,45 +163,51 @@ def transpile(
         backend_name: Union[str, None] = None,
         # backend is not used yet, but probably it will replace backend_name
         backend: Union[Backend, None] = None,
-    ):
+    ) -> List[Union[QuantumCircuit, None]]:
+        """Synthetize one or more permutation arrays into an optimized circuit equivalent. It differs from a standard synthesis process in that it takes into account where the permutations are (qargs)
+        and respects it on the synthesized circuit.
 
-        if coupling_map is not None:
-            logger.info("Running synthesis against the Qiskit Transpiler Service")
-            transpile_resps = self.request_and_wait(
-                endpoint="transpile",
-                body={
-                    "permutation": patterns,
-                    "qargs": qargs,
-                    "backend_coupling_map": coupling_map,
-                },
-                params=dict(),
-            )
-        elif backend_name is not None:
-            logger.info("Running synthesis against the Qiskit Transpiler Service")
-            transpile_resps = self.request_and_wait(
-                endpoint="transpile",
-                body={
-                    "permutation": patterns,
-                    "qargs": qargs,
-                },
-                params={"backend": backend_name},
-            )
-        else:
+        Args:
+            patterns: List[List[int]]: A list of permutation arrays to be synthesized.
+            qargs (List[List[int]]): A list of lists of qubit indices for each permutation array. Each list of qubits indices represent where the permutation array is.
+            coupling_map (Union[List[List[int]], None]): A coupling map representing the connectivity of the quantum computer.
+            backend_name (Union[str, None]): The name of the backend to use for the synthesis.
+
+        Returns:
+            List[Union[QuantumCircuit, None]]: A list of synthesized quantum circuits. If the synthesis fails for any circuit, the corresponding element in the list will be None.
+        """
+
+        # Although this function is called `transpile`, it does a synthesis. It has this name because the synthesis
+        # is made as a pass on the Qiskit Pass Manager which is used in the transpilation process.
+
+        if not coupling_map and not backend_name:
             raise ValueError(
                 "ERROR. Either a 'coupling_map' or a 'backend_name' must be provided."
             )
 
-        results = []
-        for transpile_resp in transpile_resps:
-            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
+        body_params = {
+            "permutation": patterns,
+            "qargs": qargs,
+        }
+
+        query_params = dict()
+
+        if coupling_map is not None:
+            body_params["backend_coupling_map"] = coupling_map
+        elif backend_name is not None:
+            query_params["backend"] = backend_name
+
+        logger.info("Running synthesis against the Qiskit Transpiler Service")
+
+        transpile_response = self.request_and_wait(
+            endpoint="transpile",
+            body=body_params,
+            params=query_params,
+        )
+
+        synthesized_circuits = self._handle_response(transpile_response)
+
+        return synthesized_circuits
 
 
 class AIPauliNetworkAPI(QiskitTranspilerService):
@@ -224,45 +225,35 @@ def transpile(
         # backend is not used yet, but probably it will replace backend_name
         backend: Union[Backend, None] = None,
     ):
+        if not coupling_map and not backend_name:
+            raise ValueError(
+                "ERROR. Either a 'coupling_map' or a 'backend_name' must be provided."
+            )
+
         qpy, qasm = serialize_circuits_to_qpy_or_qasm(
             circuits, self.get_qiskit_version()
         )
+        body_params = {
+            "qasm": qasm,
+            "qpy": qpy,
+            "qargs": qargs,
+        }
+
+        query_params = dict()
+
         if coupling_map is not None:
-            logger.info("Running synthesis against the Qiskit Transpiler Service")
-            transpile_resps = self.request_and_wait(
-                endpoint="transpile",
-                body={
-                    "qasm": qasm,
-                    "qpy": qpy,
-                    "qargs": qargs,
-                    "backend_coupling_map": coupling_map,
-                },
-                params={"backend": ""},
-            )
+            body_params["backend_coupling_map"] = coupling_map
         elif backend_name is not None:
-            logger.info("Running synthesis against the Qiskit Transpiler Service")
-            transpile_resps = self.request_and_wait(
-                endpoint="transpile",
-                body={
-                    "qasm": qasm,
-                    "qpy": qpy,
-                    "qargs": qargs,
-                },
-                params={"backend": backend_name},
-            )
-        else:
-            raise ValueError(
-                f"ERROR. Either a 'coupling_map' or a 'backend_name' must be provided."
-            )
+            query_params["backend"] = backend_name
 
-        results = []
-        for transpile_resp in transpile_resps:
-            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
+        logger.info("Running synthesis against the Qiskit Transpiler Service")
+
+        transpile_response = self.request_and_wait(
+            endpoint="transpile",
+            body=body_params,
+            params=query_params,
+        )
+
+        synthesized_circuits = self._handle_response(transpile_response)
+
+        return synthesized_circuits
diff --git a/qiskit_ibm_transpiler/wrappers/base.py b/qiskit_ibm_transpiler/wrappers/base.py
@@ -14,15 +14,17 @@
 import json
 import logging
 import os
-from http import HTTPStatus
 from pathlib import Path
-from typing import Dict
+from typing import Dict, List, Union
 from urllib.parse import urljoin
 
 import backoff
 import requests
+from qiskit import QuantumCircuit
 from qiskit.transpiler.exceptions import TranspilerError
 
+from ..utils import deserialize_circuit_from_qpy_or_qasm
+
 logger = logging.getLogger(__name__)
 
 
@@ -195,6 +197,21 @@ def _request_transp(endpoint: str, body: Dict, params: Dict):
         else:
             return result
 
+    def _handle_response(
+        self, transpile_response: List[dict]
+    ) -> List[Union[QuantumCircuit, None]]:
+        """Handle the transpile response from the server."""
+        synthesized_circuits = []
+        for response_element in transpile_response:
+            if response_element.get("success"):
+                circuit = deserialize_circuit_from_qpy_or_qasm(
+                    response_element.get("qpy"), response_element.get("qasm")
+                )
+                synthesized_circuits.append(circuit)
+            else:
+                synthesized_circuits.append(None)
+        return synthesized_circuits
+
 
 def _raise_transpiler_error_and_log(msg: str):
     logger.error(msg)