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