Coverage for /home/runner/work/tket/tket/pytket/pytket/backends/backend.py: 81%
194 statements
« prev ^ index » next coverage.py v7.8.2, created at 2025-06-02 12:44 +0000
1# Copyright Quantinuum
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
15"""Abstract base class for all Backend encapsulations."""
17import warnings
18from abc import ABC, abstractmethod
19from collections.abc import Iterable, Sequence
20from importlib import import_module
21from types import ModuleType
22from typing import Any, Literal, cast, overload
24from pytket.circuit import Bit, Circuit, OpType
25from pytket.passes import BasePass
26from pytket.pauli import QubitPauliString
27from pytket.predicates import Predicate
28from pytket.utils import QubitPauliOperator
29from pytket.utils.outcomearray import OutcomeArray
30from pytket.utils.results import KwargTypes
32from .backend_exceptions import (
33 CircuitNotRunError,
34 CircuitNotValidError,
35)
36from .backendinfo import BackendInfo
37from .backendresult import BackendResult
38from .resulthandle import ResultHandle, _ResultIdTuple
39from .status import CircuitStatus
42class ResultHandleTypeError(Exception):
43 """Wrong result handle type."""
46class Backend(ABC):
47 """
48 This abstract class defines the structure of a backend as something that
49 can run quantum circuits and produce output as at least one of shots,
50 counts, state, or unitary
51 """
53 _supports_shots = False
54 _supports_counts = False
55 _supports_state = False
56 _supports_unitary = False
57 _supports_density_matrix = False
58 _supports_expectation = False
59 _expectation_allows_nonhermitian = False
60 _supports_contextual_optimisation = False
61 _persistent_handles = False
63 def __init__(self) -> None:
64 self._cache: dict[ResultHandle, dict[str, Any]] = {}
66 @staticmethod
67 def empty_result(circuit: Circuit, n_shots: int) -> BackendResult:
68 """
69 Creates a :py:class:`~.BackendResult` mimicking the outcome where every
70 bit is 0 for every shot.
71 """
72 n_bits = len(circuit.bits)
73 empty_readouts = [[0] * n_bits for _ in range(n_shots)]
74 shots = OutcomeArray.from_readouts(empty_readouts)
75 c_bits = [Bit(index) for index in range(n_bits)]
76 return BackendResult(shots=shots, c_bits=c_bits)
78 @property
79 @abstractmethod
80 def required_predicates(self) -> list[Predicate]:
81 """
82 The minimum set of predicates that a circuit must satisfy before it can
83 be successfully run on this backend.
85 :return: Required predicates.
86 """
87 ...
89 def valid_circuit(self, circuit: Circuit) -> bool:
90 """
91 Checks that the circuit satisfies all of required_predicates.
93 :param circuit: The circuit to check.
94 :return: Whether or not all of required_predicates are satisfied.
95 """
96 return all(pred.verify(circuit) for pred in self.required_predicates)
98 def _check_all_circuits(
99 self, circuits: Iterable[Circuit], nomeasure_warn: bool | None = None
100 ) -> bool:
101 if nomeasure_warn is None: 101 ↛ 108line 101 didn't jump to line 108 because the condition on line 101 was always true
102 nomeasure_warn = not (
103 self._supports_state
104 or self._supports_unitary
105 or self._supports_density_matrix
106 or self._supports_expectation
107 )
108 for i, circ in enumerate(circuits):
109 errors = (
110 CircuitNotValidError(i, repr(pred))
111 for pred in self.required_predicates
112 if not pred.verify(circ)
113 )
114 for error in errors: 114 ↛ 115line 114 didn't jump to line 115 because the loop on line 114 never started
115 raise error
116 if nomeasure_warn and circ.n_gates_of_type(OpType.Measure) < 1: 116 ↛ 117line 116 didn't jump to line 117 because the condition on line 116 was never true
117 warnings.warn( # noqa: B028
118 f"Circuit with index {i} in submitted does not contain a "
119 "measure operation."
120 )
121 return True
123 @abstractmethod
124 def rebase_pass(self) -> BasePass:
125 """
126 A single compilation pass that when run converts all gates in a Circuit to
127 an OpType supported by the Backend (ignoring architecture constraints).
129 :return: Compilation pass that converts gates to primitives supported by
130 Backend.
131 """
132 ...
134 @abstractmethod
135 def default_compilation_pass(self, optimisation_level: int = 2) -> BasePass:
136 """
137 A suggested compilation pass that will will, if possible, produce an equivalent
138 circuit suitable for running on this backend.
140 At a minimum it will ensure that compatible gates are used and that all two-
141 qubit interactions are compatible with the backend's qubit architecture. At
142 higher optimisation levels, further optimisations may be applied.
144 This is a an abstract method which is implemented in the backend itself, and so
145 is tailored to the backend's requirements.
147 :param optimisation_level: The level of optimisation to perform during
148 compilation.
150 - Level 0 does the minimum required to solves the device constraints,
151 without any optimisation.
152 - Level 1 additionally performs some light optimisations.
153 - Level 2 (the default) adds more computationally intensive optimisations
154 that should give the best results from execution.
156 :return: Compilation pass guaranteeing required predicates.
157 """
158 ...
160 def get_compiled_circuit(
161 self, circuit: Circuit, optimisation_level: int = 2
162 ) -> Circuit:
163 """
164 Return a single circuit compiled with :py:meth:`default_compilation_pass`. See
165 :py:meth:`Backend.get_compiled_circuits`.
166 """
167 return_circuit = circuit.copy()
168 self.default_compilation_pass(optimisation_level).apply(return_circuit)
169 return return_circuit
171 def get_compiled_circuits(
172 self, circuits: Sequence[Circuit], optimisation_level: int = 2
173 ) -> list[Circuit]:
174 """Compile a sequence of circuits with :py:meth:`default_compilation_pass`
175 and return the list of compiled circuits (does not act in place).
177 As well as applying a degree of optimisation (controlled by the
178 `optimisation_level` parameter), this method tries to ensure that the circuits
179 can be run on the backend (i.e. successfully passed to
180 :py:meth:`process_circuits`), for example by rebasing to the supported gate set,
181 or routing to match the connectivity of the device. However, this is not always
182 possible, for example if the circuit contains classical operations that are not
183 supported by the backend. You may use :py:meth:`valid_circuit` to check whether
184 the circuit meets the backend's requirements after compilation. This validity
185 check is included in :py:meth:`process_circuits` by default, before any circuits
186 are submitted to the backend.
188 If the validity check fails, you can obtain more information about the failure
189 by iterating through the predicates in the `required_predicates` property of the
190 backend, and running the :py:meth:`~.Predicate.verify` method on each in turn with your
191 circuit.
193 :param circuits: The circuits to compile.
194 :param optimisation_level: The level of optimisation to perform during
195 compilation. See :py:meth:`default_compilation_pass` for a description of
196 the different levels (0, 1 or 2). Defaults to 2.
197 :return: Compiled circuits.
198 """
199 return [self.get_compiled_circuit(c, optimisation_level) for c in circuits]
201 @property
202 @abstractmethod
203 def _result_id_type(self) -> _ResultIdTuple:
204 """Identifier type signature for ResultHandle for this backend.
206 :return: Type signature (tuple of hashable types)
207 """
208 ...
210 def _check_handle_type(self, reshandle: ResultHandle) -> None:
211 """Check a result handle is valid for this backend, raises TypeError if not.
213 :param reshandle: Handle to check
214 :raises TypeError: Types of handle identifiers don't match those of backend.
215 """
216 if (len(reshandle) != len(self._result_id_type)) or not all(
217 isinstance(idval, ty)
218 for idval, ty in zip(reshandle, self._result_id_type, strict=False)
219 ):
220 raise ResultHandleTypeError(
221 f"{reshandle!r} does not match expected "
222 f"identifier types {self._result_id_type}"
223 )
225 def process_circuit(
226 self,
227 circuit: Circuit,
228 n_shots: int | None = None,
229 valid_check: bool = True,
230 **kwargs: KwargTypes,
231 ) -> ResultHandle:
232 """
233 Submit a single circuit to the backend for running. See
234 :py:meth:`Backend.process_circuits`.
235 """
237 return self.process_circuits(
238 [circuit], n_shots=n_shots, valid_check=valid_check, **kwargs
239 )[0]
241 @abstractmethod
242 def process_circuits(
243 self,
244 circuits: Sequence[Circuit],
245 n_shots: int | Sequence[int] | None = None,
246 valid_check: bool = True,
247 **kwargs: KwargTypes,
248 ) -> list[ResultHandle]:
249 """
250 Submit circuits to the backend for running. The results will be stored
251 in the backend's result cache to be retrieved by the corresponding
252 get_<data> method.
254 If the `postprocess` keyword argument is set to True, and the backend supports
255 the feature (see :py:meth:`supports_contextual_optimisation`), then contextual
256 optimisatioons are applied before running the circuit and retrieved results will
257 have any necessary classical postprocessing applied. This is not enabled by
258 default.
260 Use keyword arguments to specify parameters to be used in submitting circuits
261 See specific Backend derived class for available parameters, from the following
262 list:
264 * `seed`: RNG seed for simulators
265 * `postprocess`: if True, apply contextual optimisations
267 Note: If a backend is reused many times, the in-memory results cache grows
268 indefinitely. Therefore, when processing many circuits on a statevector or
269 unitary backend (whose results may occupy significant amounts of memory), it is
270 advisable to run :py:meth:`Backend.empty_cache` after each result is retrieved.
272 :param circuits: Circuits to process on the backend.
273 :param n_shots: Number of shots to run per circuit. Optionally, this can be
274 a list of shots specifying the number of shots for each circuit separately.
275 None is to be used for state/unitary simulators. Defaults to None.
276 :param valid_check: Explicitly check that all circuits satisfy all required
277 predicates to run on the backend. Defaults to True
278 :return: Handles to results for each input circuit, as an interable in
279 the same order as the circuits.
280 """
281 ...
283 @abstractmethod
284 def circuit_status(self, handle: ResultHandle) -> CircuitStatus:
285 """
286 Return a CircuitStatus reporting the status of the circuit execution
287 corresponding to the ResultHandle
288 """
289 ...
291 def empty_cache(self) -> None:
292 """Manually empty the result cache on the backend."""
293 self._cache = {}
295 def pop_result(self, handle: ResultHandle) -> dict[str, Any] | None:
296 """Remove cache entry corresponding to handle from the cache and return.
298 :param handle: ResultHandle object
299 :return: Cache entry corresponding to handle, if it was present
300 """
301 return self._cache.pop(handle, None)
303 def get_result(self, handle: ResultHandle, **kwargs: KwargTypes) -> BackendResult:
304 """Return a BackendResult corresponding to the handle.
306 Use keyword arguments to specify parameters to be used in retrieving results.
307 See specific Backend derived class for available parameters, from the following
308 list:
310 * `timeout`: maximum time to wait for remote job to finish
311 * `wait`: polling interval between remote calls to check job status
313 :param handle: handle to results
314 :return: Results corresponding to handle.
315 """
316 self._check_handle_type(handle)
317 if handle in self._cache and "result" in self._cache[handle]:
318 return cast("BackendResult", self._cache[handle]["result"])
319 raise CircuitNotRunError(handle)
321 def get_results(
322 self, handles: Iterable[ResultHandle], **kwargs: KwargTypes
323 ) -> list[BackendResult]:
324 """Return results corresponding to handles.
326 :param handles: Iterable of handles
327 :return: List of results
329 Keyword arguments are as for `get_result`, and apply to all jobs.
330 """
331 try:
332 return [self.get_result(handle, **kwargs) for handle in handles]
333 except ResultHandleTypeError as e:
334 try:
335 self._check_handle_type(cast("ResultHandle", handles))
336 except ResultHandleTypeError:
337 raise e # noqa: B904
339 raise ResultHandleTypeError(
340 "Possible use of single ResultHandle"
341 " where sequence of ResultHandles was expected."
342 ) from e
344 def run_circuit(
345 self,
346 circuit: Circuit,
347 n_shots: int | None = None,
348 valid_check: bool = True,
349 **kwargs: KwargTypes,
350 ) -> BackendResult:
351 """
352 Submits a circuit to the backend and returns results
354 :param circuit: Circuit to be executed
355 :param n_shots: Passed on to :py:meth:`Backend.process_circuit`
356 :param valid_check: Passed on to :py:meth:`Backend.process_circuit`
357 :return: Result
359 This is a convenience method equivalent to calling
360 :py:meth:`Backend.process_circuit` followed by :py:meth:`Backend.get_result`.
361 Any additional keyword arguments are passed on to
362 :py:meth:`Backend.process_circuit` and :py:meth:`Backend.get_result`.
363 """
364 return self.run_circuits(
365 [circuit], n_shots=n_shots, valid_check=valid_check, **kwargs
366 )[0]
368 def run_circuits(
369 self,
370 circuits: Sequence[Circuit],
371 n_shots: int | Sequence[int] | None = None,
372 valid_check: bool = True,
373 **kwargs: KwargTypes,
374 ) -> list[BackendResult]:
375 """
376 Submits circuits to the backend and returns results
378 :param circuits: Sequence of Circuits to be executed
379 :param n_shots: Passed on to :py:meth:`Backend.process_circuits`
380 :param valid_check: Passed on to :py:meth:`Backend.process_circuits`
381 :return: List of results
383 This is a convenience method equivalent to calling
384 :py:meth:`Backend.process_circuits` followed by :py:meth:`Backend.get_results`.
385 Any additional keyword arguments are passed on to
386 :py:meth:`Backend.process_circuits` and :py:meth:`Backend.get_results`.
387 """
388 handles = self.process_circuits(circuits, n_shots, valid_check, **kwargs)
389 results = self.get_results(handles, **kwargs)
390 for h in handles:
391 self.pop_result(h)
392 return results
394 def cancel(self, handle: ResultHandle) -> None:
395 """
396 Cancel a job.
398 :param handle: handle to job
399 :raises NotImplementedError: If backend does not support job cancellation
400 """
401 raise NotImplementedError("Backend does not support job cancellation.")
403 @property
404 def backend_info(self) -> BackendInfo | None:
405 """Retrieve all Backend properties in a BackendInfo object, including
406 device architecture, supported gate set, gate errors and other hardware-specific
407 information.
409 :return: The BackendInfo describing this backend if it exists.
410 """
411 raise NotImplementedError("Backend does not provide any device properties.")
413 @classmethod
414 def available_devices(cls, **kwargs: Any) -> list[BackendInfo]:
415 """Retrieve all available devices as a list of BackendInfo objects, including
416 device name, architecture, supported gate set, gate errors,
417 and other hardware-specific information.
419 :return: A list of BackendInfo objects describing available devices.
420 """
421 raise NotImplementedError(
422 "Backend does not provide information about available devices."
423 )
425 @property
426 def persistent_handles(self) -> bool:
427 """
428 Whether the backend produces `ResultHandle` objects that can be reused with
429 other instances of the backend class.
430 """
431 return self._persistent_handles
433 @property
434 def supports_shots(self) -> bool:
435 """
436 Does this backend support shot result retrieval via
437 :py:meth:`~.BackendResult.get_shots`.
438 """
439 return self._supports_shots
441 @property
442 def supports_counts(self) -> bool:
443 """
444 Does this backend support counts result retrieval via
445 :py:meth:`~.BackendResult.get_counts`.
446 """
447 return self._supports_counts
449 @property
450 def supports_state(self) -> bool:
451 """
452 Does this backend support statevector retrieval via
453 :py:meth:`~.BackendResult.get_state`.
454 """
455 return self._supports_state
457 @property
458 def supports_unitary(self) -> bool:
459 """
460 Does this backend support unitary retrieval via
461 :py:meth:`~.BackendResult.get_unitary`.
462 """
463 return self._supports_unitary
465 @property
466 def supports_density_matrix(self) -> bool:
467 """Does this backend support density matrix retrieval via
468 `get_density_matrix`."""
469 return self._supports_density_matrix
471 @property
472 def supports_expectation(self) -> bool:
473 """Does this backend support expectation value calculation for operators."""
474 return self._supports_expectation
476 @property
477 def expectation_allows_nonhermitian(self) -> bool:
478 """If expectations are supported, is the operator allowed to be non-Hermitan?"""
479 return self._expectation_allows_nonhermitian
481 @property
482 def supports_contextual_optimisation(self) -> bool:
483 """Does this backend support contextual optimisation?
485 See :py:meth:`process_circuits`."""
486 return self._supports_contextual_optimisation
488 def _get_extension_module(self) -> ModuleType | None:
489 """Return the extension module of the backend if it belongs to a
490 pytket-extension package.
492 :return: The extension module of the backend if it belongs to a pytket-extension
493 package.
494 """
495 mod_parts = self.__class__.__module__.split(".")[:3]
496 if not (mod_parts[0] == "pytket" and mod_parts[1] == "extensions"):
497 return None
498 return import_module(".".join(mod_parts))
500 @property
501 def __extension_name__(self) -> str | None:
502 """Retrieve the extension name of the backend if it belongs to a
503 pytket-extension package.
505 :return: The extension name of the backend if it belongs to a pytket-extension
506 package.
507 """
508 try:
509 return self._get_extension_module().__extension_name__ # type: ignore
510 except AttributeError:
511 return None
513 @property
514 def __extension_version__(self) -> str | None:
515 """Retrieve the extension version of the backend if it belongs to a
516 pytket-extension package.
518 :return: The extension version of the backend if it belongs to a
519 pytket-extension package.
520 """
521 try:
522 return self._get_extension_module().__extension_version__ # type: ignore
523 except AttributeError:
524 return None
526 @overload
527 @staticmethod
528 def _get_n_shots_as_list( 528 ↛ exitline 528 didn't return from function '_get_n_shots_as_list' because
529 n_shots: None | int | Sequence[int | None],
530 n_circuits: int,
531 optional: Literal[False],
532 ) -> list[int]: ...
534 @overload
535 @staticmethod
536 def _get_n_shots_as_list( 536 ↛ exitline 536 didn't return from function '_get_n_shots_as_list' because
537 n_shots: None | int | Sequence[int | None],
538 n_circuits: int,
539 optional: Literal[True],
540 set_zero: Literal[True],
541 ) -> list[int]: ...
543 @overload
544 @staticmethod
545 def _get_n_shots_as_list( 545 ↛ exitline 545 didn't return from function '_get_n_shots_as_list' because
546 n_shots: None | int | Sequence[int | None],
547 n_circuits: int,
548 optional: bool = True,
549 set_zero: bool = False,
550 ) -> list[int | None] | list[int]: ...
552 @staticmethod
553 def _get_n_shots_as_list(
554 n_shots: None | int | Sequence[int | None],
555 n_circuits: int,
556 optional: bool = True,
557 set_zero: bool = False,
558 ) -> list[int | None] | list[int]:
559 """
560 Convert any admissible n_shots value into List[Optional[int]] format.
562 This validates the n_shots argument for process_circuits. If a single
563 value is passed, this value is broadcast to the number of circuits.
564 Additional boolean flags control how the argument is validated.
565 Raises an exception if n_shots is in an invalid format.
567 :param n_shots: The argument to be validated.
568 :param n_circuits: Length of the converted argument returned.
569 :param optional: Whether n_shots can be None (default: True).
570 :param set_zero: Whether None values should be set to 0 (default: False).
571 :return: a list of length `n_circuits`, the converted argument
572 """
574 n_shots_list: list[int | None] = []
576 def validate_n_shots(n: int | None) -> bool:
577 return optional or (n is not None and n > 0)
579 if set_zero and not optional: 579 ↛ 580line 579 didn't jump to line 580 because the condition on line 579 was never true
580 raise ValueError("set_zero cannot be true when optional is false")
582 if hasattr(n_shots, "__iter__"):
583 assert not isinstance(n_shots, int)
584 assert n_shots is not None
586 if not all(map(validate_n_shots, n_shots)):
587 raise ValueError(
588 "n_shots values are required for all circuits for this backend"
589 )
590 n_shots_list = list(n_shots)
591 else:
592 assert n_shots is None or isinstance(n_shots, int)
594 if not validate_n_shots(n_shots):
595 raise ValueError("Parameter n_shots is required for this backend")
596 # convert n_shots to a list
597 n_shots_list = [n_shots] * n_circuits
599 if len(n_shots_list) != n_circuits:
600 raise ValueError("The length of n_shots and circuits must match")
602 if set_zero:
603 # replace None with 0
604 n_shots_list = [n or 0 for n in n_shots_list]
606 return n_shots_list
608 def get_pauli_expectation_value(
609 self, state_circuit: Circuit, pauli: QubitPauliString
610 ) -> complex:
611 """
612 Calculates the expectation value of the given circuit using
613 functionality built into the backend.
615 Raises an exception if the backend does not provide custom expectation
616 value features.
617 """
618 raise NotImplementedError
620 def get_operator_expectation_value(
621 self, state_circuit: Circuit, operator: QubitPauliOperator
622 ) -> complex:
623 """
624 Calculates the expectation value of the given circuit with respect to
625 the operator using functionality built into the backend.
627 Raises an exception if the backend does not provide custom expectation
628 value features.
629 """
630 raise NotImplementedError