Coverage for /usr/lib/python3/dist-packages/sympy/multipledispatch/dispatcher.py: 49%

1from __future__ import annotations 

2 

3from warnings import warn 

4import inspect 

5from .conflict import ordering, ambiguities, super_signature, AmbiguityWarning 

6from .utils import expand_tuples 

7import itertools as itl 

8 

9 

10class MDNotImplementedError(NotImplementedError): 

11 """ A NotImplementedError for multiple dispatch """ 

12 

13 

14### Functions for on_ambiguity 

15 

16def ambiguity_warn(dispatcher, ambiguities): 

17 """ Raise warning when ambiguity is detected 

18 

19 Parameters 

20 ---------- 

21 dispatcher : Dispatcher 

22 The dispatcher on which the ambiguity was detected 

23 ambiguities : set 

24 Set of type signature pairs that are ambiguous within this dispatcher 

25 

26 See Also: 

27 Dispatcher.add 

28 warning_text 

29 """ 

30 warn(warning_text(dispatcher.name, ambiguities), AmbiguityWarning) 

31 

32 

33class RaiseNotImplementedError: 

34 """Raise ``NotImplementedError`` when called.""" 

35 

36 def __init__(self, dispatcher): 

37 self.dispatcher = dispatcher 

38 

39 def __call__(self, *args, **kwargs): 

40 types = tuple(type(a) for a in args) 

41 raise NotImplementedError( 

42 "Ambiguous signature for %s: <%s>" % ( 

43 self.dispatcher.name, str_signature(types) 

44 )) 

45 

46def ambiguity_register_error_ignore_dup(dispatcher, ambiguities): 

47 """ 

48 If super signature for ambiguous types is duplicate types, ignore it. 

49 Else, register instance of ``RaiseNotImplementedError`` for ambiguous types. 

50 

51 Parameters 

52 ---------- 

53 dispatcher : Dispatcher 

54 The dispatcher on which the ambiguity was detected 

55 ambiguities : set 

56 Set of type signature pairs that are ambiguous within this dispatcher 

57 

58 See Also: 

59 Dispatcher.add 

60 ambiguity_warn 

61 """ 

62 for amb in ambiguities: 

63 signature = tuple(super_signature(amb)) 

64 if len(set(signature)) == 1: 

65 continue 

66 dispatcher.add( 

67 signature, RaiseNotImplementedError(dispatcher), 

68 on_ambiguity=ambiguity_register_error_ignore_dup 

69 ) 

70 

71### 

72 

73 

74_unresolved_dispatchers: set[Dispatcher] = set() 

75_resolve = [True] 

76 

77 

78def halt_ordering(): 

79 _resolve[0] = False 

80 

81 

82def restart_ordering(on_ambiguity=ambiguity_warn): 

83 _resolve[0] = True 

84 while _unresolved_dispatchers: 

85 dispatcher = _unresolved_dispatchers.pop() 

86 dispatcher.reorder(on_ambiguity=on_ambiguity) 

87 

88 

89class Dispatcher: 

90 """ Dispatch methods based on type signature 

91 

92 Use ``dispatch`` to add implementations 

93 

94 Examples 

95 -------- 

96 

97 >>> from sympy.multipledispatch import dispatch 

98 >>> @dispatch(int) 

99 ... def f(x): 

100 ... return x + 1 

101 

102 >>> @dispatch(float) 

103 ... def f(x): # noqa: F811 

104 ... return x - 1 

105 

106 >>> f(3) 

107 4 

108 >>> f(3.0) 

109 2.0 

110 """ 

111 __slots__ = '__name__', 'name', 'funcs', 'ordering', '_cache', 'doc' 

112 

113 def __init__(self, name, doc=None): 

114 self.name = self.__name__ = name 

115 self.funcs = {} 

116 self._cache = {} 

117 self.ordering = [] 

118 self.doc = doc 

119 

120 def register(self, *types, **kwargs): 

121 """ Register dispatcher with new implementation 

122 

123 >>> from sympy.multipledispatch.dispatcher import Dispatcher 

124 >>> f = Dispatcher('f') 

125 >>> @f.register(int) 

126 ... def inc(x): 

127 ... return x + 1 

128 

129 >>> @f.register(float) 

130 ... def dec(x): 

131 ... return x - 1 

132 

133 >>> @f.register(list) 

134 ... @f.register(tuple) 

135 ... def reverse(x): 

136 ... return x[::-1] 

137 

138 >>> f(1) 

139 2 

140 

141 >>> f(1.0) 

142 0.0 

143 

144 >>> f([1, 2, 3]) 

145 [3, 2, 1] 

146 """ 

