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
Timer(logger: logging.Logger, name: str)
17    def __init__(self, logger: Logger, name: str) -> None:
18        self.logger = logger
19        self.name = name
@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