remove remote simulator (#301)

* remove IBMQEmulatorbackend, replace with local

* remove testcase, local fails supports_contextual_optimisation

* remove not working simulator_stabilizer

* rename IBMQLocalEmulatorBackend to IBMQEmulatorBackend

* clean up

* update changelog

* fix mypy

* remove for loops

* rename brisbane_local_emulator_backend to brisbane_emulator_backend

* update changelog

* clean up format

* rename ibmq_(local_)emulator file

docs/api.rst

@@ -11,11 +11,6 @@ API documentation
     :show-inheritance:
     :members:
 
-.. autoclass:: pytket.extensions.qiskit.IBMQLocalEmulatorBackend
-    :special-members: __init__
-    :show-inheritance:
-    :members:
-
 .. autoclass:: pytket.extensions.qiskit.AerBackend
     :special-members: __init__
     :inherited-members:

docs/changelog.rst

@@ -5,6 +5,9 @@ Unreleased
 ----------
 
 * Update qiskit-ibm-runtime version requirement to 0.22.
+* remove all remote simulators
+* rename ``IBMQLocalEmulatorBackend`` to ``IBMQEmulatorBackend``
+* ``IBMQEmulatorBackend`` will now run locally
 
 0.50.0 (March 2024)
 -------------------

docs/intro.txt

@@ -29,7 +29,6 @@ Available IBM Backends
 
     IBMQBackend
     IBMQEmulatorBackend
-    IBMQLocalEmulatorBackend
     AerBackend
     AerStateBackend
     AerUnitaryBackend
@@ -95,7 +94,7 @@ to disk:
     IBMProvider.save_account(token=ibm_token)
     QiskitRuntimeService.save_account(channel="ibm_quantum", token=ibm_token)
 
-To see which devices you can access you can use the ``available_devices`` method on the :py:class:`IBMQBackend` or :py:class:`IBMQEmulatorBackend`. Note that it is possible to pass optional ``instance`` and ``provider`` arguments to this method. This allows you to see which devices are accessible through your IBM hub.
+To see which devices you can access you can use the ``available_devices`` method on the :py:class:`IBMQBackend`. Note that it is possible to pass optional ``instance`` and ``provider`` arguments to this method. This allows you to see which devices are accessible through your IBM hub.
 
 ::
 
@@ -160,7 +159,7 @@ Default Compilation
 
 Every :py:class:`Backend` in pytket has its own ``default_compilation_pass`` method. This method applies a sequence of optimisations to a circuit depending on the value of an ``optimisation_level`` parameter. This default compilation will ensure that the circuit meets all the constraints required to run on the :py:class:`Backend`. The passes applied by different levels of optimisation are specified in the table below.
 
-.. list-table:: **Default compilation pass for the IBMQBackend, IBMQEmulatorBackend and IBMQLocalEmulatorBackend**
+.. list-table:: **Default compilation pass for the IBMQBackend and IBMQEmulatorBackend**
    :widths: 25 25 25
    :header-rows: 1
 

pytket/extensions/qiskit/__init__.py

@@ -22,7 +22,6 @@
     AerStateBackend,
     AerUnitaryBackend,
     IBMQEmulatorBackend,
-    IBMQLocalEmulatorBackend,
 )
 from .backends.config import set_ibmq_config
 from .qiskit_convert import qiskit_to_tk, tk_to_qiskit, process_characterisation

pytket/extensions/qiskit/backends/__init__.py

@@ -16,4 +16,3 @@
 from .ibm import IBMQBackend, NoIBMQCredentialsError
 from .aer import AerBackend, AerStateBackend, AerUnitaryBackend
 from .ibmq_emulator import IBMQEmulatorBackend
-from .ibmq_local_emulator import IBMQLocalEmulatorBackend

pytket/extensions/qiskit/backends/ibm.py

