Coverage for scanpath_studio/export_status.py: 94%

1"""Shared, honest progress vocabulary for static, animation, and bulk exports.""" 

2 

3from __future__ import annotations 

4 

5import hashlib 

6import math 

7import time 

8from collections.abc import Callable 

9from dataclasses import dataclass 

10from enum import Enum 

11 

12EXPORTER_VERSION = "exp6-v1" 

13 

14 

15class ExportStage(str, Enum): 

16 PREPARING = "preparing" 

17 STARTING_RENDERER = "starting_renderer" 

18 RASTERIZING = "rasterizing" 

19 ENCODING_WRITING = "encoding_writing" 

20 FINALIZING = "finalizing" 

21 READY = "ready" 

22 ERROR = "error" 

23 

24 

25@dataclass(frozen=True) 

26class ExportStatus: 

27 """One observable job state; counts appear only when real units exist.""" 

28 

29 stage: ExportStage 

30 message: str 

31 completed: int | None = None 

32 total: int | None = None 

33 elapsed_s: float = 0.0 

34 error: str | None = None 

35 

36 @property 

37 def fraction(self) -> float | None: 

38 if self.completed is None or self.total is None or self.total <= 0: 

39 return None 

40 return min(max(self.completed / self.total, 0.0), 1.0) 

41 

42 @property 

43 def rate_per_s(self) -> float | None: 

44 """Units finished per second so far, or ``None`` before there are any.""" 

45 if not self.completed or not self.elapsed_s: 

46 return None 

47 return self.completed / self.elapsed_s 

48 

49 @property 

50 def remaining_s(self) -> float | None: 

51 """Seconds left at the rate observed so far (PERF-6). 

52 

53 ``None`` until the first unit finishes: with nothing done there is no 

54 rate, and an estimate invented from one would be a guess wearing a 

55 number. Deliberately the naive rate over the *whole* run rather than a 

56 recent window — a bulk export's per-trial cost is near-constant, and an 

57 estimate that lurches with each slow trial reads as broken. 

58 """ 

59 rate = self.rate_per_s 

60 if rate is None or self.total is None: 

61 return None 

62 return max(0.0, (self.total - self.completed) / rate) 

63 

64 

65StatusCallback = Callable[[ExportStatus], None] 

66 

67 

68def format_duration(seconds: float | None) -> str: 

69 """A duration as a person would say it: ``2h 15m``, ``1m 15s``, ``9s``.""" 

70 if seconds is None: 

71 return "" 

72 if seconds < 1: 

73 return "under a second" 

74 seconds = round(seconds) # round() on a float already returns an int 

75 hours, rest = divmod(seconds, 3600) 

76 minutes, secs = divmod(rest, 60) 

77 if hours: 

78 return f"{hours}h {minutes}m" 

79 if minutes: 

80 return f"{minutes}m {secs}s" 

81 return f"{secs}s" 

82 

83 

84def progress_caption(status: ExportStatus) -> str: 

85 """One line under the bar: how far, how fast, how much longer. 

86 

87 A bulk export over a real corpus runs for hours, and "1,203 / 20,000" only 

88 answers the first of those. The rate is what makes the estimate legible — 

89 it lets someone sanity-check the number rather than take it on faith. 

90 """ 

91 if status.completed is None or status.total is None: 

92 return status.message 

93 # The total counts export units — a multi-screen trial is several. 

94 parts = [f"{status.completed:,}/{status.total:,} done"] 

95 rate = status.rate_per_s 

96 if rate is not None: 

97 parts.append(f"{rate:.3g}/s" if rate < 10 else f"{rate:.0f}/s") 

98 remaining = status.remaining_s 

99 if remaining is not None: 

100 parts.append( 

101 "finishing…" if remaining < 1 else f"{format_duration(remaining)} left" 

102 ) 

103 parts.append(status.message) 

104 return " · ".join(parts) 

105 

106 

107def emit_status( 

108 callback: StatusCallback | None, 

109 stage: ExportStage, 

110 message: str, 

111 *, 

112 started_at: float | None = None, 

113 completed: int | None = None, 

114 total: int | None = None, 

115 error: str | None = None, 

116) -> ExportStatus: 

117 """Validate and emit one status while remaining cheap when no UI listens.""" 

118 if (completed is None) != (total is None): 

119 raise ValueError("completed and total must be supplied together") 

120 if completed is not None and (completed < 0 or total is None or total < 0): 

121 raise ValueError("progress counts must be non-negative") 

122 if completed is not None and total is not None and completed > total: 

123 raise ValueError("completed progress cannot exceed total") 

124 status = ExportStatus( 

125 stage=stage, 

126 message=message, 

127 completed=completed, 

128 total=total, 

129 elapsed_s=max(0.0, time.perf_counter() - started_at) if started_at else 0.0, 

130 error=error, 

131 ) 

132 if callback is not None: 

133 callback(status) 

134 return status 

135 

136 

137def export_signature( 

138 content: str, 

139 *, 

140 fmt: str, 

141 width: int, 

142 height: int, 

143 scale: float, 

144 exporter_version: str = EXPORTER_VERSION, 

145) -> str: 

146 """Hash every output-affecting input for safe export-byte reuse. 

147 

148 ``content`` identifies what is drawn. It used to be the figure's JSON, which 

149 costs 12 s to write for a 2,000-frame replay on every rerun; the animation 

150 export passes its replay's inputs instead (PERF-16, `tabs._ReplayView`). 

151 """ 

152 if int(width) <= 0 or int(height) <= 0: 

153 raise ValueError("export width and height must be positive") 

154 if not math.isfinite(float(scale)) or float(scale) <= 0: 

155 raise ValueError("export scale must be a positive finite number") 

156 digest = hashlib.sha256() 

157 for value in ( 

158 exporter_version, 

159 str(fmt).lower(), 

160 str(int(width)), 

161 str(int(height)), 

162 f"{float(scale):.12g}", 

163 ): 

164 digest.update(value.encode("utf-8")) 

165 digest.update(b"\0") 

166 digest.update(content.encode("utf-8")) 

167 return digest.hexdigest()