6534953321

Committed 16 Oct 2023 02:19PM UTC coverage: 35.674% (-22.7%) from 58.421%

Build # 6534953321

Build Type

push

github

Committed by

feihoo87

Commit Message

fix Coveralls

Run Details

5913 of 16575 relevant lines covered (35.67%)

3.21 hits per line

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

57.95

/waveforms/baseconfig.py

from __future__ import annotations

import copy
import json
import random
from pathlib import Path
from typing import Any, Optional, Union

from .dicttree import flattenDict as _flattenDict
from .dicttree import flattenDictIter as _flattenDictIter
from .dicttree import foldDict as _foldDict
from .dicttree import update_tree as _update


def _query(q, dct):
    return {
        k.removeprefix(q + '.'): v
        for k, v in dct.items() if k.startswith(q + '.')
    }


def randomStr(n):
    s = ('abcdefghijklmnopqrstuvwxyz'
         'ABCDEFGHIJKLMNOPQRSTUVWXYZ'
         '0123456789')
    return ''.join(random.choices(s, k=n))


def mixin(o: object, *traits: type) -> object:
    bases = (o.__class__, *traits)
    name = '_'.join([cls.__name__ for cls in bases])
    name = '_'.join([name, randomStr(6)])
    cls = type(name, bases, {})
    o.__class__ = cls
    return o


class TraitMeta(type):
    _traits = {}

    def __new__(cls, name, bases, namespace):
        cls = super().__new__(cls, name, bases, namespace)
        if name != 'Trait':
            TraitMeta._traits[name] = cls
        return cls


class Trait(metaclass=TraitMeta):
    pass


def queryKey(q, dct, prefix=None):
    if prefix is None:
        prefix = []

    keys = q.split('.', maxsplit=1)

    if not isinstance(dct, dict):
        k = '.'.join(prefix)
        raise KeyError(
            f"Query {k}.{q} error, type '{k}' is {type(dct)}, not dict.")
    try:
        sub = dct[keys[0]]
    except KeyError:
        k = '.'.join([*prefix, keys[0]])
        raise KeyError(
            f"Query {'.'.join([*prefix, q])} error, key '{k}' not found.")

    if len(keys) == 1:
        return sub

    else:
        return queryKey(keys[1], sub, [*prefix, keys[0]])


def query(q, dct, prefix=None):
    if isinstance(q, str):
        return queryKey(q, dct, prefix)
    elif isinstance(q, list):
        return [query(sub_q, dct, prefix) for sub_q in q]
    elif isinstance(q, tuple):
        return tuple([query(sub_q, dct, prefix) for sub_q in q])
    elif isinstance(q, set):
        return {sub_q: query(sub_q, dct, prefix) for sub_q in q}
    elif isinstance(q, dict):
        if prefix is None:
            prefix = []
        ret = {}
        for k, sub_q in q.items():
            if sub_q is None:
                ret[k] = queryKey(k, dct, prefix)
            else:
                ret[k] = query(sub_q, queryKey(k, dct, prefix), [*prefix, k])
        return ret
    else:
        raise TypeError


def setKey(q, value, dct, prefix=None):
    if prefix is None:
        prefix = []

    keys = q.split('.', maxsplit=1)

    if len(keys) == 1:
        if keys[0] in dct and isinstance(dct[keys[0]],
                                         dict) and not isinstance(value, dict):
            k = '.'.join([*prefix, keys[0]])
            raise ValueError(f'try to set a dict {k} to {type(value)}')
        else:
            dct[keys[0]] = value
    else:
        if keys[0] in dct and isinstance(dct[keys[0]], dict):
            sub = dct[keys[0]]
        elif keys[0] in dct and not isinstance(dct[keys[0]], dict):
            k = '.'.join([*prefix, keys[0]])
            raise ValueError(f'try to set a dict {k} to {type(value)}')
        else:
            sub = {}
            dct[keys[0]] = sub
        setKey(keys[1], sub, [*prefix, keys[0]])


