aria.ops.timer
1from __future__ import annotations 2 3from logging import Logger 4from time import time 5from types import TracebackType 6from typing import Iterable 7from typing import List 8from typing import Optional 9from typing import Tuple 10from typing import Type 11 12 13class Timer: 14 timers: List[Tuple[str, float, float]] = [] 15 16 def __init__(self, logger: Logger, name: str) -> None: 17 self.logger = logger 18 self.name = name 19 20 def __enter__(self) -> Timer: 21 self.logger.info(f"Starting '{self.name}'") 22 self.start_time = time() 23 return self 24 25 def __aenter__(self) -> Timer: 26 self.logger.info(f"Starting '{self.name}'") 27 self.start_time = time() 28 return self 29 30 def __exit__( 31 self, 32 exc_type: Optional[Type[BaseException]] = None, 33 exc_value: Optional[BaseException] = None, 34 traceback: Optional[TracebackType] = None, 35 ) -> None: 36 end_time = time() 37 self.timers.append((self.name, self.start_time, end_time)) 38 self.logger.info( 39 f"Finished '{self.name}' in {_to_time(end_time - self.start_time)}" 40 ) 41 42 def __aexit__( 43 self, 44 exc_type: Optional[Type[BaseException]] = None, 45 exc_value: Optional[BaseException] = None, 46 traceback: Optional[TracebackType] = None, 47 ) -> None: 48 end_time = time() 49 self.timers.append((self.name, self.start_time, end_time)) 50 self.logger.info( 51 f"Finished '{self.name}' in {_to_time(end_time - self.start_time)}" 52 ) 53 54 @classmethod 55 def graph(cls) -> str: 56 headers = ["Operation", "Time"] 57 full_headers = [] 58 width = 88 59 time_min = time() 60 time_max = 0.0 61 name_max = len(headers[0]) 62 for name, start, end in cls.timers: 63 if start < time_min: 64 time_min = start 65 if end > time_max: 66 time_max = end 67 if len(name) > name_max: 68 name_max = len(name) 69 70 graph_width = width - name_max - 10 71 step = (time_max - time_min) / graph_width 72 headers.append("t=0s") 73 headers.append(f"t={_to_time(time_max - time_min)}") 74 75 full_headers.append(f"{headers[0]:<{name_max}}") 76 full_headers.append(f"{headers[1]:<8}") 77 full_headers.append( 78 f"{headers[2]:<{int(graph_width / 2)}}{headers[3]:>{int((graph_width + 1) / 2)}}" 79 ) 80 81 sorted_intervals: Iterable[Tuple[str, float, float]] = sorted( 82 cls.timers, key=lambda i: i[1] 83 ) 84 graph = "Timing Graph: \n" 85 graph += _hline(full_headers, "U") 86 graph += "│".join(full_headers) + "\n" 87 graph += _hline(full_headers) 88 for name, start, end in sorted_intervals: 89 graph += ( 90 f"{name:<{name_max}}│" 91 f"{_to_time(end - start):<8}│" 92 + _graph_timespan( 93 (start - time_min) / step, (end - time_min) / step, graph_width 94 ) 95 + "\n" 96 ) 97 graph += _hline(full_headers, "L") 98 return graph 99 100 101def _hline(headers: List, loc: str = "M") -> str: 102 hl = "" 103 for header in headers: 104 for x in range(len(header)): 105 hl += "━" 106 if loc == "U": 107 hl += "┯" 108 elif loc == "M": 109 hl += "┿" 110 else: # loc = L 111 hl += "┷" 112 return f"{hl[:-1]}\n" 113 114 115def _graph_timespan(start: float, end: float, graph_width: int) -> str: 116 line = "" 117 118 def in_range(x_coord: float) -> bool: 119 return start <= x_coord < end or x_coord - 1 < start <= x_coord 120 121 for x in range(graph_width): 122 if in_range(x - 1) and in_range(x) and in_range(x + 1): 123 line += "─" 124 elif in_range(x - 1) and in_range(x): 125 line += "╼" 126 elif in_range(x) and in_range(x + 1): 127 line += "╾" 128 elif in_range(x): 129 line += "━" 130 else: 131 line += " " 132 return f"{line}" 133 134 135def _to_time(seconds: float) -> str: 136 if seconds > 3600: 137 hours = seconds / 3600 138 return f"{hours:.2f}h" 139 elif seconds > 120: 140 minutes = seconds / 60 141 return f"{minutes:.2f}m" 142 return f"{seconds:.2f}s"
class
Timer:
14class Timer: 15 timers: List[Tuple[str, float, float]] = [] 16 17 def __init__(self, logger: Logger, name: str) -> None: 18 self.logger = logger 19 self.name = name 20 21 def __enter__(self) -> Timer: 22 self.logger.info(f"Starting '{self.name}'") 23 self.start_time = time() 24 return self 25 26 def __aenter__(self) -> Timer: 27 self.logger.info(f"Starting '{self.name}'") 28 self.start_time = time() 29 return self 30 31 def __exit__( 32 self, 33 exc_type: Optional[Type[BaseException]] = None, 34 exc_value: Optional[BaseException] = None, 35 traceback: Optional[TracebackType] = None, 36 ) -> None: 37 end_time = time() 38 self.timers.append((self.name, self.start_time, end_time)) 39 self.logger.info( 40 f"Finished '{self.name}' in {_to_time(end_time - self.start_time)}" 41 ) 42 43 def __aexit__( 44 self, 45 exc_type: Optional[Type[BaseException]] = None, 46 exc_value: Optional[BaseException] = None, 47 traceback: Optional[TracebackType] = None, 48 ) -> None: 49 end_time = time() 50 self.timers.append((self.name, self.start_time, end_time)) 51 self.logger.info( 52 f"Finished '{self.name}' in {_to_time(end_time - self.start_time)}" 53 ) 54 55 @classmethod 56 def graph(cls) -> str: 57 headers = ["Operation", "Time"] 58 full_headers = [] 59 width = 88 60 time_min = time() 61 time_max = 0.0 62 name_max = len(headers[0]) 63 for name, start, end in cls.timers: 64 if start < time_min: 65 time_min = start 66 if end > time_max: 67 time_max = end 68 if len(name) > name_max: 69 name_max = len(name) 70 71 graph_width = width - name_max - 10 72 step = (time_max - time_min) / graph_width 73 headers.append("t=0s") 74 headers.append(f"t={_to_time(time_max - time_min)}") 75 76 full_headers.append(f"{headers[0]:<{name_max}}") 77 full_headers.append(f"{headers[1]:<8}") 78 full_headers.append( 79 f"{headers[2]:<{int(graph_width / 2)}}{headers[3]:>{int((graph_width + 1) / 2)}}" 80 ) 81 82 sorted_intervals: Iterable[Tuple[str, float, float]] = sorted( 83 cls.timers, key=lambda i: i[1] 84 ) 85 graph = "Timing Graph: \n" 86 graph += _hline(full_headers, "U") 87 graph += "│".join(full_headers) + "\n" 88 graph += _hline(full_headers) 89 for name, start, end in sorted_intervals: 90 graph += ( 91 f"{name:<{name_max}}│" 92 f"{_to_time(end - start):<8}│" 93 + _graph_timespan( 94 (start - time_min) / step, (end - time_min) / step, graph_width 95 ) 96 + "\n" 97 ) 98 graph += _hline(full_headers, "L") 99 return graph
@classmethod
def
graph(cls) -> str:
55 @classmethod 56 def graph(cls) -> str: 57 headers = ["Operation", "Time"] 58 full_headers = [] 59 width = 88 60 time_min = time() 61 time_max = 0.0 62 name_max = len(headers[0]) 63 for name, start, end in cls.timers: 64 if start < time_min: 65 time_min = start 66 if end > time_max: 67 time_max = end 68 if len(name) > name_max: 69 name_max = len(name) 70 71 graph_width = width - name_max - 10 72 step = (time_max - time_min) / graph_width 73 headers.append("t=0s") 74 headers.append(f"t={_to_time(time_max - time_min)}") 75 76 full_headers.append(f"{headers[0]:<{name_max}}") 77 full_headers.append(f"{headers[1]:<8}") 78 full_headers.append( 79 f"{headers[2]:<{int(graph_width / 2)}}{headers[3]:>{int((graph_width + 1) / 2)}}" 80 ) 81 82 sorted_intervals: Iterable[Tuple[str, float, float]] = sorted( 83 cls.timers, key=lambda i: i[1] 84 ) 85 graph = "Timing Graph: \n" 86 graph += _hline(full_headers, "U") 87 graph += "│".join(full_headers) + "\n" 88 graph += _hline(full_headers) 89 for name, start, end in sorted_intervals: 90 graph += ( 91 f"{name:<{name_max}}│" 92 f"{_to_time(end - start):<8}│" 93 + _graph_timespan( 94 (start - time_min) / step, (end - time_min) / step, graph_width 95 ) 96 + "\n" 97 ) 98 graph += _hline(full_headers, "L") 99 return graph