-
Notifications
You must be signed in to change notification settings - Fork 0
/
Map.py
93 lines (69 loc) · 2.63 KB
/
Map.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
import threading
import sys
sys.tracebacklimit = 0
__all__ = ['Map', 'MapAccess']
def get_const(obj):
if hasattr(obj, '__dict__'):
return Const(obj)
else:
return obj
class Const(object):
def __init__(self, obj):
self._obj = obj
self.__initialized = True
def __getattr__(self, name):
attr = getattr(self._obj,name)
return get_const(attr)
def __getitem__(self, name):
attr = self._obj[name]
return get_const(attr)
def __setattr__(self, name, value):
if '_const__initialized' not in self.__dict__:
return object.__setattr__(self, name, value)
raise TypeError()
def __delattr__(self, name):
raise TypeError()
class Map(object):
def __init__(self):
super(Map, self).__setattr__('_map', dict())
super(Map, self).__setattr__('_locks', dict())
def __getattr__(self, key):
if key[0] == '_':
return object.__getattr__(self, key)
if not key in self._locks.keys():
self._locks[key] = threading.Lock()
try:
with self._locks[key]:
return self._map[key]
except:
print("[Map] Key '{}' is not available. Available keys are : {}".format(key, str(list(self._map.keys()))))
def __setattr__(self, key, value):
if key[0] == '_':
return object.__setattr__(self, key, value)
if not key in self._locks.keys():
self._locks[key] = threading.Lock()
self._locks[key].acquire()
self._map[key] = value
self._locks[key].release()
class MapAccess(object):
def __init__(self, mapObj, input_keys=[], output_keys=[]):
super(MapAccess, self).__setattr__('_mapObj', mapObj)
super(MapAccess, self).__setattr__('_input', set(input_keys))
super(MapAccess, self).__setattr__('_output', set(output_keys))
self.__initialized = False
def __getattr__(self, key):
if key[0] == '_':
return object.__getattr__(self, key)
if key not in self._input:
raise AttributeError("[Map Access] '{}' is not declared in inputs, so read permission denied. [inputs: {}]".format(key, self._input))
if key not in self._output:
return get_const(getattr(self._mapObj, key))
return getattr(self._mapObj, key)
def __setattr__(self, key, value):
if '_MapAccess__initialized' not in self.__dict__:
return object.__setattr__(self, key, value)
elif key in ['_mapObj', '_input', '_output'] and ('_MapAccess__initialized' in self.__dict__):
raise AttributeError("[MapAcess] '{}' is read-only attribute".format(key))
if key not in self._output:
raise AttributeError("[Map Access] '{}' is not declared in outputs, so write permission denied. [outputs: {}]".format(key, self._output))
setattr(self._mapObj, key, value)