forked from WolframResearch/WolframClientForPython
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathdispatch.py
More file actions
165 lines (118 loc) · 5.69 KB
/
Copy pathdispatch.py
File metadata and controls
165 lines (118 loc) · 5.69 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
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
from __future__ import absolute_import, print_function, unicode_literals
import inspect
from wolframclient.utils.functional import flatten
# original idea by Guido in person.
# https://www.artima.com/weblogs/viewpost.jsp?thread=101605
class Dispatch:
"""A method dispatcher class allowing for multiple implementations of a function. Each implementation is associated to a specific input type.
Implementations are registered with the annotation :meth:`~wolframclient.utils.dispatch.Dispatch.dispatch`.
The Dispatch class is callable, it behaves as a function that uses the implementation corresponding to the input parameter.
When a type is a subtype, the type and its parents are checked in the order given by :data:`__mro__` (method resolution order).
*Example:* method :meth:`~wolframclient.utils.dispatch.Dispatch.resolve` applied to an instance of :class:`collections.OrderedDict`,
check for the first implementation to match with :class:`collections.OrderedDict`, then with :class:`dict`, and ultimately to :data:`object`.
Once the mapping is determined, it is cached for later use.
"""
def __init__(self):
self.clear()
def dispatch(self, *args, **opts):
"""Annotate a function and map it to a given set of type(s).
Declare an implementation to use on :data:`bytearray` input::
@dispatcher.dispatch(bytearray)
def my_func(...)
The default implementation is associated with :data:`object`. Set a default::
@dispatcher.dispatch(object)
def my_default_func(...)
A tuple can be used as input to associate more than one type with a function.
Declare a function used for both :data:`bytes` and :data:`bytearray`::
@dispatcher.dispatch((bytes, bytearray))
def my_func(...)
Implementation must be unique. By default, registering the same combination of types will raise an error.
Set `replace_existing` to :data:`True` to update the current mapping.
Or, set `keep_existing` to :data:`True` to ignore duplicate registration and keep the existing mapping.
"""
def register(func):
return self.register(func, *args, **opts)
return register
def update(self, dispatch, **opts):
"""Update current mapping with the one from `dispatch`.
`dispatch` can be a Dispatch instance or a :class:`dict`.
`**opts` are passed to :meth:`~wolframclient.utils.dispatch.Dispatch.register`
"""
if isinstance(dispatch, Dispatch):
dispatchmapping = dispatch.dispatch_dict
elif isinstance(dispatch, dict):
dispatchmapping = dispatch
else:
raise ValueError("%s is not an instance of Dispatch" % dispatch)
for t, function in dispatchmapping.items():
self.register(function, t, **opts)
def validate_types(self, types):
for t in frozenset(flatten(types)):
if not inspect.isclass(t):
raise ValueError("%s is not a class" % t)
yield t
def register(self, function, types=object, keep_existing=False, replace_existing=False):
"""Equivalent to annotation :meth:`~wolframclient.utils.dispatch.Dispatch.dispatch` but as
a function.
"""
if not callable(function):
raise ValueError("Function %s is not callable" % function)
if keep_existing and replace_existing:
raise ValueError(
"Option values keep_existing and replace_existing cannot be both True."
)
self.clear_cache()
for t in self.validate_types(types):
if replace_existing:
self.dispatch_dict[t] = function
elif t in self.dispatch_dict:
if not keep_existing:
raise TypeError("Duplicated registration for input type(s): {}".format(t))
else:
self.dispatch_dict[t] = function
return function
def unregister(self, types=object):
"""Remove implementations associated with types."""
self.clear_cache()
for t in self.validate_types(types):
try:
del self.dispatch_dict[t]
except KeyError:
pass
def clear(self):
"""Reset the dispatcher to its initial state."""
self.dispatch_dict = {}
self.dispatch_dict_cache = {}
def clear_cache(self):
if self.dispatch_dict_cache:
self.dispatch_dict_cache = {}
def resolve(self, arg):
"""Return the implementation better matching the type the argument type."""
for t in arg.__class__.__mro__:
try:
return self.dispatch_dict_cache[t]
except KeyError:
impl = self.dispatch_dict.get(t, None)
if impl:
self.dispatch_dict_cache[t] = impl
return impl
return self.default_function
def default_function(self, *args, **opts):
"""Ultimately called when no type was found."""
raise ValueError("Unable to handle args")
def __call__(self, arg, *args, **opts):
return self.resolve(arg)(arg, *args, **opts)
def as_method(self):
"""Return the dispatch as a class method.
Create a new dispatcher::
dispatch = Dispatcher()
Use the dispatcher as a class method::
class MyClass(object):
myMethod = dispatch.as_method()
Call the class method::
o = MyClass()
o.myMethod(arg, *args, **kwargs)
"""
def method(instance, arg, *args, **opts):
return self.resolve(arg)(instance, arg, *args, **opts)
return method