Coverage for src/signalk_cli/history/_time.py: 98%

1"""Time ranges for History API requests.""" 

2 

3import logging 

4import re 

5from dataclasses import dataclass 

6from datetime import UTC, datetime, timedelta 

7 

8logger = logging.getLogger("signalk_cli") 

9 

10_TIMESTAMP_FMT = "%Y-%m-%dT%H:%M:%SZ" 

11 

12_DURATION_RE = re.compile( 

13 r"^P(?:(\d+)Y)?(?:(\d+)M)?(?:(\d+)W)?(?:(\d+)D)?" 

14 r"(?:T(?:(\d+)H)?(?:(\d+)M)?(?:(\d+(?:\.\d+)?)S)?)?$" 

15) 

16 

17 

18def _has_date_parts(duration: str) -> bool: 

19 m = _DURATION_RE.match(duration) 

20 return bool(m and any(m.group(i) for i in (1, 2, 3, 4))) 

21 

22 

23def _duration_to_timedelta(duration: str) -> timedelta: 

24 m = _DURATION_RE.match(duration) 

25 if not m: 25 ↛ 26line 25 didn't jump to line 26 because the condition on line 25 was never true

26 raise ValueError(f"Cannot parse duration: {duration!r}") 

27 years, months, weeks, days, hours, minutes = ( 

28 int(m.group(i) or 0) for i in range(1, 7) 

29 ) 

30 secs = float(m.group(7) or 0) 

31 return timedelta( 

32 days=years * 365 + months * 30 + weeks * 7 + days, 

33 hours=hours, 

34 minutes=minutes, 

35 seconds=secs, 

36 ) 

37 

38 

39def normalise_duration( 

40 duration: str | None, from_: str | None, to: str | None 

41) -> tuple[str | None, str | None, str | None]: 

42 """Convert date-component durations to explicit from/to timestamps. 

43 

44 SignalK only accepts PT-prefix (time-only) durations. Durations containing 

45 Y/M/W/D are expanded to from/to pairs: 

46 from + duration → to = from + duration 

47 to + duration → from = to - duration 

48 duration alone → from = now - duration, to = now 

49 

50 Returns (from_, to, duration_or_None). 

51 """ 

52 if not duration: 

53 return from_, to, duration 

54 try: 

55 int(duration) 

56 return from_, to, duration # integer seconds, pass through 

57 except ValueError: 

58 pass 

59 if not _has_date_parts(duration): 

60 return from_, to, duration # PT-only, pass through 

61 

62 delta = _duration_to_timedelta(duration) 

63 fmt = _TIMESTAMP_FMT 

64 now = datetime.now(UTC) 

65 

66 if from_ is not None and to is not None: 

67 return from_, to, None 

68 elif from_ is not None: 

69 from_dt = datetime.fromisoformat(from_) 

70 return from_, (from_dt + delta).strftime(fmt), None 

71 elif to is not None: 

72 to_dt = datetime.fromisoformat(to) 

73 return (to_dt - delta).strftime(fmt), to, None 

74 else: 

75 return (now - delta).strftime(fmt), now.strftime(fmt), None 

76 

77 

78def apply_time_default(time_params: dict) -> dict: 

79 """If neither 'from' nor 'duration' is set, default to the hour ending at 'to' (or now).""" 

80 if "from" in time_params or "duration" in time_params: 

81 return time_params 

82 if "to" in time_params: 

83 try: 

84 to_dt = datetime.fromisoformat(time_params["to"]) 

85 except ValueError: 

86 to_dt = datetime.now(UTC) 

87 else: 

88 to_dt = datetime.now(UTC) 

89 from_dt = to_dt - timedelta(hours=1) 

90 result = { 

91 **time_params, 

92 "from": from_dt.strftime(_TIMESTAMP_FMT), 

93 "to": to_dt.strftime(_TIMESTAMP_FMT), 

94 } 

95 logger.info( 

96 "No time range specified — defaulting to from=%s to=%s", 

97 result["from"], 

98 result["to"], 

99 ) 

100 return result 

101 

102 

103def _format_time(value: str | datetime | None) -> str | None: 

104 if value is None or isinstance(value, str): 

105 return value 

106 if value.tzinfo is None: 

107 value = value.replace(tzinfo=UTC) 

108 return value.astimezone(UTC).strftime(_TIMESTAMP_FMT) 

109 

110 

111def _format_duration(value: str | int | timedelta | None) -> str | None: 

112 if value is None or isinstance(value, str): 

113 return value 

114 if isinstance(value, timedelta): 

115 seconds = value.total_seconds() 

116 return str(int(seconds)) if seconds.is_integer() else f"PT{seconds}S" 

117 return str(value) 

118 

119 

120@dataclass(frozen=True) 

121class TimeRange: 

122 """The time window for a History API request. 

123 

124 Any combination of start, end and duration may be given, as 

125 the History API allows. Timestamps may be ISO 8601 strings or datetimes 

126 (naive datetimes are taken as UTC). Durations may be ISO 8601 strings 

127 (PT15M, P1D), integer seconds, or timedeltas. 

128 

129 Examples: 

130 ```python 

131 last_hour = TimeRange(duration="PT1H") 

132 one_day = TimeRange(start="2026-05-27T00:00:00Z", duration="P1D") 

133 ``` 

134 """ 

135 

136 start: str | datetime | None = None 

137 end: str | datetime | None = None 

138 duration: str | int | timedelta | None = None 

139 

140 def resolved(self) -> "TimeRange": 

141 """Return an equivalent range the server accepts, with all values as strings. 

142 

143 The server only accepts time-only durations (PT...), so durations 

144 with date parts (P1D, P1W) are expanded to explicit start and 

145 end timestamps. With no start and no duration, the range defaults to 

146 the hour ending at end, or now. 

147 """ 

148 start, end, duration = normalise_duration( 

149 _format_duration(self.duration), 

150 _format_time(self.start), 

151 _format_time(self.end), 

152 ) 

153 params = apply_time_default(_params(start, end, duration)) 

154 return TimeRange( 

155 start=params.get("from"), 

156 end=params.get("to"), 

157 duration=params.get("duration"), 

158 ) 

159 

160 def params(self) -> dict[str, str]: 

161 """Return the from/to/duration query parameters for this range, as given.""" 

162 return _params( 

163 _format_time(self.start), 

164 _format_time(self.end), 

165 _format_duration(self.duration), 

166 ) 

167 

168 

169def _params(start: str | None, end: str | None, duration: str | None) -> dict[str, str]: 

170 p: dict[str, str] = {} 

171 if start: 

172 p["from"] = start 

173 if end: 

174 p["to"] = end 

175 if duration: 

176 p["duration"] = duration 

177 return p