1#!/usr/bin/env python3
2"""
3
4See EOF for license/metadata/notes as applicable
5"""
6
7##-- builtin imports
8from __future__ import annotations
9
10import copy
11import datetime
12import enum
13import functools as ftz
14import itertools as itz
15import logging as logmod
16import re
17import time
18import weakref
19from uuid import UUID, uuid1
20
21##-- end builtin imports
22
23from time import sleep
24import timeit
25from random import random
26import jgdv
27from jgdv.decorators import MonotonicDec
28
29# ##-- types
30# isort: off
31# General
32import abc
33import collections.abc
34import typing
35import types
36from typing import cast, assert_type, assert_never
37from typing import Generic, NewType, Never
38from typing import no_type_check, final, override, overload
39# Protocols and Interfaces:
40from typing import Protocol, runtime_checkable
41# isort: on
42# ##-- end types
43
44# ##-- type checking
45# isort: off
46if typing.TYPE_CHECKING:
47 import pathlib as pl
48 from typing import Final, ClassVar, Any, Self
49 from typing import Literal, LiteralString
50 from typing import TypeGuard
51 from collections.abc import Iterable, Iterator, Callable, Generator
52 from collections.abc import Sequence, Mapping, MutableMapping, Hashable
53
54 from jgdv import Maybe, Traceback, Func, Method
55 from logging import Logger
56## isort: on
57# ##-- end type checking
58
59##-- logging
60logging = logmod.getLogger(__name__)
61##-- end logging
62
63autorange_fmt : Final[str] = "%-*10s : %-*5d calls took: %-*8.2f seconds"
64result_fmt : Final[str] = "Attempt %-*5d : %-*8.2f seconds"
65block_fmt : Final[str] = "%-*10s : %-*8.2f seconds"
66once_fmt : Final[str] = "%-*10s : %-*8.2f seconds"
67
68##--|
[docs]
69class TimeCtx:
70 """ Utility Class to time code execution. """
71 _logger : Maybe[Logger]
72 level : int
73 group : str
74 total : int
75 total_ms : float
76 total_s : float
77 _start : int
78 _stop : int
79
80 def __init__(self, *, logger:Maybe[Logger|Literal[False]]=None, level:int=logmod.INFO, group:Maybe[str]=None) -> None:
81 self.level = level
82 self.group = f"{group}::" if group else ""
83 self._start = 0
84 self._stop = 0
85 self.total = 0
86 match logger:
87 case None:
88 self._logger = logging
89 case False:
90 self._logger = None
91 case logmod.Logger() as l:
92 self._logger = l
93
94 def __enter__(self) -> Self:
95 self._start = time.monotonic_ns()
96 return self
97
98 def __exit__(self, etype:Maybe[type], err:Maybe[Exception], tb:Maybe[Traceback]) -> bool:
99 self._stop = time.monotonic_ns()
100 self.total = (self._stop - self._start)
101 self.total_ms = self.total * 0.0001
102 self.total_s = self.total / 1_000_000_000
103 return False
104
[docs]
105 def _set_name(self, name:Maybe[str]) -> None:
106 match name:
107 case None:
108 pass
109 case str() as s:
110 self.current_name = self.group + s
111
[docs]
112 def msg(self, msg:str, *args:Any) -> None: # noqa: ANN401
113 if self._logger is None:
114 return
115
116 self._logger.log(self.level, msg, *args)
117
118
[docs]
119class TimeDec(MonotonicDec):
120 """ Decorate a callable to track its timing """
121 _logger : Maybe[Logger]
122 _cache : Maybe[pl.Path]
123
124 def __init__(self, *, cache:Maybe[pl.Path]=None, logger:Maybe[Logger|Literal[False]]=None, level:Maybe[int]=None, **kwargs:Any) -> None: # noqa: ANN401
125 kwargs.setdefault("mark", "_timetrack_mark")
126 kwargs.setdefault("data", "_timetrack_data")
127 super().__init__([], **kwargs)
128 self._level = level
129 self._cache = cache
130 match logger:
131 case logmod.Logger() as l:
132 self._logger = l
133 case None:
134 self._logger = logging
135 case False:
136 self._logger = None
137
138
[docs]
139 @override
140 def _wrap_fn_h[**I, O](self, fn:Func[I, O]) -> Func[I, O]:
141 logger, level = self._logger, self._level
142
143 def track_time_wrapper(*args:I.args, **kwargs:I.kwargs) -> O:
144 with TimeCtx(logger=logger, level=level or logmod.INFO) as timer:
145 result = fn(*args, **kwargs)
146
147 timer.msg("Timed: %s took %s seconds", fn.__qualname__, timer.total_s)
148 if self._cache:
149 now = datetime.datetime.now().strftime("%Y-%m-%d:%H-%M")
150 line = f"{now} : {fn.__qualname__} : {timer.total_s}\n"
151 with open(self._cache, "a") as f:
152 f.write(line)
153
154 return result
155
156 return track_time_wrapper
157
[docs]
158 @override
159 def _wrap_method_h[**I, O](self, fn:Func[I, O]) -> Func[I, O]:
160 return self._wrap_fn_h(fn)