#3262

Build # #3262

Build Type

push

coveralls-python

Committed by GitHub

Commit Message

<a href="https://github.com/mosquito/aiomisc/commit/<a class=hub.com/mosquito/aiomisc/commit/ae7bcfdf2e24800292bf13854667303a1da8f498">ae7bcfdf2<a href="https://github.com/mosquito/aiomisc/commit/ae7bcfdf2e24800292bf13854667303a1da8f498">">Merge </a><a class="double-link" href="https://github.com/mosquito/aiomisc/commit/<a class="double-link" href="https://github.com/mosquito/aiomisc/commit/da4bcedcc499166867a7575455d89c3c6c5c4977">da4bcedcc</a>">da4bcedcc</a><a href="https://github.com/mosquito/aiomisc/commit/ae7bcfdf2e24800292bf13854667303a1da8f498"> into bfc83dd07">bfc83dd07</a>

Run Details

564 of 726 branches covered (77.69%)

Branch coverage included in aggregate %.

7 of 7 new or added lines in 3 files covered. (100.0%)

2441 of 2791 relevant lines covered (87.46%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.91

/aiomisc/counters.py

import typing as t
from collections import Counter
from weakref import WeakSet


class Metric:
    def __init__(
        self, name: str,
        counter: t.MutableMapping[str, t.Union[float, int]],
        default: t.Union[float, int] = 0,
    ):
        self.name: str = name
        self.counter = counter
        self.counter[name] = default

    def __get__(self) -> t.Union[float, int]:
        return self.counter[self.name]

    def __set__(self, value: t.Union[float, int]) -> None:
        self.counter[self.name] = value

    def __iadd__(self, value: t.Union[float, int]) -> "Metric":
        self.counter[self.name] += value
        return self

    def __isub__(self, value: t.Union[float, int]) -> "Metric":
        self.counter[self.name] -= value
        return self

    def __eq__(self, other: t.Any) -> bool:
        return self.counter[self.name] == other

    def __hash__(self) -> int:
        return hash(self.counter[self.name])


class AbstractStatistic:
    __metrics__: t.FrozenSet[str]
    __instances__: t.MutableSet["AbstractStatistic"]
    _counter: t.MutableMapping[str, t.Union[float, int]]
    name: t.Optional[str]


CLASS_STORE: t.Set[t.Type[AbstractStatistic]] = set()


class MetaStatistic(type):
    def __new__(
        mcs, name: str,
        bases: t.Tuple[type, ...],
        dct: t.Dict[str, t.Any],
    ) -> t.Any:

        # noinspection PyTypeChecker
        klass: t.Type[AbstractStatistic] = super().__new__(
            mcs, name, bases, dct,
        )   # type: ignore

        metrics = set()

        for base_class in bases:
            if not issubclass(base_class, AbstractStatistic):
                continue

            if not hasattr(base_class, "__annotations__"):
                continue

            for prop, kind in base_class.__annotations__.items():
                if kind not in (int, float):
                    continue

                if prop.startswith("_"):
                    continue

                metrics.add(prop)

        for prop, kind in klass.__annotations__.items():
            if kind not in (int, float):
                continue

            metrics.add(prop)

        klass.__metrics__ = frozenset(metrics)

        if klass.__metrics__:
            klass.__instances__ = WeakSet()
            CLASS_STORE.add(klass)

        return klass


class Statistic(AbstractStatistic, metaclass=MetaStatistic):
    def __init__(self, name: t.Optional[str] = None) -> None:
        self._counter = Counter()   # type: ignore
        self.name = name

        for prop in self.__metrics__:
            setattr(self, prop, Metric(prop, self._counter))

        self.__instances__.add(self)


class StatisticResult(t.NamedTuple):
    kind: t.Type[AbstractStatistic]
    name: t.Optional[str]
    metric: str
    value: t.Union[int, float]


# noinspection PyProtectedMember
def get_statistics(
    *kind: t.Type[Statistic]
) -> t.Generator[t.Any, t.Tuple[Statistic, str, int], None]:
    for klass in CLASS_STORE:
        if kind and not issubclass(klass, kind):
            continue

        for instance in klass.__instances__:
            for metric, value in instance._counter.items():
                yield StatisticResult(
                    kind=klass,
                    name=instance.name,
                    metric=metric,
                    value=value,
                )

1	import typing as t	1✔
2	from collections import Counter	1✔
3	from weakref import WeakSet	1✔
4
5
6	class Metric:	1✔
7	def __init__(	1✔
8	self, name: str,
9	counter: t.MutableMapping[str, t.Union[float, int]],
10	default: t.Union[float, int] = 0,
11	):
12	self.name: str = name	1✔
13	self.counter = counter	1✔
14	self.counter[name] = default	1✔
15
16	def __get__(self) -> t.Union[float, int]:	1✔
17	return self.counter[self.name]	×
18
19	def __set__(self, value: t.Union[float, int]) -> None:	1✔
20	self.counter[self.name] = value	×
21
22	def __iadd__(self, value: t.Union[float, int]) -> "Metric":	1✔
23	self.counter[self.name] += value	1✔
24	return self	1✔
25
26	def __isub__(self, value: t.Union[float, int]) -> "Metric":	1✔
27	self.counter[self.name] -= value	1✔
28	return self	1✔
29
30	def __eq__(self, other: t.Any) -> bool:	1✔
31	return self.counter[self.name] == other	1✔
32
33	def __hash__(self) -> int:	1✔
34	return hash(self.counter[self.name])	×
35
36
37	class AbstractStatistic:	1✔
38	__metrics__: t.FrozenSet[str]	1✔
39	__instances__: t.MutableSet["AbstractStatistic"]	1✔
40	_counter: t.MutableMapping[str, t.Union[float, int]]	1✔
41	name: t.Optional[str]	1✔
42
43
44	CLASS_STORE: t.Set[t.Type[AbstractStatistic]] = set()	1✔
45
46
47	class MetaStatistic(type):	1✔
48	def __new__(	1✔
49	mcs, name: str,
50	bases: t.Tuple[type, ...],
51	dct: t.Dict[str, t.Any],
52	) -> t.Any:
53
54	# noinspection PyTypeChecker
55	klass: t.Type[AbstractStatistic] = super().__new__(	1✔
56	mcs, name, bases, dct,
57	) # type: ignore
58
59	metrics = set()	1✔
60
61	for base_class in bases:	1✔
62	if not issubclass(base_class, AbstractStatistic):	1!
63	continue	×
64
65	if not hasattr(base_class, "__annotations__"):	1!
66	continue	×
67
68	for prop, kind in base_class.__annotations__.items():	1✔
69	if kind not in (int, float):	1✔
70	continue	1✔
71
72	if prop.startswith("_"):	1!
73	continue	×
74
75	metrics.add(prop)	1✔
76
77	for prop, kind in klass.__annotations__.items():	1✔
78	if kind not in (int, float):	1✔
79	continue	1✔
80
81	metrics.add(prop)	1✔
82
83	klass.__metrics__ = frozenset(metrics)	1✔
84
85	if klass.__metrics__:	1✔
86	klass.__instances__ = WeakSet()	1✔
87	CLASS_STORE.add(klass)	1✔
88
89	return klass	1✔
90
91
92	class Statistic(AbstractStatistic, metaclass=MetaStatistic):	1✔
93	def __init__(self, name: t.Optional[str] = None) -> None:	1✔
94	self._counter = Counter() # type: ignore	1✔
95	self.name = name	1✔
96
97	for prop in self.__metrics__:	1✔
98	setattr(self, prop, Metric(prop, self._counter))	1✔
99
100	self.__instances__.add(self)	1✔
101
102
103	class StatisticResult(t.NamedTuple):	1✔
104	kind: t.Type[AbstractStatistic]	1✔
105	name: t.Optional[str]	1✔
106	metric: str	1✔
107	value: t.Union[int, float]	1✔
108
109
110	# noinspection PyProtectedMember
111	def get_statistics(	1✔
112	*kind: t.Type[Statistic]
113	) -> t.Generator[t.Any, t.Tuple[Statistic, str, int], None]:
114	for klass in CLASS_STORE:	1✔
115	if kind and not issubclass(klass, kind):	1✔
116	continue	1✔
117
118	for instance in klass.__instances__:	1✔
119	for metric, value in instance._counter.items():	1✔
120	yield StatisticResult(	1✔
121	kind=klass,
122	name=instance.name,
123	metric=metric,
124	value=value,
125	)

mosquito / aiomisc / #3262

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous