Coverage for pyTooling/Stopwatch/__init__.py: 86%

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

2# _____ _ _ ____ _ _ _ # 

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

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

5# | |_) | |_| || | (_) | (_) | | | | | | (_| |_ ___) | || (_) | |_) \ V V / (_| | || (__| | | | # 

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

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

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

9# Authors: # 

10# Patrick Lehmann # 

11# # 

12# License: # 

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

14# Copyright 2017-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""" 

32A stopwatch to measure execution times. 

33 

34.. hint:: See :ref:`high-level help <COMMON/Stopwatch>` for explanations and usage examples. 

35""" 

36 

37from datetime import datetime 

38from inspect import Traceback 

39from time import perf_counter_ns 

40from types import TracebackType 

41from typing import List, Optional as Nullable, Iterator, Tuple, Type 

42 

43# Python 3.11: use Self if returning the own object: , Self 

44 

45try: 

46 from pyTooling.Decorators import export, readonly 

47 from pyTooling.MetaClasses import SlottedObject 

48 from pyTooling.Exceptions import ToolingException 

49 from pyTooling.Platform import CurrentPlatform 

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

51 print("[pyTooling.Stopwatch] Could not import from 'pyTooling.*'!") 

52 

53 try: 

54 from Decorators import export, readonly 

55 from MetaClasses import SlottedObject 

56 from Exceptions import ToolingException 

57 from Platform import CurrentPlatform 

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

59 print("[pyTooling.Stopwatch] Could not import directly!") 

60 raise ex 

61 

62 

63@export 

64class StopwatchException(ToolingException): 

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

66 

67 

68@export 

69class ExcludeContextManager: 

70 """ 

71 A stopwatch context manager for excluding certain time spans from measurement. 

72 

73 While a normal stopwatch's embedded context manager (re)starts the stopwatch on every *enter* event and pauses the 

74 stopwatch on every *exit* event, this context manager pauses on *enter* events and restarts on every *exit* event. 

75 """ 

76 _stopwatch: "Stopwatch" #: Reference to the stopwatch. 

77 

78 def __init__(self, stopwatch: "Stopwatch") -> None: 

79 """ 

80 Initializes an excluding context manager. 

81 

82 :param stopwatch: Reference to the stopwatch. 

83 """ 

84 self._stopwatch = stopwatch 

85 

86 def __enter__(self) -> "ExcludeContextManager": # TODO: Python 3.11: -> Self: 

87 """ 

88 Enter the context and pause the stopwatch. 

89 

90 :returns: Excluding stopwatch context manager instance. 

91 """ 

92 self._stopwatch.Pause() 

93 

94 return self 

95 

96 def __exit__( 

97 self, 

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

99 exc_val: Nullable[BaseException] = None, 

100 exc_tb: Nullable[TracebackType] = None 

101 ) -> Nullable[bool]: 

102 """ 

103 Exit the context and restart stopwatch. 

104 

105 :param exc_type: Exception type 

106 :param exc_val: Exception instance 

107 :param exc_tb: Exception's traceback. 

108 :returns: ``None`` 

109 """ 

110 self._stopwatch.Resume() 

111 

112 

113@export 

114class Stopwatch(SlottedObject): 

115 """ 

116 The stopwatch implements a solution to measure and collect timings. 

117 

118 The time measurement can be started, paused, resumed and stopped. More over, split times can be taken too. The 

119 measurement is based on :func:`time.perf_counter_ns`. Additionally, starting and stopping is preserved as absolute 

120 time via :meth:`datetime.datetime.now`. 

121 

122 Every split time taken is a time delta to the previous operation. These are preserved in an internal sequence of 

123 splits. This sequence includes time deltas of activity and inactivity. Thus, a running stopwatch can be split as well 

124 as a paused stopwatch. 

125 

126 The stopwatch can also be used in a :ref:`with-statement <with>`, because it implements the :ref:`context manager protocol <context-managers>`. 