class ConfigSection(dict):
    def __init__(self, cfg: BaseConfig, key: str):
        self._cfg_ = cfg
        self._key_ = key

    def __setitem__(self, key: str, value: Any) -> None:
        if self._cfg_ is None:
            self._modified_ = True
        else:
            self._cfg_._modified_ = True
        if isinstance(value, dict) and not isinstance(value, ConfigSection):
            key, *traits = key.split(':')
            if self._cfg_ is None:
                cfg = self
                k = key
            else:
                cfg = self._cfg_
                k = '.'.join([self._key_, key])
            d = ConfigSection(cfg, k)
            d.update(value)
            value = d
        elif isinstance(value, ConfigSection):
            value.__class__ = ConfigSection
        super().__setitem__(key, value)

    def __getitem__(self, key: str) -> ValueType:
        key, *traits = key.split(':')
        if self._cfg_ is not None:
            d = self._cfg_.query(self._key_)
            if self is not d:
                self.update(d)
        ret = super().__getitem__(key)
        if isinstance(ret, ConfigSection) and len(traits) > 0:
            traits = [
                TraitMeta._traits[n] for n in traits if n in TraitMeta._traits
            ]
            mixin(ret, *traits)
        return ret

    def __delitem__(self, key: str) -> None:
        if self._cfg_ is None:
            self._modified_ = True
        else:
            self._cfg_._modified_ = True
        return super().__delitem__(key)

    def __delattr__(self, name: str) -> None:
        if name in self:
            return self.__delitem__(name)
        else:
            return super().__delattr__(name)

    def __getattr__(self, name: str) -> Any:
        try:
            return self.__getattribute__(name)
        except:
            try:
                return self.__getitem__(name)
            except:
                raise AttributeError(f'Not Find Attr: {name}')

    def __setattr__(self, name: str, value: Any) -> None:
        if name in self:
            self.__setitem__(name, value)
        else:
            super().__setattr__(name, value)

    def __deepcopy__(self, memo):
        dct = {k: copy.deepcopy(v) for k, v in self.items()}
        return dct

    def query(self, q: Union[str, set, tuple, list,
                             dict]) -> Union[dict, ValueType]:
        if self._key_ is None:
            prefix = []
        else:
            prefix = self._key_.split('.')
        return query(q, self, prefix=prefix)

    def set(self, q, value):
        if self._key_ is None:
            prefix = []
        else:
            prefix = self._key_.split('.')
        setKey(q, value, self, prefix=prefix)

    def update(self, other):
        _update(self, other)


ValueType = Union[str, int, float, list, ConfigSection]


class BaseConfig(ConfigSection):
    def __init__(self,
                 path: Optional[Union[str, Path]] = None,
                 backup: bool = False):
        super().__init__(None, None)
        if isinstance(path, str):
            path = Path(path)
        self._path_ = path
        self._backup_ = backup
        self._modified_ = False

        if self._path_ is None:
            return
        if self._path_.exists():
            self.reload()
            if '__version__' not in self:
                self['__version__'] = 1
        else:
            self._path_.parent.mkdir(parents=True, exist_ok=True)
            self['__version__'] = 1
            self._modified_ = True
            self.commit()

    def commit(self):
        if not self._modified_ or self._path_ is None:
            return
        if self._backup_ and self._path_.exists():
            v = self['__version__']
            bk = self._path_.with_stem(self._path_.stem + f"_{v}")
            self._path_.rename(bk)
        with self._path_.open('w') as f:
            self['__version__'] = self['__version__'] + 1
            json.dump(self, f, indent=4)
            self._modified_ = False

    def rollback(self):
        if not self._modified_ or self._path_ is None:
            return
        self.reload()

    def reload(self):
        with self._path_.open('r') as f:
            dct = json.load(f)
            self.clear()
            self.update(dct)
            self._modified_ = False

    @classmethod
    def fromdict(cls, d: dict) -> BaseConfig:
        ret = cls()
        ret.update(d)
        return ret

