Coverage for pyTooling/Tracing/__init__.py: 90%

1# ==================================================================================================================== # 

2# _____ _ _ _____ _ # 

3# _ __ _ |_ _|__ ___ | (_)_ __ __ _|_ _| __ __ _ ___(_)_ __ __ _ # 

4# | '_ \| | | || |/ _ \ / _ \| | | '_ \ / _` | | || '__/ _` |/ __| | '_ \ / _` | # 

5# | |_) | |_| || | (_) | (_) | | | | | | (_| |_| || | | (_| | (__| | | | | (_| | # 

6# | .__/ \__, ||_|\___/ \___/|_|_|_| |_|\__, (_)_||_| \__,_|\___|_|_| |_|\__, | # 

7# |_| |___/ |___/ |___/ # 

8# ==================================================================================================================== # 

9# Authors: # 

10# Patrick Lehmann # 

11# # 

12# License: # 

13# ==================================================================================================================== # 

14# Copyright 2025-2025 Patrick Lehmann - Bötzingen, Germany # 

15# # 

16# Licensed under the Apache License, Version 2.0 (the "License"); # 

17# you may not use this file except in compliance with the License. # 

18# You may obtain a copy of the License at # 

19# # 

20# http://www.apache.org/licenses/LICENSE-2.0 # 

21# # 

22# Unless required by applicable law or agreed to in writing, software # 

23# distributed under the License is distributed on an "AS IS" BASIS, # 

24# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # 

25# See the License for the specific language governing permissions and # 

26# limitations under the License. # 

27# # 

28# SPDX-License-Identifier: Apache-2.0 # 

29# ==================================================================================================================== # 

30# 

31"""Tools for software execution tracing.""" 

32from datetime import datetime 

33from time import perf_counter_ns 

34from threading import local 

35from types import TracebackType 

36from typing import Optional as Nullable, List, Iterator, Type, Self, Iterable 

37 

38try: 

39 from pyTooling.Decorators import export, readonly 

40 from pyTooling.MetaClasses import ExtendedType 

41 from pyTooling.Exceptions import ToolingException 

42except (ImportError, ModuleNotFoundError): # pragma: no cover 

43 print("[pyTooling.Tracing] Could not import from 'pyTooling.*'!") 

44 

45 try: 

46 from Decorators import export, readonly 

47 from MetaClasses import ExtendedType 

48 from Exceptions import ToolingException 

49 except (ImportError, ModuleNotFoundError) as ex: # pragma: no cover 

50 print("[pyTooling.Tracing] Could not import directly!") 

51 raise ex 

52 

53 

54_threadLocalData = local() 

55 

56 

57@export 

58class TracingException(ToolingException): 

59 """This exception is caused by wrong usage of the stopwatch.""" 

60 

61 

62@export 

63class Span(metaclass=ExtendedType, slots=True): 

64 _name: str 

65 _parent: "Span" 

66 _beginTime: Nullable[datetime] 

67 _endTime: Nullable[datetime] 

68 _startTime: Nullable[int] 

69 _stopTime: Nullable[int] 

70 _totalTime: Nullable[int] 

71 

72 _spans: List["Span"] 

73 

74 def __init__(self, name: str) -> None: 

75 self._name = name 

76 self._parent = None 

77 self._beginTime = None 

78 self._startTime = None 

79 self._endTime = None 

80 self._stopTime = None 

81 self._totalTime = None 

82 

83 self._spans = [] 

84 

85 @readonly 

86 def Name(self) -> str: 

87 return self._name 

88 

89 @readonly 

90 def Parent(self) -> "Span": 

91 return self._parent 

92 

93 def _AddSpan(self, span: "Span") -> Self: 

94 self._spans.append(span) 

95 span._parent = self 

96 

97 return span 

98 

99 @readonly 

100 def HasNestedSpans(self) -> bool: 

101 return len(self._spans) > 0 

102 

103 @readonly 

104 def StartTime(self) -> Nullable[datetime]: 

105 """ 

106 Read-only property returning the absolute time when the span was started. 

107 

108 :return: The time when the span was entered, otherwise None. 

109 """ 

110 return self._beginTime 

111 

112 @readonly 

113 def StopTime(self) -> Nullable[datetime]: 

114 """ 

115 Read-only property returning the absolute time when the span was stopped. 

116 

117 :return: The time when the span was exited, otherwise None. 

118 """ 

119 return self._endTime 

120 

121 @readonly 

122 def Duration(self) -> float: 

123 """ 

124 Read-only property returning the duration from start operation to stop operation. 

125 

126 If the stopwatch is not yet stopped, the duration from start to now is returned. 

127 

128 :return: Duration since stopwatch was started in seconds. If the stopwatch was never started, the return value will 

129 be 0.0. 

130 """ 

131 if self._startTime is None: 131 ↛ 132line 131 didn't jump to line 132 because the condition on line 131 was never true

132 raise TracingException(f"{self.__class__.__name__} was never started.") 

133 

134 return ((perf_counter_ns() - self._startTime) if self._stopTime is None else self._totalTime) / 1e9 

135 

