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

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

16from collections.abc import Callable 

17from datetime import datetime 

18from enum import Enum 

19from typing import Any, NamedTuple 

20 

21 

22class StatusEnum(Enum): 

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

24 

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

33 

34 

35class CircuitStatus(NamedTuple): 

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

37 

38 Optionally can also include extra fields such as: 

39 * Detailed error information. 

40 * Timestamps for changes in status. 

41 * Queue position. 

42 """ 

43 

44 status: StatusEnum 

45 message: str = "" 

46 error_detail: str | None = None 

47 

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 

55 

56 queue_position: int | None = None 

57 

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 

66 

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

79 

80 if self.queue_position is not None: 

81 circuit_status_dict["queue_position"] = self.queue_position 

82 

83 return circuit_status_dict 

84 

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 

91 

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 

96 

97 error_detail = dic.get("error_detail") 

98 

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

108 

109 queue_position = dic.get("queue_position") 

110 

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 ) 

123 

124 

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