__init__.py (4795B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\functoolsext\__init__.py 2 from collections import namedtuple 3 from functools import * 4 import inspect 5 from threading import RLock 6 _CacheInfo = namedtuple('CacheInfo', ['hits', 7 'misses', 8 'maxsize', 9 'currsize']) 10 11 class _HashedSeq(list): 12 __slots__ = 'hashvalue' 13 14 def __init__(self, tup, hash = hash): 15 self[:] = tup 16 self.hashvalue = hash(tup) 17 18 def __hash__(self): 19 return self.hashvalue 20 21 22 def _make_key(args, kwds, typed, kwd_mark = (object(),), fasttypes = set([int, 23 str, 24 frozenset, 25 type(None)]), sorted = sorted, tuple = tuple, type = type, len = len): 26 key = args 27 if kwds: 28 sorted_items = sorted(kwds.items()) 29 key += kwd_mark 30 for item in sorted_items: 31 key += item 32 33 if typed: 34 key += tuple((type(v) for v in args)) 35 if kwds: 36 key += tuple((type(v) for k, v in sorted_items)) 37 elif len(key) == 1 and type(key[0]) in fasttypes: 38 return key[0] 39 return _HashedSeq(key) 40 41 42 def lru_cache(maxsize = 128, typed = False): 43 44 def decorating_function(user_function): 45 cache = dict() 46 stats = [0, 0] 47 HITS, MISSES = (0, 1) 48 make_key = _make_key 49 cache_get = cache.get 50 _len = len 51 lock = RLock() 52 root = [] 53 root[:] = [root, 54 root, 55 None, 56 None] 57 nonlocal_root = [root] 58 PREV, NEXT, KEY, RESULT = (0, 1, 2, 3) 59 if maxsize == 0: 60 61 def wrapper(*args, **kwds): 62 result = user_function(*args, **kwds) 63 stats[MISSES] += 1 64 return result 65 66 elif maxsize is None: 67 68 def wrapper(*args, **kwds): 69 key = make_key(args, kwds, typed) 70 result = cache_get(key, root) 71 if result is not root: 72 stats[HITS] += 1 73 return result 74 result = user_function(*args, **kwds) 75 cache[key] = result 76 stats[MISSES] += 1 77 return result 78 79 else: 80 81 def wrapper(*args, **kwds): 82 key = make_key(args, kwds, typed) if kwds or typed else args 83 with lock: 84 link = cache_get(key) 85 if link is not None: 86 root, = nonlocal_root 87 link_prev, link_next, key, result = link 88 link_prev[NEXT] = link_next 89 link_next[PREV] = link_prev 90 last = root[PREV] 91 last[NEXT] = root[PREV] = link 92 link[PREV] = last 93 link[NEXT] = root 94 stats[HITS] += 1 95 return result 96 result = user_function(*args, **kwds) 97 with lock: 98 root, = nonlocal_root 99 if key in cache: 100 pass 101 elif _len(cache) >= maxsize: 102 oldroot = root 103 oldroot[KEY] = key 104 oldroot[RESULT] = result 105 root = nonlocal_root[0] = oldroot[NEXT] 106 oldkey = root[KEY] 107 oldvalue = root[RESULT] 108 root[KEY] = root[RESULT] = None 109 del cache[oldkey] 110 cache[key] = oldroot 111 else: 112 last = root[PREV] 113 link = [last, 114 root, 115 key, 116 result] 117 last[NEXT] = root[PREV] = cache[key] = link 118 stats[MISSES] += 1 119 return result 120 121 def cache_info(): 122 with lock: 123 return _CacheInfo(stats[HITS], stats[MISSES], maxsize, len(cache)) 124 125 def cache_clear(): 126 with lock: 127 cache.clear() 128 root = nonlocal_root[0] 129 root[:] = [root, 130 root, 131 None, 132 None] 133 stats[:] = [0, 0] 134 135 wrapper.__wrapped__ = user_function 136 wrapper.cache_info = cache_info 137 wrapper.cache_clear = cache_clear 138 return update_wrapper(wrapper, user_function) 139 140 return decorating_function 141 142 143 def func_takes_arguments(f): 144 try: 145 fc = f.__code__ 146 except AttributeError: 147 raise AttributeError('Cannot inspect built-ins: %s.' % f) 148 149 arguments, varargs, keywords = inspect.getargs(fc) 150 if inspect.ismethod(f): 151 arguments.pop(0) 152 if any([arguments, varargs, keywords]): 153 return True 154 return False