136 @classmethod 

137 def CurrentSpan(cls) -> "Span": 

138 global _threadLocalData 

139 

140 try: 

141 currentSpan = _threadLocalData.currentSpan 

142 except AttributeError as ex: 

143 currentSpan = None 

144 

145 return currentSpan 

146 

147 def __enter__(self) -> Self: 

148 """ 

149 Implementation of the :ref:`context manager protocol's <context-managers>` ``__enter__(...)`` method. 

150 

151 A span will be started. 

152 

153 :return: The span itself. 

154 """ 

155 global _threadLocalData 

156 

157 try: 

158 currentSpan = _threadLocalData.currentSpan 

159 except AttributeError: 

160 ex = TracingException("Can't setup span. No active trace.") 

161 ex.add_note("Use with-statement using 'Trace()' to setup software execution tracing.") 

162 raise ex 

163 

164 _threadLocalData.currentSpan = currentSpan._AddSpan(self) 

165 

166 self._beginTime = datetime.now() 

167 self._startTime = perf_counter_ns() 

168 

169 return self 

170 

171 def __exit__( 

172 self, 

173 exc_type: Nullable[Type[BaseException]] = None, 

174 exc_val: Nullable[BaseException] = None, 

175 exc_tb: Nullable[TracebackType] = None 

176 ) -> Nullable[bool]: 

177 """ 

178 Implementation of the :ref:`context manager protocol's <context-managers>` ``__exit__(...)`` method. 

179 

180 An active span will be stopped. 

181 

182 Exit the context and ...... 

183 

184 :param exc_type: Exception type 

185 :param exc_val: Exception instance 

186 :param exc_tb: Exception's traceback. 

187 :returns: ``None`` 

188 """ 

189 global _threadLocalData 

190 

191 self._stopTime = perf_counter_ns() 

192 self._endTime = datetime.now() 

193 self._totalTime = self._stopTime - self._startTime 

194 

195 currentSpan = _threadLocalData.currentSpan 

196 _threadLocalData.currentSpan = currentSpan._parent 

197 

198 def __len__(self) -> int: 

199 """ 

200 Implementation of ``len(...)`` to return the number of nested spans. 

201 

202 :return: Number of nested spans. 

203 """ 

204 return len(self._spans) 

205 

206 def __iter__(self) -> Iterator["Span"]: 

207 return iter(self._spans) 

208 

209 def Format(self, indent: int = 1, columnSize: int = 25) -> Iterable[str]: 

210 result = [] 

211 result.append(f"{' ' * indent}🕑{self._name:<{columnSize - 2 * indent}} {self._totalTime/1e6:8.3f} ms") 

212 for span in self._spans: 

213 result.extend(span.Format(indent + 1, columnSize)) 

214 

215 return result 

216 

217 def __repr__(self) -> str: 

218 return f"{self._name} -> {self._parent!r}" 

219 

220 def __str__(self) -> str: 

221 return self._name 

222 

223 

224@export 

225class Trace(Span): 

226 def __init__(self, name: str) -> None: 

227 super().__init__(name) 

228 

229 def __enter__(self) -> Self: 

230 global _threadLocalData 

231 

232 # TODO: check if a trace is already setup 

233 # try: 

234 # currentTrace = _threadLocalData.currentTrace 

235 # except AttributeError: 

236 # pass 

237 

238 _threadLocalData.currentTrace = self 

239 _threadLocalData.currentSpan = self 

240 

241 self._beginTime = datetime.now() 

242 self._startTime = perf_counter_ns() 

243 

244 return self 

245 

246 def __exit__( 

247 self, 

248 exc_type: Nullable[Type[BaseException]] = None, 

249 exc_val: Nullable[BaseException] = None, 

250 exc_tb: Nullable[TracebackType] = None 

251 ) -> Nullable[bool]: 

252 """ 

253 Exit the context and ...... 

254 

255 :param exc_type: Exception type 

256 :param exc_val: Exception instance 

257 :param exc_tb: Exception's traceback. 

258 :returns: ``None`` 

259 """ 

260 global _threadLocalData 

261 

262 self._stopTime = perf_counter_ns() 

263 self._endTime = datetime.now() 

264 self._totalTime = self._stopTime - self._startTime 

265 

266 _threadLocalData.currentTrace = None 

267 _threadLocalData.currentSpan = None 

268 

269 return None 

270 

271 @classmethod 

272 def CurrentTrace(cls) -> "Trace": 

273 try: 

274 currentTrace = _threadLocalData.currentTrace 

275 except AttributeError: 

276 currentTrace = None 

277 

278 return currentTrace 

279 

280 def Format(self, indent: int = 0, columnSize: int = 25) -> Iterable[str]: 

281 result = [] 

282 result.append(f"{' ' * indent}Software Execution Trace: {self._totalTime/1e6:8.3f} ms") 

283 result.append(f"{' ' * indent}📉{self._name:<{columnSize - 2}} {self._totalTime/1e6:8.3f} ms") 

284 for span in self._spans: 

285 result.extend(span.Format(indent + 1, columnSize - 2)) 

286 

287 return result