127 """ 

128 

129 _name: Nullable[str] 

130 _preferPause: bool 

131 

132 _beginTime: Nullable[datetime] 

133 _endTime: Nullable[datetime] 

134 _startTime: Nullable[int] 

135 _resumeTime: Nullable[int] 

136 _pauseTime: Nullable[int] 

137 _stopTime: Nullable[int] 

138 _totalTime: Nullable[int] 

139 _splits: List[Tuple[float, bool]] 

140 

141 _excludeContextManager: ExcludeContextManager 

142 

143 def __init__(self, name: str = None, started: bool = False, preferPause: bool = False) -> None: 

144 """ 

145 Initializes the fields of the stopwatch. 

146 

147 If parameter ``started`` is set to true, the stopwatch will immediately start. 

148 

149 :param name: Optional name of the stopwatch. 

150 :param preferPause: Optional setting, if __exit__(...) in a contex should prefer pause or stop behavior. 

151 :param started: Optional flag, if the stopwatch should be started immediately. 

152 """ 

153 self._name = name 

154 self._preferPause = preferPause 

155 

156 self._endTime = None 

157 self._pauseTime = None 

158 self._stopTime = None 

159 self._totalTime = None 

160 self._splits = [] 

161 

162 self._excludeContextManager = None 

163 

164 if started is False: 164 ↛ 169line 164 didn't jump to line 169 because the condition on line 164 was always true

165 self._beginTime = None 

166 self._startTime = None 

167 self._resumeTime = None 

168 else: 

169 self._beginTime = datetime.now() 

170 self._resumeTime = self._startTime = perf_counter_ns() 

171 

172 def Start(self) -> None: 

173 """ 

174 Start the stopwatch. 

175 

176 A stopwatch can only be started once. There is no restart or reset operation provided. 

177 

178 :raises StopwatchException: If stopwatch was already started. 

179 :raises StopwatchException: If stopwatch was already started and stopped. 

180 """ 

181 if self._startTime is not None: 

182 raise StopwatchException("Stopwatch was already started.") 

183 if self._stopTime is not None: 183 ↛ 184line 183 didn't jump to line 184 because the condition on line 183 was never true

184 raise StopwatchException("Stopwatch was already used (started and stopped).") 

185 

186 self._beginTime = datetime.now() 

187 self._resumeTime = self._startTime = perf_counter_ns() 

188 

189 def Split(self) -> float: 

190 """ 

191 Take a split time and return the time delta to the previous stopwatch operation. 

192 

193 The stopwatch needs to be running to take a split time. See property :data:`IsRunning` to check if the stopwatch 

194 is running and the split operation is possible. |br| 

195 Depending on the previous operation, the time delta will be: 

196 

197 * the duration from start operation to the first split. 

198 * the duration from last resume to this split. 

199 

200 :returns: Duration in seconds since last stopwatch operation 

201 :raises StopwatchException: If stopwatch was not started or resumed. 

202 """ 

203 pauseTime = perf_counter_ns() 

204 

205 if self._resumeTime is None: 

206 raise StopwatchException("Stopwatch was not started or resumed.") 

207 

208 diff = (pauseTime - self._resumeTime) / 1e9 

209 self._splits.append((diff, True)) 

210 self._resumeTime = pauseTime 

211 

212 return diff 

213 

214 def Pause(self) -> float: 

215 """ 

216 Pause the stopwatch and return the time delta to the previous stopwatch operation. 

217 

218 The stopwatch needs to be running to pause it. See property :data:`IsRunning` to check if the stopwatch is running 

219 and the pause operation is possible. |br| 

220 Depending on the previous operation, the time delta will be: 

221 

222 * the duration from start operation to the first pause. 

223 * the duration from last resume to this pause. 

224 

225 :returns: Duration in seconds since last stopwatch operation 

226 :raises StopwatchException: If stopwatch was not started or resumed. 

227 """ 

228 self._pauseTime = perf_counter_ns() 

229 

230 if self._resumeTime is None: 

231 raise StopwatchException("Stopwatch was not started or resumed.") 

232 

233 diff = (self._pauseTime - self._resumeTime) / 1e9 

234 self._splits.append((diff, True)) 

235 self._resumeTime = None 

236 

237 return diff 

238 

239 def Resume(self) -> float: 

240 """ 

