Coverage for /home/runner/work/tket/tket/pytket/pytket/backends/status.py: 95%

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"""Status classes for circuits submitted to backends.""" 

16 

17from datetime import datetime 

18from enum import Enum 

19from typing import TYPE_CHECKING, Any, NamedTuple 

20 

21if TYPE_CHECKING: 21 ↛ 22line 21 didn't jump to line 22 because the condition on line 21 was never true

22 from collections.abc import Callable 

23 

24 

25class StatusEnum(Enum): 

26 """Enumeration for the possible status of a circuit submitted to a backend.""" 

27 

28 COMPLETED = "Circuit has completed. Results are ready." 

29 QUEUED = "Circuit is queued." 

30 SUBMITTED = "Circuit has been submitted." 

31 RUNNING = "Circuit is running." 

32 RETRYING = "Circuit is being retried." 

33 CANCELLING = "Cancellation has been requested." 

34 CANCELLED = "Circuit has been cancelled." 

35 ERROR = "Circuit has errored. Check CircuitStatus.message for error message." 

36 

37 

38class CircuitStatus(NamedTuple): 

39 """The status of a circuit along with an optional description. 

40 

41 Optionally can also include extra fields such as: 

42 * Detailed error information. 

43 * Timestamps for changes in status. 

44 * Queue position. 

45 """ 

46 

47 status: StatusEnum 

48 message: str = "" 

49 error_detail: str | None = None 

50 

51 # Timestamp for when a status was last entered. 

52 completed_time: datetime | None = None 

53 queued_time: datetime | None = None 

54 submitted_time: datetime | None = None 

55 running_time: datetime | None = None 

56 cancelled_time: datetime | None = None 

57 error_time: datetime | None = None 

58 

59 queue_position: int | None = None 

60 

61 def to_dict(self) -> dict[str, Any]: 

62 """Return JSON serializable dictionary representation.""" 

63 circuit_status_dict: dict[str, Any] = { 

64 "status": self.status.name, 

65 "message": self.message, 

66 } 

67 if self.error_detail is not None: 

68 circuit_status_dict["error_detail"] = self.error_detail 

69 

70 if self.completed_time is not None: 

71 circuit_status_dict["completed_time"] = self.completed_time.isoformat() 

72 if self.queued_time is not None: 

73 circuit_status_dict["queued_time"] = self.queued_time.isoformat() 

74 if self.submitted_time is not None: 

75 circuit_status_dict["submitted_time"] = self.submitted_time.isoformat() 

76 if self.running_time is not None: 

77 circuit_status_dict["running_time"] = self.running_time.isoformat() 

78 if self.cancelled_time is not None: 

79 circuit_status_dict["cancelled_time"] = self.cancelled_time.isoformat() 

80 if self.error_time is not None: 

81 circuit_status_dict["error_time"] = self.error_time.isoformat() 

82 

83 if self.queue_position is not None: 

84 circuit_status_dict["queue_position"] = self.queue_position 

85 

86 return circuit_status_dict 

87 

88 @classmethod 

89 def from_dict(cls, dic: dict[str, Any]) -> "CircuitStatus": 

90 """Construct from JSON serializable dictionary.""" 

91 invalid = ValueError(f"Dictionary invalid format for CircuitStatus: {dic}") 

92 if "message" not in dic or "status" not in dic: 92 ↛ 93line 92 didn't jump to line 93 because the condition on line 92 was never true

93 raise invalid 

94 

95 try: 

96 status = next(s for s in StatusEnum if dic["status"] == s.name) 

97 except StopIteration as e: 

98 raise invalid from e 

99 

100 error_detail = dic.get("error_detail") 

101 

102 read_optional_datetime: Callable[[str], datetime | None] = lambda key: ( 

103 datetime.fromisoformat(x) if (x := dic.get(key)) is not None else None 

104 ) 

105 completed_time = read_optional_datetime("completed_time") 

106 queued_time = read_optional_datetime("queued_time") 

107 submitted_time = read_optional_datetime("submitted_time") 

108 running_time = read_optional_datetime("running_time") 

109 cancelled_time = read_optional_datetime("cancelled_time") 

110 error_time = read_optional_datetime("error_time") 

111 

112 queue_position = dic.get("queue_position") 

113 

114 return cls( 

115 status, 

116 dic["message"], 

117 error_detail, 

118 completed_time, 

119 queued_time, 

120 submitted_time, 

121 running_time, 

122 cancelled_time, 

123 error_time, 

124 queue_position, 

125 ) 

126 

127 

128WAITING_STATUS = {StatusEnum.QUEUED, StatusEnum.SUBMITTED, StatusEnum.RUNNING}