-
Notifications
You must be signed in to change notification settings - Fork 76
/
__init__.py
180 lines (136 loc) · 4.78 KB
/
__init__.py
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
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
# vi: ts=4 sw=4 sts=4 expandtab
import inspect
import logging
import warnings
from collections import OrderedDict
from .epics_pvs import * # noqa: F401, F403
from .errors import * # noqa: F401, F403
from .paths import make_dir_tree, makedirs # noqa: F401
logger = logging.getLogger(__name__)
def enum(**enums):
"""Create an enum from the keyword arguments"""
return type("Enum", (object,), enums)
class OrderedDefaultDict(OrderedDict):
"""
a combination of defaultdict and OrderedDict
source: http://stackoverflow.com/a/6190500/1221924
"""
def __init__(self, default_factory=None, *a, **kw):
if default_factory is not None and not callable(default_factory):
raise TypeError("first argument must be callable")
super().__init__(*a, **kw)
self.default_factory = default_factory
def __getitem__(self, key):
try:
return super().__getitem__(key)
except KeyError:
return self.__missing__(key)
def __missing__(self, key):
if self.default_factory is None:
raise KeyError(key)
self[key] = value = self.default_factory()
return value
def __reduce__(self):
if self.default_factory is None:
args = tuple()
else:
args = (self.default_factory,)
return type(self), args, None, None, self.items()
def copy(self):
return self.__copy__()
def __copy__(self):
return type(self)(self.default_factory, self)
def __deepcopy__(self, memo):
import copy
return type(self)(self.default_factory, copy.deepcopy(self.items()))
def __repr__(self):
return "%s(%s, %s)" % (
self.__class__.__name__,
self.default_factory,
super().__repr__(),
)
def doc_annotation_forwarder(base_klass):
def wrapper(f):
f_name = getattr(f, "__name__")
base_func = getattr(base_klass, f_name)
base_docs = getattr(base_func, "__doc__")
base_annotation = getattr(base_func, "__annotations__")
f.__doc__ = base_docs
f.__annotations__ = base_annotation
return f
return wrapper
def _filtered_ip_ns():
import IPython
return {
k: v for k, v in IPython.get_ipython().user_ns.items() if not k.startswith("_")
}
def instances_from_namespace(classes, *, ns=None):
"""Get instances of `classes` from the user namespace
Parameters
----------
classes : type, or sequence of types
Passed directly to isinstance(), only instances of these classes
will be returned.
ns : Dict[str, Any], optional
namespace to pull from, defaults to getting the
"""
if ns is None:
ns = _filtered_ip_ns()
return [val for val in ns.values() if isinstance(val, classes)]
def ducks_from_namespace(attrs, *, ns=None):
"""Get instances that have all of attributes.
"Ducks" is a reference to "duck-typing." If it looks like a duck....
Parameters
----------
attr : Union[str, Iterable[str]]
name of attribute or list of names
"""
if isinstance(attrs, str):
attrs = [attrs]
if ns is None:
ns = _filtered_ip_ns()
return [val for val in ns.values() if all(hasattr(val, attr) for attr in attrs)]
def underscores_to_camel_case(underscores):
"Convert abc_def_ghi to abcDefGhi"
if "_" in underscores and underscores.strip("_") != "":
return "".join(s.capitalize() for s in underscores.split("_"))
return underscores.capitalize()
def getattrs(obj, gen):
"For each item in the generator `gen`, yields (attr, getattr(obj, attr))"
for attr in gen:
yield attr, getattr(obj, attr)
class DO_NOT_USE:
"sentinel value"
...
def adapt_old_callback_signature(callback):
"""
If callback has signature callback(), wrap in signature callback(status).
Parameters
----------
callback: callable
Expected signature ``callback(status)`` or ``callback()``
Returns
-------
callback: callable
Signature ``callback(status)``
"""
# Handle callback with signature callback() for back-compat.
sig = inspect.signature(callback)
try:
# Does this callback accept one positional argument?
sig.bind(None)
except TypeError:
warnings.warn(
"The signature of a Status callback is now expected to "
"be cb(status). The signature cb() is "
"supported, but support will be removed in a future release "
"of ophyd.",
DeprecationWarning,
stacklevel=3,
)
raw_callback = callback
def callback(status):
# Do nothing with status because the user-provided callback cannot
# accept it.
raw_callback()
return callback