@@ -355,8 +355,7 @@ def default_compilation_pass(
         This is a an abstract method which is implemented in the backend itself, and so
         is tailored to the backend's requirements.
 
-        The default compilation passes for the :py:class:`IBMQBackend`,
-        :py:class:`IBMQEmulatorBackend` and the
+        The default compilation passes for the :py:class:`IBMQBackend` and the
         Aer simulators support an optional ``placement_options`` dictionary containing
         arguments to override the default settings in :py:class:`NoiseAwarePlacement`.
 

pytket/extensions/qiskit/backends/ibmq_emulator.py

@@ -12,57 +12,47 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from ast import literal_eval
 from collections import Counter
-import itertools
-import json
 from typing import (
-    cast,
     Dict,
     Optional,
     List,
     Sequence,
     Tuple,
     Union,
 )
-from warnings import warn
 
 from qiskit_aer.noise.noise_model import NoiseModel  # type: ignore
-from qiskit_ibm_runtime import (  # type: ignore
-    QiskitRuntimeService,
-    Session,
-    Options,
-    Sampler,
-    RuntimeJob,
-)
+
 from qiskit_ibm_provider import IBMProvider  # type: ignore
 
-from pytket.backends import Backend, CircuitNotRunError, ResultHandle, CircuitStatus
+from pytket.backends import (
+    Backend,
+    ResultHandle,
+    CircuitStatus,
+)
 from pytket.backends.backendinfo import BackendInfo
 from pytket.backends.backendresult import BackendResult
 from pytket.backends.resulthandle import _ResultIdTuple
-from pytket.circuit import Bit, Circuit, OpType
-from pytket.extensions.qiskit.qiskit_convert import tk_to_qiskit
+from pytket.circuit import Circuit
 from pytket.passes import BasePass
 from pytket.predicates import Predicate
-from pytket.utils import prepare_circuit
-from pytket.utils.outcomearray import OutcomeArray
 from pytket.utils.results import KwargTypes
 
+from .aer import AerBackend
 from .ibm import IBMQBackend
-from .ibm_utils import _STATUS_MAP, _batch_circuits
 
 
 class IBMQEmulatorBackend(Backend):
-    """A Backend which uses the ibmq_qasm_simulator to emulate the behaviour of
+    """A backend which uses the AerBackend to loaclly emulate the behaviour of
     IBMQBackend. Identical to :py:class:`IBMQBackend` except there is no `monitor`
     parameter. Performs the same compilation and predicate checks as IBMQBackend.
-    Requires a valid IBMQ account.
+    Requires a valid IBM account.
     """
 
     _supports_shots = False
     _supports_counts = True
-    _supports_contextual_optimisation = True
+    _supports_contextual_optimisation = False
     _persistent_handles = False
     _supports_expectation = False
 
@@ -81,14 +71,12 @@ def __init__(
             token=token,
         )
 
-        self._service = QiskitRuntimeService(
-            channel="ibm_quantum", instance=instance, token=token
-        )
-        self._session = Session(service=self._service, backend="ibmq_qasm_simulator")
-
         # Get noise model:
         self._noise_model = NoiseModel.from_backend(self._ibmq._backend)
 
+        # Construct AerBackend based on noise model:
+        self._aer = AerBackend(noise_model=self._noise_model)
+
         # cache of results keyed by job id and circuit index
         self._ibm_res_cache: Dict[Tuple[str, int], Counter] = dict()
 
@@ -112,8 +100,7 @@ def default_compilation_pass(
 
     @property
     def _result_id_type(self) -> _ResultIdTuple:
-        # job ID, index, stringified sequence of measured bits, post-processing circuit
-        return (str, int, str, str)
+        return self._aer._result_id_type
 
     def rebase_pass(self) -> BasePass:
         return self._ibmq.rebase_pass()
@@ -129,123 +116,21 @@ def process_circuits(
         See :py:meth:`pytket.backends.Backend.process_circuits`.
         Supported kwargs: `seed`, `postprocess`.
         """
-        circuits = list(circuits)
-        n_shots_list = Backend._get_n_shots_as_list(
-            n_shots,
-            len(circuits),
-            optional=False,
+        if valid_check:
+            self._ibmq._check_all_circuits(circuits)
+        return self._aer.process_circuits(
+            circuits, n_shots=n_shots, valid_check=False, **kwargs
         )
 
-        handle_list: List[Optional[ResultHandle]] = [None] * len(circuits)
-        seed = kwargs.get("seed")
-        circuit_batches, batch_order = _batch_circuits(circuits, n_shots_list)
-
-        batch_id = 0  # identify batches for debug purposes only
-        for (n_shots, batch), indices in zip(circuit_batches, batch_order):
-            for chunk in itertools.zip_longest(
-                *([iter(zip(batch, indices))] * self._ibmq._max_per_job)
-            ):
-                filtchunk = list(filter(lambda x: x is not None, chunk))
-                batch_chunk, indices_chunk = zip(*filtchunk)
-
-                if valid_check:
-                    self._check_all_circuits(batch_chunk)
-
-                postprocess = kwargs.get("postprocess", False)
-
-                qcs, c_bit_strs, ppcirc_strs = [], [], []
-                for tkc in batch_chunk:
-                    if postprocess:
-                        c0, ppcirc = prepare_circuit(tkc, allow_classical=False)
-                        ppcirc_rep = ppcirc.to_dict()
-                    else:
-                        c0, ppcirc_rep = tkc, None
-                    qcs.append(tk_to_qiskit(c0))
-                    measured_bits = sorted(
-                        [cmd.args[1] for cmd in tkc if cmd.op.type == OpType.Measure]
-                    )
-                    c_bit_strs.append(
-                        repr([(b.reg_name, b.index) for b in measured_bits])
-                    )
-                    ppcirc_strs.append(json.dumps(ppcirc_rep))
-                options = Options()
-                options.resilience_level = 0
-                options.execution.shots = n_shots
-                options.simulator.noise_model = self._noise_model
-                options.seed_simulator = seed
-                if type(seed) is int:
-                    seed += 1
-                sampler = Sampler(session=self._session, options=options)
-                job = sampler.run(circuits=qcs)
-                job_id = job.job_id()
-                for i, ind in enumerate(indices_chunk):
-                    handle_list[ind] = ResultHandle(
-                        job_id, i, c_bit_strs[i], ppcirc_strs[i]
-                    )
-            batch_id += 1
-        for handle in handle_list:
-            assert handle is not None
-            self._cache[handle] = dict()
-        return cast(List[ResultHandle], handle_list)
-
-    def _retrieve_job(self, jobid: str) -> RuntimeJob:
-        return self._service.job(jobid)
-
     def cancel(self, handle: ResultHandle) -> None:
-        jobid = cast(str, handle[0])
-        job = self._retrieve_job(jobid)
-        try:
-            job.cancel()
-        except Exception as e:
-            warn(f"Unable to cancel job {jobid}: {e}")
+        self._aer.cancel(handle)
 
     def circuit_status(self, handle: ResultHandle) -> CircuitStatus:
-        self._check_handle_type(handle)
-        jobid = cast(str, handle[0])
-        job = self._service.job(jobid)
-        ibmstatus = job.status()
-        return CircuitStatus(_STATUS_MAP[ibmstatus], ibmstatus.value)
+        return self._aer.circuit_status(handle)
 
     def get_result(self, handle: ResultHandle, **kwargs: KwargTypes) -> BackendResult:
         """
         See :py:meth:`pytket.backends.Backend.get_result`.
-        Supported kwargs: `timeout`, `wait`.
+        Supported kwargs: none.
         """
-        self._check_handle_type(handle)
-        if handle in self._cache:
-            cached_result = self._cache[handle]
-            if "result" in cached_result:
-                return cast(BackendResult, cached_result["result"])
-        jobid, index, c_bit_str, ppcirc_str = handle
-        c_bits = [Bit(reg_name, index) for reg_name, index in literal_eval(c_bit_str)]
-        ppcirc_rep = json.loads(ppcirc_str)
-        ppcirc = Circuit.from_dict(ppcirc_rep) if ppcirc_rep is not None else None
-        cache_key = (jobid, index)
-        if cache_key not in self._ibm_res_cache:
-            try:
-                job = self._retrieve_job(jobid)
-            except Exception as e:
-                warn(f"Unable to retrieve job {jobid}: {e}")
-                raise CircuitNotRunError(handle)
-
-            res = job.result(timeout=kwargs.get("timeout", None))
-            for circ_index, (r, d) in enumerate(zip(res.quasi_dists, res.metadata)):
-                self._ibm_res_cache[(jobid, circ_index)] = Counter(
-                    {n: int(0.5 + d["shots"] * p) for n, p in r.items()}
-                )
-
-        counts = self._ibm_res_cache[cache_key]  # Counter[int]
-        # Convert to `OutcomeArray`:
-        tket_counts: Counter = Counter()
-        for outcome_key, sample_count in counts.items():
-            array = OutcomeArray.from_ints(
-                ints=[outcome_key],
-                width=len(c_bits),
-                big_endian=False,
-            )
-            tket_counts[array] = sample_count
-        # Convert to `BackendResult`:
-        result = BackendResult(c_bits=c_bits, counts=tket_counts, ppcirc=ppcirc)
-
-        self._cache[handle] = {"result": result}
-        return result
+        return self._aer.get_result(handle)