241 Resume the stopwatch and return the time delta to the previous pause operation. 

242 

243 The stopwatch needs to be paused to resume it. See property :data:`IsPaused` to check if the stopwatch is paused 

244 and the resume operation is possible. |br| 

245 The time delta will be the duration from last pause to this resume. 

246 

247 :returns: Duration in seconds since last pause operation 

248 :raises StopwatchException: If stopwatch was not paused. 

249 """ 

250 self._resumeTime = perf_counter_ns() 

251 

252 if self._pauseTime is None: 

253 raise StopwatchException("Stopwatch was not paused.") 

254 

255 diff = (self._resumeTime - self._pauseTime) / 1e9 

256 self._splits.append((diff, False)) 

257 self._pauseTime = None 

258 

259 return diff 

260 

261 def Stop(self): 

262 """ 

263 Stop the stopwatch and return the time delta to the previous stopwatch operation. 

264 

265 The stopwatch needs to be started to stop it. See property :data:`IsStarted` to check if the stopwatch was started 

266 and the stop operation is possible. |br| 

267 Depending on the previous operation, the time delta will be: 

268 

269 * the duration from start operation to the stop operation. 

270 * the duration from last resume to the stop operation. 

271 

272 :returns: Duration in seconds since last stopwatch operation 

273 :raises StopwatchException: If stopwatch was not started. 

274 :raises StopwatchException: If stopwatch was already stopped. 

275 """ 

276 self._stopTime = perf_counter_ns() 

277 self._endTime = datetime.now() 

278 

279 if self._startTime is None: 

280 raise StopwatchException("Stopwatch was never started.") 

281 if self._totalTime is not None: 281 ↛ 282line 281 didn't jump to line 282 because the condition on line 281 was never true

282 raise StopwatchException("Stopwatch was already stopped.") 

283 

284 if len(self._splits) == 0: # was never paused 

285 diff = (self._stopTime - self._startTime) / 1e9 

286 elif self._resumeTime is None: # is paused 

287 diff = (self._stopTime - self._pauseTime) / 1e9 

288 self._splits.append((diff, False)) 

289 else: # is running 

290 diff = (self._stopTime - self._resumeTime) / 1e9 

291 self._splits.append((diff, True)) 

292 

293 self._pauseTime = None 

294 self._resumeTime = None 

295 self._totalTime = self._stopTime - self._startTime 

296 

297 beginEndDiff = self._endTime - self._beginTime 

298 

299 return diff 

300 

301 @readonly 

302 def Name(self) -> Nullable[str]: 

303 """ 

304 Read-only property returning the name of the stopwatch. 

305 

306 :return: Name of the stopwatch. 

307 """ 

308 return self._name 

309 

310 @readonly 

311 def IsStarted(self) -> bool: 

312 """ 

313 Read-only property returning the IsStarted state of the stopwatch. 

314 

315 :return: True, if stopwatch was started. 

316 """ 

317 return self._startTime is not None and self._stopTime is None 

318 

319 @readonly 

320 def IsRunning(self) -> bool: 

321 """ 

322 Read-only property returning the IsRunning state of the stopwatch. 

323 

324 :return: True, if stopwatch was started and is currently not paused. 

325 """ 

326 return self._startTime is not None and self._resumeTime is not None 

327 

328 @readonly 

329 def IsPaused(self) -> bool: 

330 """ 

331 Read-only property returning the IsPaused state of the stopwatch. 

332 

333 :return: True, if stopwatch was started and is currently paused. 

334 """ 

335 return self._startTime is not None and self._pauseTime is not None 

336 

337 @readonly 

338 def IsStopped(self) -> bool: 

339 """ 

340 Read-only property returning the IsStopped state of the stopwatch. 

341 

342 :return: True, if stopwatch was stopped. 

343 """ 

344 return self._stopTime is not None 

345 

346 @readonly 

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

348 """ 

349 Read-only property returning the absolute time when the stopwatch was started. 

350 

351 :return: The time when the stopwatch was started, otherwise None. 

