forked from acts-project/acts
-
Notifications
You must be signed in to change notification settings - Fork 6
Expand file tree
/
Copy path_adapter.py
More file actions
95 lines (77 loc) · 2.92 KB
/
_adapter.py
File metadata and controls
95 lines (77 loc) · 2.92 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
import inspect
import functools
from typing import Optional, Callable, Dict, Any
from pathlib import Path
def _make_config_adapter(fn):
@functools.wraps(fn)
def wrapped(self, *args, **kwargs):
if len(args) > 0:
maybe_config = args[0]
if isinstance(maybe_config, inspect.unwrap(type(self).Config)):
# is already config, nothing to do here
fn(self, maybe_config, *args[1:], **kwargs)
return
if "config" in kwargs:
config = kwargs.pop("config")
fn(self, config, *args, **kwargs)
return
cfg = type(self).Config()
_kwargs = {}
for k, v in kwargs.items():
if isinstance(v, Path):
v = str(v)
if hasattr(cfg, k):
try:
setattr(cfg, k, v)
except TypeError as e:
raise RuntimeError(
"{}: Failed to set {}={}".format(type(cfg), k, v)
) from e
else:
_kwargs[k] = v
try:
fn(self, cfg, *args, **_kwargs)
except TypeError as e:
import textwrap
print("-" * 80)
print("Patched config constructor failed for", type(self))
message = (
"This is most likely because one of the following kwargs "
"could not be assigned to the Config object, and the constructor "
"did not accept it as an additional argument:"
)
print("\n".join(textwrap.wrap(message, width=80)))
print("->", ", ".join(_kwargs.keys()))
members = inspect.getmembers(type(cfg), lambda a: not inspect.isroutine(a))
members = [m for m, _ in members if not m.startswith("_")]
print(type(cfg), "has the following properties:\n->", ", ".join(members))
print("-" * 80)
raise e
return wrapped
def _make_config_constructor(
cls, proc: Optional[Callable[[Dict[str, Any]], Dict[str, Any]]] = None
):
fn = cls.__init__
@functools.wraps(fn)
def wrapped(self, *args, **kwargs):
_kwargs = {}
for k in list(kwargs.keys()):
if hasattr(cls, k):
_kwargs[k] = kwargs.pop(k)
fn(self, *args, **kwargs)
if proc is not None:
_kwargs = proc(_kwargs)
for k, v in _kwargs.items():
setattr(self, k, v)
return wrapped
def _patchKwargsConstructor(
cls, proc: Optional[Callable[[Dict[str, Any]], Dict[str, Any]]] = None
):
cls.__init__ = _make_config_constructor(cls, proc)
def _patch_config(m):
for name, cls in inspect.getmembers(m, inspect.isclass):
if name == "Config":
_patchKwargsConstructor(cls)
if hasattr(cls, "Config"):
cls.__init__ = _make_config_adapter(cls.__init__)
_patchKwargsConstructor(cls.Config)