Remove Qiskit batch plumbling

azure-quantum/azure/quantum/qiskit/backends/backend.py

@@ -10,7 +10,7 @@
 
 logger = logging.getLogger(__name__)
 
-from typing import Any, Dict, Tuple, Union, List, Optional
+from typing import Any, Dict, Union, List, Optional
 from azure.quantum.version import __version__
 from azure.quantum.qiskit.job import (
     MICROSOFT_OUTPUT_DATA_FORMAT,
@@ -27,7 +27,6 @@
     from qiskit.providers import Provider
     from qiskit.providers.models import BackendConfiguration
     from qiskit.qobj import QasmQobj, PulseQobj
-    import pyqir as pyqir
     from qsharp.interop.qiskit import QSharpBackend
     from qsharp import TargetProfile
 
@@ -101,7 +100,7 @@ def run(
 
         Args:
             run_input (QuantumCircuit or List[QuantumCircuit]): An individual or a
-            list of :class:`~qiskit.circuits.QuantumCircuit` to run on the backend.
+            list of one :class:`~qiskit.circuits.QuantumCircuit` to run on the backend.
             shots (int, optional): Number of shots, defaults to None.
             options: Any kwarg options to pass to the backend for running the
             config. If a key is also present in the options
@@ -334,7 +333,7 @@ def run(
 
         Args:
             run_input (QuantumCircuit or List[QuantumCircuit]): An individual or a
-            list of :class:`~qiskit.circuits.QuantumCircuit` to run on the backend.
+            list of one :class:`~qiskit.circuits.QuantumCircuit` to run on the backend.
             shots (int, optional): Number of shots, defaults to None.
             options: Any kwarg options to pass to the backend for running the
             config. If a key is also present in the options
@@ -349,17 +348,16 @@ def run(
         options.pop("run_input", None)
         options.pop("circuit", None)
 
-        circuits = list([])
-        if isinstance(run_input, QuantumCircuit):
-            circuits = [run_input]
-        else:
-            circuits = run_input
-
-        max_circuits_per_job = self.configuration().max_experiments
-        if len(circuits) > max_circuits_per_job:
-            raise NotImplementedError(
-                f"This backend only supports running a maximum of {max_circuits_per_job} circuits per job."
-            )
+        circuit = run_input
+        if isinstance(run_input, list):
+            # just in case they passed a list, we only support single-experiment jobs
+            if len(run_input) != 1:
+                raise NotImplementedError(
+                    f"This backend only supports running a single circuit per job."
+                )
+            circuit = run_input[0]
+            if not isinstance(circuit, QuantumCircuit):
+                raise ValueError("Invalid input: expected a QuantumCircuit.")
 
         # config normalization
         input_params = self._get_input_params(options, shots=shots)
@@ -369,18 +367,11 @@ def run(
         if self._can_send_shots_input_param():
             shots_count = input_params.get(self.__class__._SHOTS_PARAM_NAME)
 
-        job_name = ""
-        if len(circuits) > 1:
-            job_name = f"batch-{len(circuits)}"
-            if shots_count is not None:
-                job_name = f"{job_name}-{shots_count}"
-        else:
-            job_name = circuits[0].name
-        job_name = options.pop("job_name", job_name)
+        job_name = options.pop("job_name", circuit.name)
 
-        metadata = options.pop("metadata", self._prepare_job_metadata(circuits))
+        metadata = options.pop("metadata", self._prepare_job_metadata(circuit))
 
-        input_data = self._translate_input(circuits, input_params)
+        input_data = self._translate_input(circuit, input_params)
 
         job = super()._run(job_name, input_data, input_params, metadata, **options)
         logger.info(
@@ -389,28 +380,19 @@ def run(
 
         return job
 
-    def _prepare_job_metadata(self, circuits: List[QuantumCircuit]) -> Dict[str, str]:
+    def _prepare_job_metadata(self, circuit: QuantumCircuit) -> Dict[str, str]:
         """Returns the metadata relative to the given circuits that will be attached to the Job"""
-        if len(circuits) == 1:
-            circuit: QuantumCircuit = circuits[0]
-            return {
-                "qiskit": str(True),
-                "name": circuit.name,
-                "num_qubits": circuit.num_qubits,
-                "metadata": json.dumps(circuit.metadata),
-            }
-        # for batch jobs, we don't want to store the metadata of each circuit
-        # we fill out the result header in output processing.
-        # These headers don't matter for execution are are only used for
-        # result processing.
-        return {}
-
-    def _generate_qir(
-        self, circuits: List[QuantumCircuit], target_profile: TargetProfile, **kwargs
-    ) -> pyqir.Module:
-
-        if len(circuits) == 0:
-            raise ValueError("No QuantumCircuits provided")
+        return {
+            "qiskit": str(True),
+            "name": circuit.name,
+            "num_qubits": circuit.num_qubits,
+            "metadata": json.dumps(circuit.metadata),
+        }
+
+
+    def _get_qir_str(
+        self, circuit: QuantumCircuit, target_profile: TargetProfile, **kwargs
+    ) -> str:
 
         config = self.configuration()
         # Barriers aren't removed by transpilation and must be explicitly removed in the Qiskit to QIR translation.
@@ -424,72 +406,36 @@ def _generate_qir(
             **kwargs,
         )
 
-        name = "batch"
-        if len(circuits) == 1:
-            name = circuits[0].name
-
-        if isinstance(circuits, list):
-            for value in circuits:
-                if not isinstance(value, QuantumCircuit):
-                    raise ValueError("Input must be List[QuantumCircuit]")
-        else:
-            raise ValueError("Input must be List[QuantumCircuit]")
-
-        context = pyqir.Context()
-        llvm_module = pyqir.qir_module(context, name)
-        for circuit in circuits:
-            qir_str = backend.qir(circuit)
-            module = pyqir.Module.from_ir(context, qir_str)
-            entry_point = next(filter(pyqir.is_entry_point, module.functions))
-            entry_point.name = circuit.name
-            llvm_module.link(module)
-        err = llvm_module.verify()
-        if err is not None:
-            raise Exception(err)
-
-        return llvm_module
+        qir_str = backend.qir(circuit)
+
+        return qir_str
 
-    def _get_qir_str(
-        self,
-        circuits: List[QuantumCircuit],
-        target_profile: TargetProfile,
-        **to_qir_kwargs,
-    ) -> str:
-        module = self._generate_qir(circuits, target_profile, **to_qir_kwargs)
-        return str(module)
 
     def _translate_input(
-        self, circuits: Union[QuantumCircuit, List[QuantumCircuit]], input_params: Dict[str, Any]
+        self, circuit: QuantumCircuit, input_params: Dict[str, Any]
     ) -> bytes:
         """Translates the input values to the QIR expected by the Backend."""
         logger.info(f"Using QIR as the job's payload format.")
-        if not (isinstance(circuits, list)):
-            circuits = [circuits]
 
         target_profile = self._get_target_profile(input_params)
 
         if logger.isEnabledFor(logging.DEBUG):
-            qir = self._get_qir_str(circuits, target_profile, skip_transpilation=True)
+            qir = self._get_qir_str(circuit, target_profile, skip_transpilation=True)
             logger.debug(f"QIR:\n{qir}")
 
         # We'll transpile automatically to the supported gates in QIR unless explicitly skipped.
         skip_transpilation = input_params.pop("skipTranspile", False)
 
-        module = self._generate_qir(
-            circuits, target_profile, skip_transpilation=skip_transpilation
+        qir_str = self._get_qir_str(
+            circuit, target_profile, skip_transpilation=skip_transpilation
         )
 
-        def get_func_name(func: pyqir.Function) -> str:
-            return func.name
-
-        entry_points = list(
-            map(get_func_name, filter(pyqir.is_entry_point, module.functions))
-        )
+        entry_points = ["ENTTRYPOINT_main"]
 
         if not skip_transpilation:
             # We'll only log the QIR again if we performed a transpilation.
             if logger.isEnabledFor(logging.DEBUG):
-                qir = str(module)
+                qir = str(qir_str)
                 logger.debug(f"QIR (Post-transpilation):\n{qir}")
 
         if "items" not in input_params:
@@ -498,7 +444,7 @@ def get_func_name(func: pyqir.Function) -> str:
                 {"entryPoint": name, "arguments": arguments} for name in entry_points
             ]
 
-        return str(module).encode("utf-8")
+        return qir_str.encode("utf-8")
 
     def _get_target_profile(self, input_params) -> TargetProfile:
         # Default to Adaptive_RI if not specified on the backend

azure-quantum/azure/quantum/target/target.py

@@ -2,7 +2,7 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License.
 ##
-from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union, Type,  Protocol, runtime_checkable
+from typing import TYPE_CHECKING, Any, Dict, Union, Type,  Protocol, runtime_checkable
 from dataclasses import dataclass
 import io
 import json
@@ -339,54 +339,3 @@ def _get_azure_target_id(self) -> str:
 
     def _get_azure_provider_id(self) -> str:
         return self.provider_id
-
-    @classmethod
-    def _calculate_qir_module_gate_stats(self, qir_module) -> GateStats:
-        try:
-            from pyqir import Module, is_qubit_type, is_result_type, entry_point, is_entry_point, Function
-
-        except ImportError:
-            raise ImportError(
-                "Missing optional 'qiskit' dependencies. \
-        To install run: pip install azure-quantum[qiskit]"
-            )
-
-        module: Module = qir_module
-
-        one_qubit_gates = 0
-        multi_qubit_gates = 0
-        measurement_gates = 0
-
-        function_entry_points: list[Function] = filter(is_entry_point, module.functions)
-
-        # Iterate over the blocks and their instructions
-        for function in function_entry_points:
-            for block in function.basic_blocks:
-                for instruction in block.instructions:
-                    qubit_count = 0
-                    result_count = 0
-
-                    # If the instruction is of type quantum rt, do not include this is the price calculation
-                    if len(instruction.operands) > 0 and "__quantum__rt" not in instruction.operands[-1].name:
-                        # Check each operand in the instruction
-                        for operand in instruction.operands:
-                            value_type = operand.type
-
-                            if is_qubit_type(value_type):
-                                qubit_count += 1
-                            elif is_result_type(value_type):
-                                result_count += 1
-
-                    # Determine the type of gate based on the counts
-                    if qubit_count == 1 and result_count == 0:
-                        one_qubit_gates += 1
-                    if qubit_count >= 2 and result_count == 0:
-                        multi_qubit_gates += 1
-                    if result_count > 0:
-                        measurement_gates += 1
-
-        return GateStats (
-            one_qubit_gates, 
-            multi_qubit_gates, 
-            measurement_gates
-        )

azure-quantum/requirements-qiskit.txt

@@ -1,5 +1,4 @@
 qiskit-ionq>=0.5,<0.6
 qsharp[qiskit]>=1.9.0,<2.0
-pyqir>=0.10.6,<0.11
 Markdown>=3.4.1,<4.0
 python-markdown-math>=0.8.0,<1.0

azure-quantum/tests/unit/test_qiskit.py

@@ -127,7 +127,7 @@ def _default_options(cls):
         return None
 
     def _translate_input(
-        self, circuits: List[QuantumCircuit], input_params: Dict[str, Any]
+        self, circuit: QuantumCircuit, input_params: Dict[str, Any]
     ) -> bytes:
         return None
 
@@ -436,7 +436,7 @@ def test_plugins_submit_qiskit_multi_circuit_experiment_to_ionq(self):
 
         with pytest.raises(NotImplementedError) as exc:
             backend.run(circuit=[circuit, circuit], shots=500)
-        self.assertEqual(str(exc.value), "This backend only supports running a maximum of 1 circuits per job.")
+        self.assertEqual(str(exc.value), "This backend only supports running a single circuit per job.")
 
     @pytest.mark.ionq
     @pytest.mark.live_test
@@ -1111,7 +1111,7 @@ def test_plugins_submit_qiskit_multi_circuit_experiment_to_quantinuum(self):
 
         with self.assertRaises(NotImplementedError) as context:
             backend.run(circuit=[circuit, circuit], shots=None)
-        self.assertEqual(str(context.exception), "This backend only supports running a maximum of 1 circuits per job.")
+        self.assertEqual(str(context.exception), "This backend only supports running a single circuit per job.")
 
     @pytest.mark.quantinuum
     @pytest.mark.live_test