Coverage for src / lexigram / contracts / core / clock.py: 0%

1"""Clock contracts for the Lexigram framework.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass 

6from datetime import datetime, timedelta 

7from functools import total_ordering 

8import re 

9from typing import Protocol, runtime_checkable 

10 

11_DURATION_PATTERN = re.compile(r"(\d+(?:\.\d+)?)([smhd])", re.IGNORECASE) 

12_UNIT_TO_SECONDS = { 

13 "s": 1.0, 

14 "m": 60.0, 

15 "h": 3600.0, 

16 "d": 86400.0, 

17} 

18 

19 

20@total_ordering 

21@dataclass(frozen=True, slots=True) 

22class Duration: 

23 """Immutable duration value object. 

24 

25 The class stores total seconds internally and provides lightweight parsing, 

26 formatting, and arithmetic helpers for configuration and API boundaries. 

27 """ 

28 

29 total_seconds: float 

30 

31 @classmethod 

32 def seconds(cls, value: float) -> Duration: 

33 """Create a duration from seconds.""" 

34 return cls(float(value)) 

35 

36 @classmethod 

37 def minutes(cls, value: float) -> Duration: 

38 """Create a duration from minutes.""" 

39 return cls(float(value) * 60.0) 

40 

41 @classmethod 

42 def hours(cls, value: float) -> Duration: 

43 """Create a duration from hours.""" 

44 return cls(float(value) * 3600.0) 

45 

46 @classmethod 

47 def days(cls, value: float) -> Duration: 

48 """Create a duration from days.""" 

49 return cls(float(value) * 86400.0) 

50 

51 @classmethod 

52 def zero(cls) -> Duration: 

53 """Create a zero duration.""" 

54 return cls(0.0) 

55 

56 @classmethod 

57 def parse(cls, value: str) -> Duration: 

58 """Parse a human-readable duration string. 

59 

60 Supported units are seconds (``s``), minutes (``m``), hours (``h``), 

61 and days (``d``). Chained forms like ``1h30m`` are supported. 

62 """ 

63 normalized = value.strip().lower() 

64 if not normalized: 

65 msg = "Duration string cannot be empty" 

66 raise ValueError(msg) 

67 

68 matches = list(_DURATION_PATTERN.finditer(normalized)) 

69 if not matches or "".join(match.group(0) for match in matches) != normalized: 

70 msg = f"Invalid duration string: {value!r}" 

71 raise ValueError(msg) 

72 

73 total = 0.0 

74 for match in matches: 

75 amount = float(match.group(1)) 

76 unit = match.group(2).lower() 

77 total += amount * _UNIT_TO_SECONDS[unit] 

78 return cls(total) 

79 

80 @property 

81 def seconds_value(self) -> float: 

82 """Backward-compatible alias for the total seconds.""" 

83 return self.total_seconds 

84 

85 def to_timedelta(self) -> timedelta: 

86 """Convert the duration to :class:`datetime.timedelta`.""" 

87 return timedelta(seconds=self.total_seconds) 

88 

89 def __add__(self, other: Duration) -> Duration: 

90 return Duration(self.total_seconds + other.total_seconds) 

91 

92 def __sub__(self, other: Duration) -> Duration: 

93 return Duration(self.total_seconds - other.total_seconds) 

94 

95 def __mul__(self, factor: float) -> Duration: 

96 return Duration(self.total_seconds * float(factor)) 

97 

98 def __rmul__(self, factor: float) -> Duration: 

99 return self * factor 

100 

101 def __eq__(self, other: object) -> bool: 

102 if not isinstance(other, Duration): 

103 return NotImplemented 

104 return self.total_seconds == other.total_seconds 

105 

106 def __hash__(self) -> int: 

107 return hash(self.total_seconds) 

108 

109 def __lt__(self, other: Duration) -> bool: 

110 if not isinstance(other, Duration): 

111 return NotImplemented 

112 return self.total_seconds < other.total_seconds 

113 

114 def __bool__(self) -> bool: 

115 return self.total_seconds != 0.0 

116 

117 def __str__(self) -> str: 

118 total = self.total_seconds 

119 if total == 0: 

120 return "0s" 

121 

122 remaining = abs(total) 

123 parts: list[str] = [] 

124 for suffix, unit_seconds in (("d", 86400.0), ("h", 3600.0), ("m", 60.0)): 

125 amount = int(remaining // unit_seconds) 

126 if amount: 

127 parts.append(f"{amount}{suffix}") 

128 remaining -= amount * unit_seconds 

129 if remaining: 

130 if remaining.is_integer(): 

131 parts.append(f"{int(remaining)}s") 

132 else: 

133 parts.append(f"{remaining:g}s") 

134 

135 rendered = "".join(parts) or "0s" 

136 return f"-{rendered}" if total < 0 else rendered 

137 

138 

139@runtime_checkable 

140class ClockProtocol(Protocol): 

141 """Injectable time source for the framework.""" 

142 

143 def now(self) -> datetime: 

144 """Return the current timezone-aware UTC time.""" 

145 ... 

146 

147 def monotonic(self) -> float: 

148 """Return a monotonic elapsed-time counter.""" 

149 ... 

150 

151 def timestamp(self) -> float: 

152 """Return the current Unix timestamp in UTC seconds.""" 

153 ... 

154 

155 def time(self) -> float: 

156 """Backward-compatible alias for :meth:`timestamp`.""" 

157 ... 

158 

159 

160__all__ = ["ClockProtocol", "Duration"]