Coverage for core / src / sensorkit / common / logging.py: 71%

1# SPDX-License-Identifier: Apache-2.0 

2from __future__ import annotations 

3 

4import functools 

5import os 

6import pathlib 

7import sys 

8import time 

9from datetime import datetime 

10from typing import TYPE_CHECKING, Any, Callable, Self, cast 

11 

12from loguru import logger 

13 

14if TYPE_CHECKING: 

15 from loguru import Logger 

16else: 

17 Logger = Any 

18 

19DEFAULT_DEBUG_FILE = "sensorkit.log" 

20DEFAULT_LIMIT_INTERVAL = 5.0 

21 

22 

23def _format_time(dt: datetime): 

24 return dt.strftime(f"[%m/%d/%y %H:%M:%S.{dt.microsecond // 1000}]") 

25 

26 

27def configure_logging( 

28 *, 

29 level: str | None = None, 

30 format: str | Callable | None = None, 

31 force_color: bool = True, 

32): 

33 if level is None: 

34 level = "DEBUG" if os.environ.get("SENSORKIT_DEBUG") else "INFO" 

35 

36 if force_color: 

37 os.environ["FORCE_COLOR"] = "1" 

38 

39 if format is None: 

40 format = ( 

41 "<green>[{time:YYYY-MM-DD HH:mm:ss.SSS}]</green> " 

42 "<level>{level: <8}</level> " 

43 "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> " 

44 "<level>{message}</level>" 

45 ) 

46 

47 logger.remove() 

48 logger.add( 

49 sys.stderr, 

50 format=format, 

51 level=level, 

52 colorize=True, 

53 diagnose=False, 

54 backtrace=False, 

55 ) 

56 

57 

58def add_debug_logger( 

59 *, 

60 file: str | None = None, 

61 syslog_id: str = "sensorkit", 

62 append: bool = True, 

63 backtrace: bool = True, 

64 diagnose: bool = True, 

65) -> str: 

66 if file is None: 

67 try: 

68 from systemd.journal import JournalHandler 

69 

70 logger.add( 

71 JournalHandler(SYSLOG_IDENTIFIER=syslog_id), 

72 level="DEBUG", 

73 colorize=False, 

74 enqueue=True, 

75 backtrace=backtrace, 

76 diagnose=diagnose, 

77 ) 

78 return f"system log -- run: `journalctl -t {syslog_id} -f` to watch" 

79 except ImportError: 

80 # systemd not available, fall back to default. 

81 pass 

82 

83 path = pathlib.Path(file or DEFAULT_DEBUG_FILE) 

84 logger.add( 

85 path, 

86 mode="a" if append else "w", 

87 level="DEBUG", 

88 colorize=False, 

89 enqueue=True, 

90 backtrace=True, 

91 diagnose=True, 

92 ) 

93 return str(path.absolute()) 

94 

95 

96class NullLogger: 

97 """Stand-in for a loguru logger that discards everything sent to it. 

98 

99 Any attribute resolves to a call accepting anything and returning the same 

100 object, so chained forms such as `opt(exception=e).warning(...)` stay valid. 

101 """ 

102 

103 def __getattr__(self, name: str) -> Callable[..., Self]: 

104 return self.discard 

105 

106 def discard(self, *args: Any, **kwargs: Any) -> Self: 

107 return self 

108 

109 

110class RateLimiter: 

111 __slots__ = ("deadline",) 

112 

113 def __init__(self): 

114 self.deadline = float("-inf") 

115 

116 def allow(self, interval: float) -> bool: 

117 now = time.monotonic() 

118 

119 if now < self.deadline: 

120 return False 

121 

122 self.deadline = now + interval 

123 return True 

124 

125 

126NULL_LOGGER = cast(Logger, cast(object, NullLogger())) 

127 

128 

129@functools.lru_cache(maxsize=1024) 

130def _logger_limiter(_: str) -> RateLimiter: 

131 return RateLimiter() 

132 

133 

134def limited_logger(subject: str | None = None, *, interval: float = DEFAULT_LIMIT_INTERVAL) -> Logger: 

135 """Get a logger that emits at most once per interval for the calling site. 

136 

137 The limit applies to the call site rather than to message content, so while a 

138 site is limited, every call made through the returned logger is discarded. The 

139 first call always emits and opens the next window. 

140 

141 The result *must* used immediately, as in `limited_logger().info(...)`. 

142 Acquiring it is what consumes the window, so holding one in a variable and 

143 reusing it defeats the limit. For the same reason, wrapping this function in a 

144 helper keys every one of that helper's callers to the helper's own line. 

145 

146 Args: 

147 subject: Optional qualifier dividing one call site into independent limits. 

148 Sites are keyed by file and line, so equal subjects arising at different 

149 call sites never share a window. 

150 interval: Minimum seconds between emissions. It is read on every call, so a 

151 site can be limited at different rates as conditions change. 

152 

153 Returns: 

154 The loguru logger when the call site is outside its window, otherwise a 

155 stand-in that discards everything sent to it. 

156 """ 

157 import inspect 

158 

159 frame = inspect.currentframe() 

160 caller = frame.f_back if frame is not None else None 

161 

162 if caller is None: 

163 # No frame support, so there is no call site to key on. Emit unlimited 

164 # rather than share one window across every unrelated call site. 

165 return logger 

166 

167 key = f"{caller.f_code.co_filename}:{caller.f_lineno}" 

168 

169 if subject is not None: 

170 key = f"{key}:{subject}" 

171 

172 if _logger_limiter(key).allow(interval): 

173 return logger 

174 

175 return NULL_LOGGER