147 def _(func): 

148 self.add(types, func, **kwargs) 

149 return func 

150 return _ 

151 

152 @classmethod 

153 def get_func_params(cls, func): 

154 if hasattr(inspect, "signature"): 

155 sig = inspect.signature(func) 

156 return sig.parameters.values() 

157 

158 @classmethod 

159 def get_func_annotations(cls, func): 

160 """ Get annotations of function positional parameters 

161 """ 

162 params = cls.get_func_params(func) 

163 if params: 

164 Parameter = inspect.Parameter 

165 

166 params = (param for param in params 

167 if param.kind in 

168 (Parameter.POSITIONAL_ONLY, 

169 Parameter.POSITIONAL_OR_KEYWORD)) 

170 

171 annotations = tuple( 

172 param.annotation 

173 for param in params) 

174 

175 if not any(ann is Parameter.empty for ann in annotations): 

176 return annotations 

177 

178 def add(self, signature, func, on_ambiguity=ambiguity_warn): 

179 """ Add new types/method pair to dispatcher 

180 

181 >>> from sympy.multipledispatch import Dispatcher 

182 >>> D = Dispatcher('add') 

183 >>> D.add((int, int), lambda x, y: x + y) 

184 >>> D.add((float, float), lambda x, y: x + y) 

185 

186 >>> D(1, 2) 

187 3 

188 >>> D(1, 2.0) 

189 Traceback (most recent call last): 

190 ... 

191 NotImplementedError: Could not find signature for add: <int, float> 

192 

193 When ``add`` detects a warning it calls the ``on_ambiguity`` callback 

194 with a dispatcher/itself, and a set of ambiguous type signature pairs 

195 as inputs. See ``ambiguity_warn`` for an example. 

196 """ 

197 # Handle annotations 

198 if not signature: 

199 annotations = self.get_func_annotations(func) 

200 if annotations: 

201 signature = annotations 

202 

203 # Handle union types 

204 if any(isinstance(typ, tuple) for typ in signature): 

205 for typs in expand_tuples(signature): 

206 self.add(typs, func, on_ambiguity) 

207 return 

208 

209 for typ in signature: 

210 if not isinstance(typ, type): 

211 str_sig = ', '.join(c.__name__ if isinstance(c, type) 

212 else str(c) for c in signature) 

213 raise TypeError("Tried to dispatch on non-type: %s\n" 

214 "In signature: <%s>\n" 

215 "In function: %s" % 

216 (typ, str_sig, self.name)) 

217 

218 self.funcs[signature] = func 

219 self.reorder(on_ambiguity=on_ambiguity) 

220 self._cache.clear() 

221 

222 def reorder(self, on_ambiguity=ambiguity_warn): 

223 if _resolve[0]: 

224 self.ordering = ordering(self.funcs) 

225 amb = ambiguities(self.funcs) 

226 if amb: 

227 on_ambiguity(self, amb) 

228 else: 

229 _unresolved_dispatchers.add(self) 

230 

231 def __call__(self, *args, **kwargs): 

232 types = tuple([type(arg) for arg in args]) 

233 try: 

234 func = self._cache[types] 

235 except KeyError: 

236 func = self.dispatch(*types) 

237 if not func: 

238 raise NotImplementedError( 

239 'Could not find signature for %s: <%s>' % 

240 (self.name, str_signature(types))) 

241 self._cache[types] = func 

242 try: 

243 return func(*args, **kwargs) 

244 

245 except MDNotImplementedError: 

246 funcs = self.dispatch_iter(*types) 

247 next(funcs) # burn first 

248 for func in funcs: 

249 try: 

250 return func(*args, **kwargs) 

251 except MDNotImplementedError: 

252 pass 

253 raise NotImplementedError("Matching functions for " 

254 "%s: <%s> found, but none completed successfully" 

255 % (self.name, str_signature(types))) 

256 

257 def __str__(self): 

258 return "<dispatched %s>" % self.name 

259 __repr__ = __str__ 

260 

261 def dispatch(self, *types): 

262 """ Deterimine appropriate implementation for this type signature 

263 

264 This method is internal. Users should call this object as a function. 

265 Implementation resolution occurs within the ``__call__`` method. 

266 

267 >>> from sympy.multipledispatch import dispatch 

268 >>> @dispatch(int) 

269 ... def inc(x): 

270 ... return x + 1 

271 

272 >>> implementation = inc.dispatch(int) 

273 >>> implementation(3) 

274 4 

275 

276 >>> print(inc.dispatch(float)) 

277 None 

278 

279 See Also: 

280 ``sympy.multipledispatch.conflict`` - module to determine resolution order 

281 """ 

