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 11:30 +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. 

14 

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 

22 

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 

30 

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 

39 

40ResultCache = dict[str, Any] 

41 

42 

43class ResultHandleTypeError(Exception): 

44 """Wrong result handle type.""" 

45 

46 

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 """ 

53 

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 

63 

64 def __init__(self) -> None: 

65 self._cache: dict[ResultHandle, ResultCache] = {} 

66 

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) 

74 

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. 

81 

82 :return: Required predicates. 

83 :rtype: List[Predicate] 

84 """ 

85 ... 

86 

87 def valid_circuit(self, circuit: Circuit) -> bool: 

88 """ 

89 Checks that the circuit satisfies all of required_predicates. 

90 

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) 

97 

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 

123 

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). 

129 

130 :return: Compilation pass that converts gates to primitives supported by 

131 Backend. 

132 :rtype: BasePass 

133 """ 

134 ... 

135 

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. 

141 

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. 

145 

146 This is a an abstract method which is implemented in the backend itself, and so 

147 is tailored to the backend's requirements. 

148 

149 :param optimisation_level: The level of optimisation to perform during 

150 compilation. 

151 

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. 

157 

158 :type optimisation_level: int, optional 

159 :return: Compilation pass guaranteeing required predicates. 

160 :rtype: BasePass 

161 """ 

162 ... 

163 

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 

174 

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). 

180 

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. 

191 

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. 

196 

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] 

207 

208 @property 

209 @abstractmethod 

210 def _result_id_type(self) -> _ResultIdTuple: 

211 """Identifier type signature for ResultHandle for this backend. 

212 

213 :return: Type signature (tuple of hashable types) 

214 :rtype: _ResultIdTuple 

215 """ 

216 ... 

217 

218 def _check_handle_type(self, reshandle: ResultHandle) -> None: 

219 """Check a result handle is valid for this backend, raises TypeError if not. 

220 

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 ) 

232 

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 """ 

244 

245 return self.process_circuits( 

246 [circuit], n_shots=n_shots, valid_check=valid_check, **kwargs 

247 )[0] 

248 

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. 

261 

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. 

267 

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: 

271 

272 * `seed`: RNG seed for simulators 

273 * `postprocess`: if True, apply contextual optimisations 

274 

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. 

279 

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 ... 

294 

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 ... 

302 

303 def empty_cache(self) -> None: 

304 """Manually empty the result cache on the backend.""" 

305 self._cache = {} 

306 

307 def pop_result(self, handle: ResultHandle) -> ResultCache | None: 

308 """Remove cache entry corresponding to handle from the cache and return. 

309 

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) 

316 

317 def get_result(self, handle: ResultHandle, **kwargs: KwargTypes) -> BackendResult: 

318 """Return a BackendResult corresponding to the handle. 

319 

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: 

323 

324 * `timeout`: maximum time to wait for remote job to finish 

325 * `wait`: polling interval between remote calls to check job status 

326 

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) 

336 

337 def get_results( 

338 self, handles: Iterable[ResultHandle], **kwargs: KwargTypes 

339 ) -> list[BackendResult]: 

340 """Return results corresponding to handles. 

341 

342 :param handles: Iterable of handles 

343 :return: List of results 

344 

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 

354 

355 raise ResultHandleTypeError( 

356 "Possible use of single ResultHandle" 

357 " where sequence of ResultHandles was expected." 

358 ) from e 

359 

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 

369 

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 

374 

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] 

383 

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 

393 

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 

398 

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 

409 

410 def cancel(self, handle: ResultHandle) -> None: 

411 """ 

412 Cancel a job. 

413 

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.") 

419 

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. 

425 

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.") 

430 

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. 

436 

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 ) 

443 

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 

451 

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 

459 

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 

467 

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 

475 

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 

483 

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 

489 

490 @property 

491 def supports_expectation(self) -> bool: 

492 """Does this backend support expectation value calculation for operators.""" 

493 return self._supports_expectation 

494 

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 

499 

500 @property 

501 def supports_contextual_optimisation(self) -> bool: 

502 """Does this backend support contextual optimisation? 

503 

504 See :py:meth:`process_circuits`.""" 

505 return self._supports_contextual_optimisation 

506 

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. 

510 

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)) 

519 

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. 

524 

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 

533 

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. 

538 

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 

547 

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]: ... 

555 

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]: ... 

564 

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]: ... 

573 

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. 

583 

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. 

588 

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 """ 

599 

600 n_shots_list: list[int | None] = [] 

601 

602 def validate_n_shots(n: int | None) -> bool: 

603 return optional or (n is not None and n > 0) 

604 

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") 

607 

608 if hasattr(n_shots, "__iter__"): 

609 assert not isinstance(n_shots, int) 

610 assert n_shots is not None 

611 

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) 

619 

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 

624 

625 if len(n_shots_list) != n_circuits: 

626 raise ValueError("The length of n_shots and circuits must match") 

627 

628 if set_zero: 

629 # replace None with 0 

630 n_shots_list = list(map(lambda n: n or 0, n_shots_list)) 

631 

632 return n_shots_list 

633 

634 def get_pauli_expectation_value( 

635 self, state_circuit: Circuit, pauli: QubitPauliString 

636 ) -> complex: 

637 raise NotImplementedError 

638 

639 def get_operator_expectation_value( 

640 self, state_circuit: Circuit, operator: QubitPauliOperator 

641 ) -> complex: 

642 raise NotImplementedError