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

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: 17 ↛ 18line 17 didn't jump to line 18 because the condition on line 17 was never true

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

19 if hours: 19 ↛ 20line 19 didn't jump to line 20 because the condition on line 19 was never true

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

21 if minutes: 21 ↛ 22line 21 didn't jump to line 22 because the condition on line 21 was never true

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

23 if secs or not parts: 23 ↛ 26line 23 didn't jump to line 26 because the condition on line 23 was always true

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

25 

26 if len(parts) > 1: 26 ↛ 27line 26 didn't jump to line 27 because the condition on line 26 was never true

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