352 """ 

353 return self._beginTime 

354 

355 @readonly 

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

357 """ 

358 Read-only property returning the absolute time when the stopwatch was stopped. 

359 

360 :return: The time when the stopwatch was stopped, otherwise None. 

361 """ 

362 return self._endTime 

363 

364 @readonly 

365 def HasSplitTimes(self) -> bool: 

366 """ 

367 Read-only property checking if split times have been taken. 

368 

369 :return: True, if split times have been taken. 

370 """ 

371 return len(self._splits) > 1 

372 

373 @readonly 

374 def SplitCount(self) -> int: 

375 """ 

376 Read-only property returning the number of split times. 

377 

378 :return: Number of split times. 

379 """ 

380 return len(self._splits) 

381 

382 @readonly 

383 def ActiveCount(self) -> int: 

384 """ 

385 Read-only property returning the number of active split times. 

386 

387 :return: Number of active split times. 

388 

389 .. warning:: 

390 

391 This won't include all activities, unless the stopwatch got stopped. 

392 """ 

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

394 return 0 

395 

396 return len(list(t for t, a in self._splits if a is True)) 

397 

398 @readonly 

399 def InactiveCount(self) -> int: 

400 """ 

401 Read-only property returning the number of active split times. 

402 

403 :return: Number of active split times. 

404 

405 .. warning:: 

406 

407 This won't include all inactivities, unless the stopwatch got stopped. 

408 """ 

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

410 return 0 

411 

412 return len(list(t for t, a in self._splits if a is False)) 

413 

414 @readonly 

415 def Activity(self) -> float: 

416 """ 

417 Read-only property returning the duration of all active split times. 

418 

419 If the stopwatch is currently running, the duration since start or last resume operation will be included. 

420 

421 :return: Duration of all active split times in seconds. If the stopwatch was never started, the return value will 

422 be 0.0. 

423 """ 

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

425 return 0.0 

426 

427 currentDiff = 0.0 if self._resumeTime is None else ((perf_counter_ns() - self._resumeTime) / 1e9) 

428 return sum(t for t, a in self._splits if a is True) + currentDiff 

429 

430 @readonly 

431 def Inactivity(self) -> float: 

432 """ 

433 Read-only property returning the duration of all inactive split times. 

434 

435 If the stopwatch is currently paused, the duration since last pause operation will be included. 

436 

437 :return: Duration of all inactive split times in seconds. If the stopwatch was never started, the return value will 

438 be 0.0. 

439 """ 

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

441 return 0.0 

442 

443 currentDiff = 0.0 if self._pauseTime is None else ((perf_counter_ns() - self._pauseTime) / 1e9) 

444 return sum(t for t, a in self._splits if a is False) + currentDiff 

445 

446 @readonly 

447 def Duration(self) -> float: 

448 """ 

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

450 

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

452 

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

454 be 0.0. 

455 """ 

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

457 return 0.0 

458 

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

460 

461 @readonly 

462 def Exclude(self) -> ExcludeContextManager: 

463 """ 

464 Return an *exclude* context manager for the stopwatch instance. 

465 

466 :returns: An excluding context manager. 

467 """ 

468 if self._excludeContextManager is None: 

469 excludeContextManager = ExcludeContextManager(self) 

470 self._excludeContextManager = excludeContextManager 

471 

472 return excludeContextManager 

473 

474 def __enter__(self) -> "Stopwatch": # TODO: Python 3.11: -> Self: 

475 """ 

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

477 

478 An unstarted stopwatch will be started. A paused stopwatch will be resumed. 

479 

480 :return: The stopwatch itself. 

