Package smach :: Module user_data

Source Code for Module smach.user_data

  1   
  2  import threading 
  3  import copy 
  4   
  5  import smach 
  6   
  7  __all__ = ['UserData','Remapper'] 
  8   
9 -class UserData(object):
10 """SMACH user data structure."""
11 - def __init__(self):
12 self._data = {} 13 self._locks = {} 14 self.__initialized = True
15
16 - def update(self, other_userdata):
17 """Combine this userdata struct with another. 18 This overwrites duplicate keys with values from C{other_userdata}. 19 """ 20 # Merge data 21 self._data.update(other_userdata._data)
22
23 - def extract(self, keys, remapping):
24 ud = UserData() 25 reverse_remapping = {remapping[k]: k for k in remapping} 26 if len(reverse_remapping) != len(remapping): 27 smach.logerr("SMACH userdata remapping is not one-to-one: " + str(remapping)) 28 for k in keys: 29 rmk = k 30 if k in reverse_remapping: 31 rmk = reverse_remapping[k] 32 ud[rmk] = copy.copy(self[k]) 33 return ud
34
35 - def merge(self, ud, keys, remapping):
36 for k in keys: 37 rmk = k 38 if k in remapping: 39 rmk = remapping[k] 40 self[rmk] = copy.copy(ud[k])
41
42 - def __getitem__(self, key):
43 return self.__getattr__(key)
44
45 - def __setitem__(self, key, item):
46 self._data[key] = item
47
48 - def keys(self):
49 return list(self._data.keys())
50
51 - def __contains__(self,key):
52 return key in self._data
53
54 - def __getattr__(self, name):
55 """Override getattr to be thread safe.""" 56 if name[0] == '_': 57 return object.__getattr__(self, name) 58 if not name in self._locks.keys(): 59 self._locks[name] = threading.Lock() 60 61 try: 62 with self._locks[name]: 63 temp = self._data[name] 64 except: 65 smach.logerr("Userdata key '%s' not available. Available keys are: %s" % (name, str(list(self._data.keys())))) 66 raise KeyError() 67 68 return temp
69
70 - def __setattr__(self, name, value):
71 """Override setattr to be thread safe.""" 72 # If we're still in __init__ don't do anything special 73 if name[0] == '_' or '_UserData__initialized' not in self.__dict__: 74 return object.__setattr__(self, name, value) 75 76 if not name in self._locks.keys(): 77 self._locks[name] = threading.Lock() 78 79 self._locks[name].acquire() 80 self._data[name] = value 81 self._locks[name].release()
82 83 # Const wrapper
84 -def get_const(obj):
85 """Get a const reference to an object if it has "user-defined" attributes.""" 86 if hasattr(obj,'__dict__'): 87 smach.logdebug("Making const '%s'" % str(obj)) 88 return Const(obj) 89 else: 90 return obj
91
92 -class Const(object):
93 """Wrapper that treats "user-defined" fields as immutable. 94 95 This wrapper class is used when user data keys are specified as input keys, 96 but not as output keys. 97 """
98 - def __init__(self, obj):
99 smach.logdebug("Making const '%s'" % str(obj)) 100 self._obj = obj 101 self.__initialized = True
102
103 - def __getattr__(self, name):
104 smach.logdebug("Getting '%s' from const wrapper." % name) 105 attr = getattr(self._obj,name) 106 return get_const(attr)
107
108 - def __getitem__(self, name):
109 smach.logdebug("Getting '%s' from const wrapper." % name) 110 attr = self._obj[name] 111 return get_const(attr)
112
113 - def __setattr__(self, name, value):
114 if '_const__initialized' not in self.__dict__: 115 return object.__setattr__(self, name, value) 116 smach.logerr("Attempting to set '%s' but this member is read-only." % name) 117 raise TypeError()
118
119 - def __delattr__(self, name):
120 smach.logerr("Attempting to delete '%s' but this member is read-only." % name) 121 raise TypeError()
122
123 -class Remapper(object):
124 """Key-remapping proxy to a SMACH userdata structure."""
125 - def __init__(self, ud, input_keys=[], output_keys=[], remapping={}):
126 self._ud = ud 127 self._input = input_keys 128 self._output = output_keys 129 self._map = remapping 130 self.__initialized = True
131
132 - def _remap(self, key):
133 """Return either the key or it's remapped value.""" 134 if key in self._map: 135 return self._map[key] 136 return key
137
138 - def update(self, other_userdata):
139 self._ud.update(other_userdata)
140
141 - def __getitem__(self, key):
142 if key not in self._input: 143 raise smach.InvalidUserCodeError("Reading from SMACH userdata key '%s' but the only keys that were declared as input to this state were: %s. This key needs to be declaread as input to this state. " % (key, self._input)) 144 if key not in self._output: 145 return get_const(self._ud.__getitem__(self._remap(key))) 146 return self._ud.__getitem__(self._remap(key))
147
148 - def __setitem__(self, key, item):
149 if key not in self._output: 150 smach.logerr("Writing to SMACH userdata key '%s' but the only keys that were declared as output from this state were: %s." % (key, self._output)) 151 return 152 self._ud.__setitem__(self._remap(key),item)
153
154 - def keys(self):
155 return [self._remap(key) for key in self._ud.keys() if key in self._input]
156
157 - def __contains__(self,key):
158 if key in self._input: 159 return self._remap(key) in self._ud 160 else: 161 return False
162
163 - def __getattr__(self, name):
164 if name[0] == '_': 165 return object.__getattr__(self, name) 166 if name not in self._input: 167 raise smach.InvalidUserCodeError("Reading from SMACH userdata key '%s' but the only keys that were declared as input to this state were: %s. This key needs to be declaread as input to this state. " % (name, self._input)) 168 if name not in self._output: 169 return get_const(getattr(self._ud, self._remap(name))) 170 return getattr(self._ud, self._remap(name))
171
172 - def __setattr__(self, name, value):
173 if name[0] == '_' or '_Remapper__initialized' not in self.__dict__: 174 return object.__setattr__(self, name, value) 175 if name not in self._output: 176 smach.logerr("Writing to SMACH userdata key '%s' but the only keys that were declared as output from this state were: %s." % (name, self._output)) 177 return None 178 setattr(self._ud, self._remap(name), value)
179