1	from __future__ import annotations	9✔
2
3	import copy	9✔
4	import json	9✔
5	import random	9✔
6	from pathlib import Path	9✔
7	from typing import Any, Optional, Union	9✔
8
9	from .dicttree import flattenDict as _flattenDict	9✔
10	from .dicttree import flattenDictIter as _flattenDictIter	9✔
11	from .dicttree import foldDict as _foldDict	9✔
12	from .dicttree import update_tree as _update	9✔
13
14
15	def _query(q, dct):	9✔
16	return {	×
17	k.removeprefix(q + '.'): v
18	for k, v in dct.items() if k.startswith(q + '.')
19	}
20
21
22	def randomStr(n):	9✔
23	s = ('abcdefghijklmnopqrstuvwxyz'	9✔
24	'ABCDEFGHIJKLMNOPQRSTUVWXYZ'
25	'0123456789')
26	return ''.join(random.choices(s, k=n))	9✔
27
28
29	def mixin(o: object, *traits: type) -> object:	9✔
30	bases = (o.__class__, *traits)	9✔
31	name = '_'.join([cls.__name__ for cls in bases])	9✔
32	name = '_'.join([name, randomStr(6)])	9✔
33	cls = type(name, bases, {})	9✔
34	o.__class__ = cls	9✔
35	return o	9✔
36
37
38	class TraitMeta(type):	9✔
39	_traits = {}	9✔
40
41	def __new__(cls, name, bases, namespace):	9✔
42	cls = super().__new__(cls, name, bases, namespace)	9✔
43	if name != 'Trait':	9✔
44	TraitMeta._traits[name] = cls	9✔
45	return cls	9✔
46
47
48	class Trait(metaclass=TraitMeta):	9✔
49	pass	9✔
50
51
52	def queryKey(q, dct, prefix=None):	9✔
53	if prefix is None:	9✔
54	prefix = []	×
55
56	keys = q.split('.', maxsplit=1)	9✔
57
58	if not isinstance(dct, dict):	9✔
59	k = '.'.join(prefix)	×
60	raise KeyError(	×
61	f"Query {k}.{q} error, type '{k}' is {type(dct)}, not dict.")
62	try:	9✔
63	sub = dct[keys[0]]	9✔
64	except KeyError:	×
65	k = '.'.join([*prefix, keys[0]])	×
66	raise KeyError(	×
67	f"Query {'.'.join([*prefix, q])} error, key '{k}' not found.")
68
69	if len(keys) == 1:	9✔
70	return sub	9✔
71
72	else:
73	return queryKey(keys[1], sub, [*prefix, keys[0]])	9✔
74
75
76	def query(q, dct, prefix=None):	9✔
77	if isinstance(q, str):	9✔
78	return queryKey(q, dct, prefix)	9✔
79	elif isinstance(q, list):	×
80	return [query(sub_q, dct, prefix) for sub_q in q]	×
81	elif isinstance(q, tuple):	×
82	return tuple([query(sub_q, dct, prefix) for sub_q in q])	×
83	elif isinstance(q, set):	×
84	return {sub_q: query(sub_q, dct, prefix) for sub_q in q}	×
85	elif isinstance(q, dict):	×
86	if prefix is None:	×
87	prefix = []	×
88	ret = {}	×
89	for k, sub_q in q.items():	×
90	if sub_q is None:	×
91	ret[k] = queryKey(k, dct, prefix)	×
92	else:
93	ret[k] = query(sub_q, queryKey(k, dct, prefix), [*prefix, k])	×
94	return ret	×
95	else:
96	raise TypeError	×
97
98
99	def setKey(q, value, dct, prefix=None):	9✔
100	if prefix is None:	×
101	prefix = []	×
102
103	keys = q.split('.', maxsplit=1)	×
104
105	if len(keys) == 1:	×
106	if keys[0] in dct and isinstance(dct[keys[0]],	×
107	dict) and not isinstance(value, dict):
108	k = '.'.join([*prefix, keys[0]])	×
109	raise ValueError(f'try to set a dict {k} to {type(value)}')	×
110	else:
111	dct[keys[0]] = value	×
112	else:
113	if keys[0] in dct and isinstance(dct[keys[0]], dict):	×
114	sub = dct[keys[0]]	×
115	elif keys[0] in dct and not isinstance(dct[keys[0]], dict):	×
116	k = '.'.join([*prefix, keys[0]])	×
117	raise ValueError(f'try to set a dict {k} to {type(value)}')	×
118	else:
119	sub = {}	×
120	dct[keys[0]] = sub	×
121	setKey(keys[1], sub, [*prefix, keys[0]])	×
122
123
124	class ConfigSection(dict):	9✔
125	def __init__(self, cfg: BaseConfig, key: str):	9✔
126	self._cfg_ = cfg	9✔
127	self._key_ = key	9✔
128
129	def __setitem__(self, key: str, value: Any) -> None:	9✔
130	if self._cfg_ is None:	9✔
131	self._modified_ = True	9✔
132	else:
133	self._cfg_._modified_ = True	9✔
134	if isinstance(value, dict) and not isinstance(value, ConfigSection):	9✔
135	key, *traits = key.split(':')	9✔
136	if self._cfg_ is None:	9✔
137	cfg = self	9✔
138	k = key	9✔
139	else:
140	cfg = self._cfg_	9✔
141	k = '.'.join([self._key_, key])	9✔
142	d = ConfigSection(cfg, k)	9✔
143	d.update(value)	9✔
144	value = d	9✔
145	elif isinstance(value, ConfigSection):	9✔
146	value.__class__ = ConfigSection	×
147	super().__setitem__(key, value)	9✔
148
149	def __getitem__(self, key: str) -> ValueType:	9✔
150	key, *traits = key.split(':')	9✔
151	if self._cfg_ is not None:	9✔
152	d = self._cfg_.query(self._key_)	9✔
153	if self is not d:	9✔
154	self.update(d)	×
155	ret = super().__getitem__(key)	9✔
156	if isinstance(ret, ConfigSection) and len(traits) > 0:	9✔
157	traits = [	9✔
158	TraitMeta._traits[n] for n in traits if n in TraitMeta._traits
159	]
160	mixin(ret, *traits)	9✔
161	return ret	9✔
162
163	def __delitem__(self, key: str) -> None:	9✔
164	if self._cfg_ is None:	×
165	self._modified_ = True	×
166	else:
167	self._cfg_._modified_ = True	×
168	return super().__delitem__(key)	×
169
170	def __delattr__(self, name: str) -> None:	9✔
171	if name in self:	×
172	return self.__delitem__(name)	×
173	else:
174	return super().__delattr__(name)	×
175
176	def __getattr__(self, name: str) -> Any:	9✔
177	try:	9✔
178	return self.__getattribute__(name)	9✔
179	except:	9✔
180	try:	9✔
181	return self.__getitem__(name)	9✔
182	except:	×
183	raise AttributeError(f'Not Find Attr: {name}')	×
184
185	def __setattr__(self, name: str, value: Any) -> None:	9✔
186	if name in self:	9✔
187	self.__setitem__(name, value)	×
188	else:
189	super().__setattr__(name, value)	9✔
190
191	def __deepcopy__(self, memo):	9✔
192	dct = {k: copy.deepcopy(v) for k, v in self.items()}	9✔
193	return dct	9✔
194
195	def query(self, q: Union[str, set, tuple, list,	9✔
196	dict]) -> Union[dict, ValueType]:
197	if self._key_ is None:	9✔
198	prefix = []	9✔
199	else:
200	prefix = self._key_.split('.')	9✔
201	return query(q, self, prefix=prefix)	9✔
202
203	def set(self, q, value):	9✔
204	if self._key_ is None:	×
205	prefix = []	×
206	else:
207	prefix = self._key_.split('.')	×
208	setKey(q, value, self, prefix=prefix)	×
209
210	def update(self, other):	9✔
211	_update(self, other)	9✔
212
213
214	ValueType = Union[str, int, float, list, ConfigSection]	9✔
215
216
217	class BaseConfig(ConfigSection):	9✔
218	def __init__(self,	9✔
219	path: Optional[Union[str, Path]] = None,
220	backup: bool = False):
221	super().__init__(None, None)	9✔
222	if isinstance(path, str):	9✔
223	path = Path(path)	×
224	self._path_ = path	9✔
225	self._backup_ = backup	9✔
226	self._modified_ = False	9✔
227
228	if self._path_ is None:	9✔
229	return	9✔
230	if self._path_.exists():	×
231	self.reload()	×
232	if '__version__' not in self:	×
233	self['__version__'] = 1	×
234	else:
235	self._path_.parent.mkdir(parents=True, exist_ok=True)	×
236	self['__version__'] = 1	×
237	self._modified_ = True	×
238	self.commit()	×
239
240	def commit(self):	9✔
241	if not self._modified_ or self._path_ is None:	×
242	return	×
243	if self._backup_ and self._path_.exists():	×
244	v = self['__version__']	×
245	bk = self._path_.with_stem(self._path_.stem + f"_{v}")	×
246	self._path_.rename(bk)	×
247	with self._path_.open('w') as f:	×
248	self['__version__'] = self['__version__'] + 1	×
249	json.dump(self, f, indent=4)	×
250	self._modified_ = False	×
251
252	def rollback(self):	9✔
253	if not self._modified_ or self._path_ is None:	×
254	return	×
255	self.reload()	×
256
257	def reload(self):	9✔
258	with self._path_.open('r') as f:	×
259	dct = json.load(f)	×
260	self.clear()	×
261	self.update(dct)	×
262	self._modified_ = False	×
263
264	@classmethod	9✔
265	def fromdict(cls, d: dict) -> BaseConfig:	9✔
266	ret = cls()	9✔
267	ret.update(d)	9✔
268	return ret	9✔

feihoo87 / waveforms / 6534953321

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