282 

283 if types in self.funcs: 

284 return self.funcs[types] 

285 

286 try: 

287 return next(self.dispatch_iter(*types)) 

288 except StopIteration: 

289 return None 

290 

291 def dispatch_iter(self, *types): 

292 n = len(types) 

293 for signature in self.ordering: 

294 if len(signature) == n and all(map(issubclass, types, signature)): 

295 result = self.funcs[signature] 

296 yield result 

297 

298 def resolve(self, types): 

299 """ Deterimine appropriate implementation for this type signature 

300 

301 .. deprecated:: 0.4.4 

302 Use ``dispatch(*types)`` instead 

303 """ 

304 warn("resolve() is deprecated, use dispatch(*types)", 

305 DeprecationWarning) 

306 

307 return self.dispatch(*types) 

308 

309 def __getstate__(self): 

310 return {'name': self.name, 

311 'funcs': self.funcs} 

312 

313 def __setstate__(self, d): 

314 self.name = d['name'] 

315 self.funcs = d['funcs'] 

316 self.ordering = ordering(self.funcs) 

317 self._cache = {} 

318 

319 @property 

320 def __doc__(self): 

321 docs = ["Multiply dispatched method: %s" % self.name] 

322 

323 if self.doc: 

324 docs.append(self.doc) 

325 

326 other = [] 

327 for sig in self.ordering[::-1]: 

328 func = self.funcs[sig] 

329 if func.__doc__: 

330 s = 'Inputs: <%s>\n' % str_signature(sig) 

331 s += '-' * len(s) + '\n' 

332 s += func.__doc__.strip() 

333 docs.append(s) 

334 else: 

335 other.append(str_signature(sig)) 

336 

337 if other: 

338 docs.append('Other signatures:\n ' + '\n '.join(other)) 

339 

340 return '\n\n'.join(docs) 

341 

342 def _help(self, *args): 

343 return self.dispatch(*map(type, args)).__doc__ 

344 

345 def help(self, *args, **kwargs): 

346 """ Print docstring for the function corresponding to inputs """ 

347 print(self._help(*args)) 

348 

349 def _source(self, *args): 

350 func = self.dispatch(*map(type, args)) 

351 if not func: 

352 raise TypeError("No function found") 

353 return source(func) 

354 

355 def source(self, *args, **kwargs): 

356 """ Print source code for the function corresponding to inputs """ 

357 print(self._source(*args)) 

358 

359 

360def source(func): 

361 s = 'File: %s\n\n' % inspect.getsourcefile(func) 

362 s = s + inspect.getsource(func) 

363 return s 

364 

365 

366class MethodDispatcher(Dispatcher): 

367 """ Dispatch methods based on type signature 

368 

369 See Also: 

370 Dispatcher 

371 """ 

372 

373 @classmethod 

374 def get_func_params(cls, func): 

375 if hasattr(inspect, "signature"): 

376 sig = inspect.signature(func) 

377 return itl.islice(sig.parameters.values(), 1, None) 

378 

379 def __get__(self, instance, owner): 

380 self.obj = instance 

381 self.cls = owner 

382 return self 

383 

384 def __call__(self, *args, **kwargs): 

385 types = tuple([type(arg) for arg in args]) 

386 func = self.dispatch(*types) 

387 if not func: 

388 raise NotImplementedError('Could not find signature for %s: <%s>' % 

389 (self.name, str_signature(types))) 

390 return func(self.obj, *args, **kwargs) 

391 

392 

393def str_signature(sig): 

394 """ String representation of type signature 

395 

396 >>> from sympy.multipledispatch.dispatcher import str_signature 

397 >>> str_signature((int, float)) 

398 'int, float' 

399 """ 

400 return ', '.join(cls.__name__ for cls in sig) 

401 

402 

403def warning_text(name, amb): 

404 """ The text for ambiguity warnings """ 

405 text = "\nAmbiguities exist in dispatched function %s\n\n" % (name) 

406 text += "The following signatures may result in ambiguous behavior:\n" 

407 for pair in amb: 

408 text += "\t" + \ 

409 ', '.join('[' + str_signature(s) + ']' for s in pair) + "\n" 

410 text += "\n\nConsider making the following additions:\n\n" 

411 text += '\n\n'.join(['@dispatch(' + str_signature(super_signature(s)) 

412 + ')\ndef %s(...)' % name for s in amb]) 

413 return text