Coverage for /home/runner/work/tket/tket/pytket/pytket/backends/status.py: 98%
63 statements
« prev ^ index » next coverage.py v7.6.12, created at 2025-03-14 11:30 +0000
« 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.
15"""Status classes for circuits submitted to backends."""
16from collections.abc import Callable
17from datetime import datetime
18from enum import Enum
19from typing import Any, NamedTuple
22class StatusEnum(Enum):
23 """Enumeration for the possible status of a circuit submitted to a backend."""
25 COMPLETED = "Circuit has completed. Results are ready."
26 QUEUED = "Circuit is queued."
27 SUBMITTED = "Circuit has been submitted."
28 RUNNING = "Circuit is running."
29 RETRYING = "Circuit is being retried."
30 CANCELLING = "Cancellation has been requested."
31 CANCELLED = "Circuit has been cancelled."
32 ERROR = "Circuit has errored. Check CircuitStatus.message for error message."
35class CircuitStatus(NamedTuple):
36 """The status of a circuit along with an optional description.
38 Optionally can also include extra fields such as:
39 * Detailed error information.
40 * Timestamps for changes in status.
41 * Queue position.
42 """
44 status: StatusEnum
45 message: str = ""
46 error_detail: str | None = None
48 # Timestamp for when a status was last entered.
49 completed_time: datetime | None = None
50 queued_time: datetime | None = None
51 submitted_time: datetime | None = None
52 running_time: datetime | None = None
53 cancelled_time: datetime | None = None
54 error_time: datetime | None = None
56 queue_position: int | None = None
58 def to_dict(self) -> dict[str, Any]:
59 """Return JSON serializable dictionary representation."""
60 circuit_status_dict: dict[str, Any] = {
61 "status": self.status.name,
62 "message": self.message,
63 }
64 if self.error_detail is not None:
65 circuit_status_dict["error_detail"] = self.error_detail
67 if self.completed_time is not None:
68 circuit_status_dict["completed_time"] = self.completed_time.isoformat()
69 if self.queued_time is not None:
70 circuit_status_dict["queued_time"] = self.queued_time.isoformat()
71 if self.submitted_time is not None:
72 circuit_status_dict["submitted_time"] = self.submitted_time.isoformat()
73 if self.running_time is not None:
74 circuit_status_dict["running_time"] = self.running_time.isoformat()
75 if self.cancelled_time is not None:
76 circuit_status_dict["cancelled_time"] = self.cancelled_time.isoformat()
77 if self.error_time is not None:
78 circuit_status_dict["error_time"] = self.error_time.isoformat()
80 if self.queue_position is not None:
81 circuit_status_dict["queue_position"] = self.queue_position
83 return circuit_status_dict
85 @classmethod
86 def from_dict(cls, dic: dict[str, Any]) -> "CircuitStatus":
87 """Construct from JSON serializable dictionary."""
88 invalid = ValueError(f"Dictionary invalid format for CircuitStatus: {dic}")
89 if "message" not in dic or "status" not in dic: 89 ↛ 90line 89 didn't jump to line 90 because the condition on line 89 was never true
90 raise invalid
92 try:
93 status = next(s for s in StatusEnum if dic["status"] == s.name)
94 except StopIteration as e:
95 raise invalid from e
97 error_detail = dic.get("error_detail")
99 read_optional_datetime: Callable[[str], datetime | None] = lambda key: (
100 datetime.fromisoformat(x) if (x := dic.get(key)) is not None else None
101 )
102 completed_time = read_optional_datetime("completed_time")
103 queued_time = read_optional_datetime("queued_time")
104 submitted_time = read_optional_datetime("submitted_time")
105 running_time = read_optional_datetime("running_time")
106 cancelled_time = read_optional_datetime("cancelled_time")
107 error_time = read_optional_datetime("error_time")
109 queue_position = dic.get("queue_position")
111 return cls(
112 status,
113 dic["message"],
114 error_detail,
115 completed_time,
116 queued_time,
117 submitted_time,
118 running_time,
119 cancelled_time,
120 error_time,
121 queue_position,
122 )
125WAITING_STATUS = {StatusEnum.QUEUED, StatusEnum.SUBMITTED, StatusEnum.RUNNING}