481 """ 

482 if self._startTime is None: # start stopwatch 

483 self._beginTime = datetime.now() 

484 self._resumeTime = self._startTime = perf_counter_ns() 

485 elif self._pauseTime is not None: # resume after pause 

486 self._resumeTime = perf_counter_ns() 

487 

488 diff = (self._resumeTime - self._pauseTime) / 1e9 

489 self._splits.append((diff, False)) 

490 self._pauseTime = None 

491 elif self._resumeTime is not None: # is running? 491 ↛ 492line 491 didn't jump to line 492 because the condition on line 491 was never true

492 raise StopwatchException("Stopwatch is currently running and can not be started/resumed again.") 

493 elif self._stopTime is not None: # is stopped? 493 ↛ 496line 493 didn't jump to line 496 because the condition on line 493 was always true

494 raise StopwatchException(f"Stopwatch was already stopped.") 

495 else: 

496 raise StopwatchException(f"Internal error.") 

497 

498 return self 

499 

500 def __exit__(self, exc_type: Type[Exception], exc_val: Exception, exc_tb: Traceback) -> bool: 

501 """ 

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

503 

504 A running stopwatch will be paused or stopped depending on the configured ``preferPause`` behavior. 

505 

506 :param exc_type: Exception type, otherwise None. 

507 :param exc_val: Exception object, otherwise None. 

508 :param exc_tb: Exception's traceback, otherwise None. 

509 :returns: True, if exceptions should be suppressed. 

510 """ 

511 if self._startTime is None: # never started? 511 ↛ 512line 511 didn't jump to line 512 because the condition on line 511 was never true

512 raise StopwatchException("Stopwatch was never started.") 

513 elif self._stopTime is not None: 513 ↛ 514line 513 didn't jump to line 514 because the condition on line 513 was never true

514 raise StopwatchException("Stopwatch was already stopped.") 

515 elif self._resumeTime is not None: # pause or stop 515 ↛ 532line 515 didn't jump to line 532 because the condition on line 515 was always true

516 if self._preferPause: 

517 self._pauseTime = perf_counter_ns() 

518 diff = (self._pauseTime - self._resumeTime) / 1e9 

519 self._splits.append((diff, True)) 

520 self._resumeTime = None 

521 else: 

522 self._stopTime = perf_counter_ns() 

523 self._endTime = datetime.now() 

524 

525 diff = (self._stopTime - self._resumeTime) / 1e9 

526 self._splits.append((diff, True)) 

527 

528 self._pauseTime = None 

529 self._resumeTime = None 

530 self._totalTime = self._stopTime - self._startTime 

531 else: 

532 raise StopwatchException("Stopwatch was not resumed.") 

533 

534 

535 def __len__(self): 

536 """ 

537 Implementation of ``len(...)`` to return the number of split times. 

538 

539 :return: Number of split times. 

540 """ 

541 return len(self._splits) 

542 

543 def __getitem__(self, index: int) -> Tuple[float, bool]: 

544 """ 

545 Implementation of ``split = object[i]`` to return the i-th split time. 

546 

547 :param index: Index to access the i-th split time. 

548 :return: i-th split time as a tuple of: |br| 

549 (1) delta time to the previous stopwatch operation and |br| 

550 (2) a boolean indicating if the split was an activity (true) or inactivity (false). 

551 :raises KeyError: If index *i* doesn't exist. 

552 """ 

553 return self._splits[index] 

554 

555 def __iter__(self) -> Iterator[Tuple[float, bool]]: 

556 """ 

557 Return an iterator of tuples to iterate all split times. 

558 

559 If the stopwatch is not stopped yet, the last split won't be included. 

560 

561 :return: Iterator of split time tuples of: |br| 

562 (1) delta time to the previous stopwatch operation and |br| 

563 (2) a boolean indicating if the split was an activity (true) or inactivity (false). 

564 """ 

565 return self._splits.__iter__() 

566 

567 def __str__(self): 

568 """ 

569 Returns the stopwatch's state and its measured time span. 

570 

571 :returns: The string equivalent of the stopwatch. 

572 """ 

573 name = f" {self._name}" if self._name is not None else "" 

574 if self.IsStopped: 

575 return f"Stopwatch{name} (stopped): {self._beginTime} -> {self._endTime}: {self._totalTime}" 

576 elif self.IsRunning: 

577 return f"Stopwatch{name} (running): {self._beginTime} -> now: {self.Duration}" 

578 elif self.IsPaused: 

579 return f"Stopwatch{name} (paused): {self._beginTime} -> now: {self.Duration}" 

580 else: 

581 return f"Stopwatch{name}: not started"