Coverage for src / ocarina / opinionated / plugins / reports / timing.py: 4.26%

35 statements  

« prev     ^ index     » next       coverage.py v7.13.5, created at 2026-06-04 15:56 +0200

1"""Context manager logging how long an action took to run.""" 

2 

3import time 

4from contextlib import contextmanager 

5 

6from ocarina.infra.drivers_pool import WarmupTimeoutError 

7 

8 

9def format_elapsed(seconds: float) -> str: 

10 """Format elapsed time in seconds.""" 

11 secs = int(seconds) 

12 days, secs = divmod(secs, 86400) 

13 hours, secs = divmod(secs, 3600) 

14 minutes, secs = divmod(secs, 60) 

15 

16 parts = [] 

17 if days: 

18 parts.append(f"{days} day{'s' if days > 1 else ''}") 

19 if hours: 

20 parts.append(f"{hours} hour{'s' if hours > 1 else ''}") 

21 if minutes: 

22 parts.append(f"{minutes} minute{'s' if minutes > 1 else ''}") 

23 if secs or not parts: 

24 parts.append(f"{secs} second{'s' if secs > 1 else ''}") 

25 

26 if len(parts) > 1: 

27 return ", ".join(parts[:-1]) + " and " + parts[-1] 

28 return parts[0] 

29 

30 

31@contextmanager 

32def timing(*, prefix: str = "Duration:", seconds_label: str = "seconds"): 

33 """Context manager logging how long an action took to run.""" 

34 start = time.perf_counter() 

35 interrupted = False 

36 

37 try: 

38 yield 

39 except KeyboardInterrupt, WarmupTimeoutError: 

40 interrupted = True 

41 raise 

42 finally: 

43 if not interrupted: # pragma: no branch 

44 end = time.perf_counter() 

45 elapsed = end - start 

46 

47 human_readable = format_elapsed(elapsed) 

48 

49 p = f"{prefix} " if prefix else "" 

50 print(f"\n{p}{human_readable} ({elapsed:.2f} {seconds_label})") # noqa: T201