Coverage for /home/runner/work/tket/tket/pytket/pytket/backends/backend.py: 81%
195 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-05-09 15:08 +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
41ResultCache = dict[str, Any]
44class ResultHandleTypeError(Exception):
45 """Wrong result handle type."""
48class Backend(ABC):
49 """
50 This abstract class defines the structure of a backend as something that
51 can run quantum circuits and produce output as at least one of shots,
52 counts, state, or unitary
53 """
55 _supports_shots = False
56 _supports_counts = False
57 _supports_state = False
58 _supports_unitary = False
59 _supports_density_matrix = False
60 _supports_expectation = False
61 _expectation_allows_nonhermitian = False
62 _supports_contextual_optimisation = False
63 _persistent_handles = False
65 def __init__(self) -> None:
66 self._cache: dict[ResultHandle, ResultCache] = {}
68 @staticmethod
69 def empty_result(circuit: Circuit, n_shots: int) -> BackendResult:
70 """
71 Creates a :py:class:`BackendResult` mimicking the outcome where every
72 bit is 0 for every shot.
73 """
74 n_bits = len(circuit.bits)
75 empty_readouts = [[0] * n_bits for _ in range(n_shots)]
76 shots = OutcomeArray.from_readouts(empty_readouts)
77 c_bits = [Bit(index) for index in range(n_bits)]
78 return BackendResult(shots=shots, c_bits=c_bits)
80 @property
81 @abstractmethod
82 def required_predicates(self) -> list[Predicate]:
83 """
84 The minimum set of predicates that a circuit must satisfy before it can
85 be successfully run on this backend.
87 :return: Required predicates.
88 :rtype: List[Predicate]
89 """
90 ...
92 def valid_circuit(self, circuit: Circuit) -> bool:
93 """
94 Checks that the circuit satisfies all of required_predicates.
96 :param circuit: The circuit to check.
97 :type circuit: Circuit
98 :return: Whether or not all of required_predicates are satisfied.
99 :rtype: bool
100 """
101 return all(pred.verify(circuit) for pred in self.required_predicates)
103 def _check_all_circuits(
104 self, circuits: Iterable[Circuit], nomeasure_warn: bool | None = None
105 ) -> bool:
106 if nomeasure_warn is None: 106 ↛ 113line 106 didn't jump to line 113 because the condition on line 106 was always true
107 nomeasure_warn = not (
108 self._supports_state
109 or self._supports_unitary
110 or self._supports_density_matrix
111 or self._supports_expectation
112 )
113 for i, circ in enumerate(circuits):
114 errors = (
115 CircuitNotValidError(i, repr(pred))
116 for pred in self.required_predicates
117 if not pred.verify(circ)
118 )
119 for error in errors: 119 ↛ 120line 119 didn't jump to line 120 because the loop on line 119 never started
120 raise error
121 if nomeasure_warn and circ.n_gates_of_type(OpType.Measure) < 1: 121 ↛ 122line 121 didn't jump to line 122 because the condition on line 121 was never true
122 warnings.warn( # noqa: B028
123 f"Circuit with index {i} in submitted does not contain a "
124 "measure operation."
125 )
126 return True
128 @abstractmethod
129 def rebase_pass(self) -> BasePass:
130 """
131 A single compilation pass that when run converts all gates in a Circuit to
132 an OpType supported by the Backend (ignoring architecture constraints).
134 :return: Compilation pass that converts gates to primitives supported by
135 Backend.
136 :rtype: BasePass
137 """
138 ...
140 @abstractmethod
141 def default_compilation_pass(self, optimisation_level: int = 2) -> BasePass:
142 """
143 A suggested compilation pass that will will, if possible, produce an equivalent
144 circuit suitable for running on this backend.
146 At a minimum it will ensure that compatible gates are used and that all two-
147 qubit interactions are compatible with the backend's qubit architecture. At
148 higher optimisation levels, further optimisations may be applied.
150 This is a an abstract method which is implemented in the backend itself, and so
151 is tailored to the backend's requirements.
153 :param optimisation_level: The level of optimisation to perform during
154 compilation.
156 - Level 0 does the minimum required to solves the device constraints,
157 without any optimisation.
158 - Level 1 additionally performs some light optimisations.
159 - Level 2 (the default) adds more computationally intensive optimisations
160 that should give the best results from execution.
162 :type optimisation_level: int, optional
163 :return: Compilation pass guaranteeing required predicates.
164 :rtype: BasePass
165 """
166 ...
168 def get_compiled_circuit(
169 self, circuit: Circuit, optimisation_level: int = 2
170 ) -> Circuit:
171 """
172 Return a single circuit compiled with :py:meth:`default_compilation_pass`. See
173 :py:meth:`Backend.get_compiled_circuits`.
174 """
175 return_circuit = circuit.copy()
176 self.default_compilation_pass(optimisation_level).apply(return_circuit)
177 return return_circuit
179 def get_compiled_circuits(
180 self, circuits: Sequence[Circuit], optimisation_level: int = 2
181 ) -> list[Circuit]:
182 """Compile a sequence of circuits with :py:meth:`default_compilation_pass`
183 and return the list of compiled circuits (does not act in place).
185 As well as applying a degree of optimisation (controlled by the
186 `optimisation_level` parameter), this method tries to ensure that the circuits
187 can be run on the backend (i.e. successfully passed to
188 :py:meth:`process_circuits`), for example by rebasing to the supported gate set,
189 or routing to match the connectivity of the device. However, this is not always
190 possible, for example if the circuit contains classical operations that are not
191 supported by the backend. You may use :py:meth:`valid_circuit` to check whether
192 the circuit meets the backend's requirements after compilation. This validity
193 check is included in :py:meth:`process_circuits` by default, before any circuits
194 are submitted to the backend.
196 If the validity check fails, you can obtain more information about the failure
197 by iterating through the predicates in the `required_predicates` property of the
198 backend, and running the :py:meth:`verify` method on each in turn with your
199 circuit.
201 :param circuits: The circuits to compile.
202 :type circuit: Sequence[Circuit]
203 :param optimisation_level: The level of optimisation to perform during
204 compilation. See :py:meth:`default_compilation_pass` for a description of
205 the different levels (0, 1 or 2). Defaults to 2.
206 :type optimisation_level: int, optional
207 :return: Compiled circuits.
208 :rtype: List[Circuit]
209 """
210 return [self.get_compiled_circuit(c, optimisation_level) for c in circuits]
212 @property
213 @abstractmethod
214 def _result_id_type(self) -> _ResultIdTuple:
215 """Identifier type signature for ResultHandle for this backend.
217 :return: Type signature (tuple of hashable types)
218 :rtype: _ResultIdTuple
219 """
220 ...
222 def _check_handle_type(self, reshandle: ResultHandle) -> None:
223 """Check a result handle is valid for this backend, raises TypeError if not.
225 :param reshandle: Handle to check
226 :type reshandle: ResultHandle
227 :raises TypeError: Types of handle identifiers don't match those of backend.
228 """
229 if (len(reshandle) != len(self._result_id_type)) or not all(
230 isinstance(idval, ty)
231 for idval, ty in zip(reshandle, self._result_id_type, strict=False)
232 ):
233 raise ResultHandleTypeError(
234 f"{reshandle!r} does not match expected "
235 f"identifier types {self._result_id_type}"
236 )
238 def process_circuit(
239 self,
240 circuit: Circuit,
241 n_shots: int | None = None,
242 valid_check: bool = True,
243 **kwargs: KwargTypes,
244 ) -> ResultHandle:
245 """
246 Submit a single circuit to the backend for running. See
247 :py:meth:`Backend.process_circuits`.
248 """
250 return self.process_circuits(
251 [circuit], n_shots=n_shots, valid_check=valid_check, **kwargs
252 )[0]
254 @abstractmethod
255 def process_circuits(
256 self,
257 circuits: Sequence[Circuit],
258 n_shots: int | Sequence[int] | None = None,
259 valid_check: bool = True,
260 **kwargs: KwargTypes,
261 ) -> list[ResultHandle]:
262 """
263 Submit circuits to the backend for running. The results will be stored
264 in the backend's result cache to be retrieved by the corresponding
265 get_<data> method.
267 If the `postprocess` keyword argument is set to True, and the backend supports
268 the feature (see :py:meth:`supports_contextual_optimisation`), then contextual
269 optimisatioons are applied before running the circuit and retrieved results will
270 have any necessary classical postprocessing applied. This is not enabled by
271 default.
273 Use keyword arguments to specify parameters to be used in submitting circuits
274 See specific Backend derived class for available parameters, from the following
275 list:
277 * `seed`: RNG seed for simulators
278 * `postprocess`: if True, apply contextual optimisations
280 Note: If a backend is reused many times, the in-memory results cache grows
281 indefinitely. Therefore, when processing many circuits on a statevector or
282 unitary backend (whose results may occupy significant amounts of memory), it is
283 advisable to run :py:meth:`Backend.empty_cache` after each result is retrieved.
285 :param circuits: Circuits to process on the backend.
286 :type circuits: Sequence[Circuit]
287 :param n_shots: Number of shots to run per circuit. Optionally, this can be
288 a list of shots specifying the number of shots for each circuit separately.
289 None is to be used for state/unitary simulators. Defaults to None.
290 :type n_shots: Optional[Union[int, Iterable[int]], optional
291 :param valid_check: Explicitly check that all circuits satisfy all required
292 predicates to run on the backend. Defaults to True
293 :type valid_check: bool, optional
294 :return: Handles to results for each input circuit, as an interable in
295 the same order as the circuits.
296 :rtype: List[ResultHandle]
297 """
298 ...
300 @abstractmethod
301 def circuit_status(self, handle: ResultHandle) -> CircuitStatus:
302 """
303 Return a CircuitStatus reporting the status of the circuit execution
304 corresponding to the ResultHandle
305 """
306 ...
308 def empty_cache(self) -> None:
309 """Manually empty the result cache on the backend."""
310 self._cache = {}
312 def pop_result(self, handle: ResultHandle) -> ResultCache | None:
313 """Remove cache entry corresponding to handle from the cache and return.
315 :param handle: ResultHandle object
316 :type handle: ResultHandle
317 :return: Cache entry corresponding to handle, if it was present
318 :rtype: Optional[ResultCache]
319 """
320 return self._cache.pop(handle, None)
322 def get_result(self, handle: ResultHandle, **kwargs: KwargTypes) -> BackendResult:
323 """Return a BackendResult corresponding to the handle.
325 Use keyword arguments to specify parameters to be used in retrieving results.
326 See specific Backend derived class for available parameters, from the following
327 list:
329 * `timeout`: maximum time to wait for remote job to finish
330 * `wait`: polling interval between remote calls to check job status
332 :param handle: handle to results
333 :type handle: ResultHandle
334 :return: Results corresponding to handle.
335 :rtype: BackendResult
336 """
337 self._check_handle_type(handle)
338 if handle in self._cache and "result" in self._cache[handle]:
339 return cast("BackendResult", self._cache[handle]["result"])
340 raise CircuitNotRunError(handle)
342 def get_results(
343 self, handles: Iterable[ResultHandle], **kwargs: KwargTypes
344 ) -> list[BackendResult]:
345 """Return results corresponding to handles.
347 :param handles: Iterable of handles
348 :return: List of results
350 Keyword arguments are as for `get_result`, and apply to all jobs.
351 """
352 try:
353 return [self.get_result(handle, **kwargs) for handle in handles]
354 except ResultHandleTypeError as e:
355 try:
356 self._check_handle_type(cast("ResultHandle", handles))
357 except ResultHandleTypeError:
358 raise e # noqa: B904
360 raise ResultHandleTypeError(
361 "Possible use of single ResultHandle"
362 " where sequence of ResultHandles was expected."
363 ) from e
365 def run_circuit(
366 self,
367 circuit: Circuit,
368 n_shots: int | None = None,
369 valid_check: bool = True,
370 **kwargs: KwargTypes,
371 ) -> BackendResult:
372 """
373 Submits a circuit to the backend and returns results
375 :param circuit: Circuit to be executed
376 :param n_shots: Passed on to :py:meth:`Backend.process_circuit`
377 :param valid_check: Passed on to :py:meth:`Backend.process_circuit`
378 :return: Result
380 This is a convenience method equivalent to calling
381 :py:meth:`Backend.process_circuit` followed by :py:meth:`Backend.get_result`.
382 Any additional keyword arguments are passed on to
383 :py:meth:`Backend.process_circuit` and :py:meth:`Backend.get_result`.
384 """
385 return self.run_circuits(
386 [circuit], n_shots=n_shots, valid_check=valid_check, **kwargs
387 )[0]
389 def run_circuits(
390 self,
391 circuits: Sequence[Circuit],
392 n_shots: int | Sequence[int] | None = None,
393 valid_check: bool = True,
394 **kwargs: KwargTypes,
395 ) -> list[BackendResult]:
396 """
397 Submits circuits to the backend and returns results
399 :param circuits: Sequence of Circuits to be executed
400 :param n_shots: Passed on to :py:meth:`Backend.process_circuits`
401 :param valid_check: Passed on to :py:meth:`Backend.process_circuits`
402 :return: List of results
404 This is a convenience method equivalent to calling
405 :py:meth:`Backend.process_circuits` followed by :py:meth:`Backend.get_results`.
406 Any additional keyword arguments are passed on to
407 :py:meth:`Backend.process_circuits` and :py:meth:`Backend.get_results`.
408 """
409 handles = self.process_circuits(circuits, n_shots, valid_check, **kwargs)
410 results = self.get_results(handles, **kwargs)
411 for h in handles:
412 self.pop_result(h)
413 return results
415 def cancel(self, handle: ResultHandle) -> None:
416 """
417 Cancel a job.
419 :param handle: handle to job
420 :type handle: ResultHandle
421 :raises NotImplementedError: If backend does not support job cancellation
422 """
423 raise NotImplementedError("Backend does not support job cancellation.")
425 @property
426 def backend_info(self) -> BackendInfo | None:
427 """Retrieve all Backend properties in a BackendInfo object, including
428 device architecture, supported gate set, gate errors and other hardware-specific
429 information.
431 :return: The BackendInfo describing this backend if it exists.
432 :rtype: Optional[BackendInfo]
433 """
434 raise NotImplementedError("Backend does not provide any device properties.")
436 @classmethod
437 def available_devices(cls, **kwargs: Any) -> list[BackendInfo]:
438 """Retrieve all available devices as a list of BackendInfo objects, including
439 device name, architecture, supported gate set, gate errors,
440 and other hardware-specific information.
442 :return: A list of BackendInfo objects describing available devices.
443 :rtype: List[BackendInfo]
444 """
445 raise NotImplementedError(
446 "Backend does not provide information about available devices."
447 )
449 @property
450 def persistent_handles(self) -> bool:
451 """
452 Whether the backend produces `ResultHandle` objects that can be reused with
453 other instances of the backend class.
454 """
455 return self._persistent_handles
457 @property
458 def supports_shots(self) -> bool:
459 """
460 Does this backend support shot result retrieval via
461 :py:meth:`backendresult.BackendResult.get_shots`.
462 """
463 return self._supports_shots
465 @property
466 def supports_counts(self) -> bool:
467 """
468 Does this backend support counts result retrieval via
469 :py:meth:`backendresult.BackendResult.get_counts`.
470 """
471 return self._supports_counts
473 @property
474 def supports_state(self) -> bool:
475 """
476 Does this backend support statevector retrieval via
477 :py:meth:`backendresult.BackendResult.get_state`.
478 """
479 return self._supports_state
481 @property
482 def supports_unitary(self) -> bool:
483 """
484 Does this backend support unitary retrieval via
485 :py:meth:`backendresult.BackendResult.get_unitary`.
486 """
487 return self._supports_unitary
489 @property
490 def supports_density_matrix(self) -> bool:
491 """Does this backend support density matrix retrieval via
492 `get_density_matrix`."""
493 return self._supports_density_matrix
495 @property
496 def supports_expectation(self) -> bool:
497 """Does this backend support expectation value calculation for operators."""
498 return self._supports_expectation
500 @property
501 def expectation_allows_nonhermitian(self) -> bool:
502 """If expectations are supported, is the operator allowed to be non-Hermitan?"""
503 return self._expectation_allows_nonhermitian
505 @property
506 def supports_contextual_optimisation(self) -> bool:
507 """Does this backend support contextual optimisation?
509 See :py:meth:`process_circuits`."""
510 return self._supports_contextual_optimisation
512 def _get_extension_module(self) -> ModuleType | None:
513 """Return the extension module of the backend if it belongs to a
514 pytket-extension package.
516 :return: The extension module of the backend if it belongs to a pytket-extension
517 package.
518 :rtype: Optional[ModuleType]
519 """
520 mod_parts = self.__class__.__module__.split(".")[:3]
521 if not (mod_parts[0] == "pytket" and mod_parts[1] == "extensions"):
522 return None
523 return import_module(".".join(mod_parts))
525 @property
526 def __extension_name__(self) -> str | None:
527 """Retrieve the extension name of the backend if it belongs to a
528 pytket-extension package.
530 :return: The extension name of the backend if it belongs to a pytket-extension
531 package.
532 :rtype: Optional[str]
533 """
534 try:
535 return self._get_extension_module().__extension_name__ # type: ignore
536 except AttributeError:
537 return None
539 @property
540 def __extension_version__(self) -> str | None:
541 """Retrieve the extension version of the backend if it belongs to a
542 pytket-extension package.
544 :return: The extension version of the backend if it belongs to a
545 pytket-extension package.
546 :rtype: Optional[str]
547 """
548 try:
549 return self._get_extension_module().__extension_version__ # type: ignore
550 except AttributeError:
551 return None
553 @overload
554 @staticmethod
555 def _get_n_shots_as_list( 555 ↛ exitline 555 didn't return from function '_get_n_shots_as_list' because
556 n_shots: None | int | Sequence[int | None],
557 n_circuits: int,
558 optional: Literal[False],
559 ) -> list[int]: ...
561 @overload
562 @staticmethod
563 def _get_n_shots_as_list( 563 ↛ exitline 563 didn't return from function '_get_n_shots_as_list' because
564 n_shots: None | int | Sequence[int | None],
565 n_circuits: int,
566 optional: Literal[True],
567 set_zero: Literal[True],
568 ) -> list[int]: ...
570 @overload
571 @staticmethod
572 def _get_n_shots_as_list( 572 ↛ exitline 572 didn't return from function '_get_n_shots_as_list' because
573 n_shots: None | int | Sequence[int | None],
574 n_circuits: int,
575 optional: bool = True,
576 set_zero: bool = False,
577 ) -> list[int | None] | list[int]: ...
579 @staticmethod
580 def _get_n_shots_as_list(
581 n_shots: None | int | Sequence[int | None],
582 n_circuits: int,
583 optional: bool = True,
584 set_zero: bool = False,
585 ) -> list[int | None] | list[int]:
586 """
587 Convert any admissible n_shots value into List[Optional[int]] format.
589 This validates the n_shots argument for process_circuits. If a single
590 value is passed, this value is broadcast to the number of circuits.
591 Additional boolean flags control how the argument is validated.
592 Raises an exception if n_shots is in an invalid format.
594 :param n_shots: The argument to be validated.
595 :type n_shots: Union[None, int, Sequence[Optional[int]]]
596 :param n_circuits: Length of the converted argument returned.
597 :type n_circuits: int
598 :param optional: Whether n_shots can be None (default: True).
599 :type optional: bool
600 :param set_zero: Whether None values should be set to 0 (default: False).
601 :type set_zero: bool
602 :return: a list of length `n_circuits`, the converted argument
603 """
605 n_shots_list: list[int | None] = []
607 def validate_n_shots(n: int | None) -> bool:
608 return optional or (n is not None and n > 0)
610 if set_zero and not optional: 610 ↛ 611line 610 didn't jump to line 611 because the condition on line 610 was never true
611 raise ValueError("set_zero cannot be true when optional is false")
613 if hasattr(n_shots, "__iter__"):
614 assert not isinstance(n_shots, int)
615 assert n_shots is not None
617 if not all(map(validate_n_shots, n_shots)):
618 raise ValueError(
619 "n_shots values are required for all circuits for this backend"
620 )
621 n_shots_list = list(n_shots)
622 else:
623 assert n_shots is None or isinstance(n_shots, int)
625 if not validate_n_shots(n_shots):
626 raise ValueError("Parameter n_shots is required for this backend")
627 # convert n_shots to a list
628 n_shots_list = [n_shots] * n_circuits
630 if len(n_shots_list) != n_circuits:
631 raise ValueError("The length of n_shots and circuits must match")
633 if set_zero:
634 # replace None with 0
635 n_shots_list = [n or 0 for n in n_shots_list]
637 return n_shots_list
639 def get_pauli_expectation_value(
640 self, state_circuit: Circuit, pauli: QubitPauliString
641 ) -> complex:
642 """
643 Calculates the expectation value of the given circuit using
644 functionality built into the backend.
646 Raises an exception if the backend does not provide custom expectation
647 value features.
648 """
649 raise NotImplementedError
651 def get_operator_expectation_value(
652 self, state_circuit: Circuit, operator: QubitPauliOperator
653 ) -> complex:
654 """
655 Calculates the expectation value of the given circuit with respect to
656 the operator using functionality built into the backend.
658 Raises an exception if the backend does not provide custom expectation
659 value features.
660 """
661 raise NotImplementedError