-
Notifications
You must be signed in to change notification settings - Fork 74
/
msg_loader.py
489 lines (426 loc) · 20.3 KB
/
msg_loader.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
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
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
# Software License Agreement (BSD License)
#
# Copyright (c) 2008, Willow Garage, Inc.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions
# are met:
#
# * Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# * Redistributions in binary form must reproduce the above
# copyright notice, this list of conditions and the following
# disclaimer in the documentation and/or other materials provided
# with the distribution.
# * Neither the name of Willow Garage, Inc. nor the names of its
# contributors may be used to endorse or promote products derived
# from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
# FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
# COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
# BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
from __future__ import print_function
"""
Loader for messages and :class:`MsgContext` that assumes a
dictionary-based search path scheme (keys are the package/namespace,
values are the paths). Compatible with ROS package system and other
possible layouts.
"""
import ast
import os
import sys
try:
from cStringIO import StringIO # Python 2.x
except ImportError:
from io import StringIO # Python 3.x
from . base import InvalidMsgSpec, log, SEP, COMMENTCHAR, CONSTCHAR, IODELIM, EXT_MSG, EXT_SRV
from . msgs import MsgSpec, TIME, TIME_MSG, DURATION, DURATION_MSG, HEADER, HEADER_FULL_NAME, \
is_builtin, is_valid_msg_field_name, is_valid_msg_type, bare_msg_type, is_valid_constant_type, \
Field, Constant, resolve_type
from . names import normalize_package_context, package_resource_name
from . srvs import SrvSpec
class MsgNotFound(Exception):
def __init__(self, message, base_type=None, package=None, search_path=None):
super(MsgNotFound, self).__init__(message)
self.base_type = base_type
self.package = package
self.search_path = search_path
def get_msg_file(package, base_type, search_path, ext=EXT_MSG):
"""
Determine the file system path for the specified ``.msg`` on
*search_path*.
:param package: name of package file is in, ``str``
:param base_type: type name of message, e.g. 'Point2DFloat32', ``str``
:param search_path: dictionary mapping message namespaces to a directory locations
:param ext: msg file extension. Override with EXT_SRV to search for services instead.
:returns: filesystem path of requested file, ``str``
:raises: :exc:`MsgNotFound` If message cannot be located.
"""
log("msg_file(%s, %s, %s)" % (package, base_type, str(search_path)))
if not isinstance(search_path, dict):
raise ValueError("search_path must be a dictionary of {namespace: dirpath}")
if not package in search_path:
raise MsgNotFound("Cannot locate message [%s]: unknown package [%s] on search path [%s]" \
% (base_type, package, search_path), base_type, package, search_path)
else:
for path_tmp in search_path[package]:
path = os.path.join(path_tmp, "%s%s"%(base_type, ext))
if os.path.isfile(path):
return path
raise MsgNotFound("Cannot locate message [%s] in package [%s] with paths [%s]"%
(base_type, package, str(search_path[package])), base_type, package, search_path)
def get_srv_file(package, base_type, search_path):
"""
Determine the file system path for the specified .srv on path.
:param package: name of package ``.srv`` file is in, ``str``
:param base_type: type name of service, e.g. 'Empty', ``str``
:param search_path: dictionary mapping message namespaces to a directory locations
:returns: file path of ``.srv`` file in specified package, ``str``
:raises: :exc:`MsgNotFound` If service file cannot be located.
"""
return get_msg_file(package, base_type, search_path, ext=EXT_SRV)
def load_msg_by_type(msg_context, msg_type, search_path):
"""
Load message specification for specified type.
NOTE: this will register the message in the *msg_context*.
:param msg_context: :class:`MsgContext` for finding loaded dependencies
:param msg_type: relative or full message type.
:param search_path: dictionary mapping message namespaces to a directory locations
:returns: :class:`MsgSpec` instance, ``(str, MsgSpec)``
:raises: :exc:`MsgNotFound` If message cannot be located.
"""
log("load_msg_by_type(%s, %s)" % (msg_type, str(search_path)))
if not isinstance(search_path, dict):
raise ValueError("search_path must be a dictionary of {namespace: dirpath}")
if msg_type == HEADER:
msg_type = HEADER_FULL_NAME
package_name, base_type = package_resource_name(msg_type)
file_path = get_msg_file(package_name, base_type, search_path)
log("file_path", file_path)
spec = load_msg_from_file(msg_context, file_path, msg_type)
msg_context.set_file(msg_type, file_path)
return spec
def load_srv_by_type(msg_context, srv_type, search_path):
"""
Load service specification for specified type.
NOTE: services are *never* registered in a :class:`MsgContext`.
:param msg_context: :class:`MsgContext` for finding loaded dependencies
:param srv_type: relative or full message type.
:param search_path: dictionary mapping message namespaces to a directory locations
:returns: :class:`MsgSpec` instance, ``(str, MsgSpec)``
:raises: :exc:`MsgNotFound` If message cannot be located.
"""
log("load_srv_by_type(%s, %s)" % (srv_type, str(search_path)))
if not isinstance(search_path, dict):
raise ValueError("search_path must be a dictionary of {namespace: dirpath}")
package_name, base_type = package_resource_name(srv_type)
file_path = get_srv_file(package_name, base_type, search_path)
log("file_path", file_path)
return load_srv_from_file(msg_context, file_path, srv_type)
def convert_constant_value(field_type, val):
"""
Convert constant value declaration to python value. Does not do
type-checking, so ValueError or other exceptions may be raised.
:param field_type: ROS field type, ``str``
:param val: string representation of constant, ``str``
:raises: :exc:`ValueError` If unable to convert to python representation
:raises: :exc:`InvalidMsgSpec` If value exceeds specified integer width
"""
if field_type in ['float32','float64']:
return float(val)
elif field_type in ['string']:
return val.strip() #string constants are always stripped
elif field_type in ['int8', 'uint8', 'int16','uint16','int32','uint32','int64','uint64', 'char', 'byte']:
# bounds checking
bits = [('int8', 8), ('uint8', 8), ('int16', 16),('uint16', 16),\
('int32', 32),('uint32', 32), ('int64', 64),('uint64', 64),\
('byte', 8), ('char', 8)]
b = [b for t, b in bits if t == field_type][0]
import math
if field_type[0] == 'u' or field_type == 'char':
lower = 0
upper = int(math.pow(2, b)-1)
else:
upper = int(math.pow(2, b-1)-1)
lower = -upper - 1 #two's complement min
val = int(val) #python will autocast to long if necessary
if val > upper or val < lower:
raise InvalidMsgSpec("cannot coerce [%s] to %s (out of bounds)"%(val, field_type))
return val
elif field_type == 'bool':
return True if ast.literal_eval(val) else False
raise InvalidMsgSpec("invalid constant type: [%s]"%field_type)
def _load_constant_line(orig_line):
"""
:raises: :exc:`InvalidMsgSpec`
"""
clean_line = _strip_comments(orig_line)
line_splits = [s for s in [x.strip() for x in clean_line.split(" ")] if s] #split type/name, filter out empties
field_type = line_splits[0]
if not is_valid_constant_type(field_type):
raise InvalidMsgSpec("%s is not a legal constant type"%field_type)
if field_type == 'string':
# strings contain anything to the right of the equals sign, there are no comments allowed
idx = orig_line.find(CONSTCHAR)
name = orig_line[orig_line.find(' ')+1:idx]
val = orig_line[idx+1:]
else:
line_splits = [x.strip() for x in ' '.join(line_splits[1:]).split(CONSTCHAR)] #resplit on '='
if len(line_splits) != 2:
raise InvalidMsgSpec("Invalid constant declaration: %s"%orig_line)
name = line_splits[0]
val = line_splits[1]
try:
val_converted = convert_constant_value(field_type, val)
except Exception as e:
raise InvalidMsgSpec("Invalid constant value: %s"%e)
return Constant(field_type, name, val_converted, val.strip())
def _load_field_line(orig_line, package_context):
"""
:returns: (field_type, name) tuple, ``(str, str)``
:raises: :exc:`InvalidMsgSpec`
"""
#log("_load_field_line", orig_line, package_context)
clean_line = _strip_comments(orig_line)
line_splits = [s for s in [x.strip() for x in clean_line.split(" ")] if s] #split type/name, filter out empties
if len(line_splits) != 2:
raise InvalidMsgSpec("Invalid declaration: %s"%(orig_line))
field_type, name = line_splits
if not is_valid_msg_field_name(name):
raise InvalidMsgSpec("%s is not a legal message field name"%name)
if not is_valid_msg_type(field_type):
raise InvalidMsgSpec("%s is not a legal message field type"%field_type)
if package_context and not SEP in field_type:
if field_type == HEADER:
field_type = HEADER_FULL_NAME
elif not is_builtin(bare_msg_type(field_type)):
field_type = "%s/%s"%(package_context, field_type)
elif field_type == HEADER:
field_type = HEADER_FULL_NAME
return field_type, name
def _strip_comments(line):
return line.split(COMMENTCHAR)[0].strip() #strip comments
def load_msg_from_string(msg_context, text, full_name):
"""
Load message specification from a string.
NOTE: this will register the message in the *msg_context*.
:param msg_context: :class:`MsgContext` for finding loaded dependencies
:param text: .msg text , ``str``
:returns: :class:`MsgSpec` specification
:raises: :exc:`InvalidMsgSpec` If syntax errors or other problems are detected in file
"""
log("load_msg_from_string", full_name)
package_name, short_name = package_resource_name(full_name)
types = []
names = []
constants = []
for orig_line in text.split('\n'):
clean_line = _strip_comments(orig_line)
if not clean_line:
continue #ignore empty lines
if CONSTCHAR in clean_line:
constants.append(_load_constant_line(orig_line))
else:
field_type, name = _load_field_line(orig_line, package_name)
types.append(field_type)
names.append(name)
spec = MsgSpec(types, names, constants, text, full_name, package_name)
msg_context.register(full_name, spec)
return spec
def load_msg_from_file(msg_context, file_path, full_name):
"""
Convert the .msg representation in the file to a :class:`MsgSpec` instance.
NOTE: this will register the message in the *msg_context*.
:param file_path: path of file to load from, ``str``
:returns: :class:`MsgSpec` instance
:raises: :exc:`InvalidMsgSpec`: if syntax errors or other problems are detected in file
"""
log("Load spec from", file_path)
with open(file_path, 'r') as f:
text = f.read()
try:
return load_msg_from_string(msg_context, text, full_name)
except InvalidMsgSpec as e:
raise InvalidMsgSpec('%s: %s'%(file_path, e))
def load_msg_depends(msg_context, spec, search_path):
"""
Add the list of message types that spec depends on to depends.
:param msg_context: :class:`MsgContext` instance to load dependencies into/from.
:param spec: message to compute dependencies for, :class:`MsgSpec`/:class:`SrvSpec`
:param search_path: dictionary mapping message namespaces to a directory locations
:param deps: for recursion use only, do not set
:returns: list of dependency names, ``[str]``
:raises: :exc:`MsgNotFound` If dependency cannot be located.
"""
package_context = spec.package
log("load_msg_depends <spec>", spec.full_name, package_context)
depends = []
# Iterate over each field, loading as necessary
for unresolved_type in spec.types:
bare_type = bare_msg_type(unresolved_type)
resolved_type = resolve_type(bare_type, package_context)
if is_builtin(resolved_type):
continue
# Retrieve the MsgSpec instance of the field
if msg_context.is_registered(resolved_type):
depspec = msg_context.get_registered(resolved_type)
else:
# load and register on demand
depspec = load_msg_by_type(msg_context, resolved_type, search_path)
msg_context.register(resolved_type, depspec)
# Update dependencies
depends.append(resolved_type)
# - check to see if we have compute dependencies of field
dep_dependencies = msg_context.get_depends(resolved_type)
if dep_dependencies is None:
load_msg_depends(msg_context, depspec, search_path)
assert spec.full_name, "MsgSpec must have a properly set full name"
msg_context.set_depends(spec.full_name, depends)
# have to copy array in order to prevent inadvertent mutation (we've stored this list in set_dependencies)
return depends[:]
def load_depends(msg_context, spec, msg_search_path):
"""
Compute dependencies of *spec* and load their MsgSpec dependencies
into *msg_context*.
NOTE: *msg_search_path* is only for finding .msg files. ``.srv``
files have a separate and distinct search path. As services
cannot depend on other services, it is not necessary to provide
the srv search path here.
:param msg_context: :class:`MsgContext` instance to load dependencies into/from.
:param spec: :class:`MsgSpec` or :class:`SrvSpec` instance to load dependencies for.
:param msg_search_path: dictionary mapping message namespaces to a directory locations.
:raises: :exc:`MsgNotFound` If dependency cannot be located.
"""
if isinstance(spec, MsgSpec):
return load_msg_depends(msg_context, spec, msg_search_path)
elif isinstance(spec, SrvSpec):
depends = load_msg_depends(msg_context, spec.request, msg_search_path)
depends.extend(load_msg_depends(msg_context, spec.response, msg_search_path))
return depends
else:
raise ValueError("spec does not appear to be a message or service")
class MsgContext(object):
"""
Context object for storing :class:`MsgSpec` instances and related
metadata.
NOTE: All APIs work on :class:`MsgSpec` instance information.
Thus, for services, there is information for the request and
response messages, but there is no direct information about the
:class:`SrvSpec` instance.
"""
def __init__(self):
self._registered_packages = {}
self._files = {}
self._dependencies = {}
def set_file(self, full_msg_type, file_path):
self._files[full_msg_type] = file_path
def get_file(self, full_msg_type):
return self._files.get(full_msg_type, None)
def set_depends(self, full_msg_type, dependencies):
"""
:param dependencies: direct first order
dependencies for *full_msg_type*
"""
log("set_depends", full_msg_type, dependencies)
self._dependencies[full_msg_type] = dependencies
def get_depends(self, full_msg_type):
"""
:returns: List of dependencies for *full_msg_type*,
only first order dependencies
"""
return self._dependencies.get(full_msg_type, None)
def get_all_depends(self, full_msg_type):
all_deps = []
depends = self.get_depends(full_msg_type)
if depends is None:
raise KeyError(full_msg_type)
for d in depends:
all_deps.extend([d])
all_deps.extend(self.get_all_depends(d))
return all_deps
@staticmethod
def create_default():
msg_context = MsgContext()
# register builtins (needed for serialization). builtins have no package.
load_msg_from_string(msg_context, TIME_MSG, TIME)
load_msg_from_string(msg_context, DURATION_MSG, DURATION)
return msg_context
def register(self, full_msg_type, msgspec):
full_msg_type = bare_msg_type(full_msg_type)
package, base_type = package_resource_name(full_msg_type)
if package not in self._registered_packages:
self._registered_packages[package] = {}
self._registered_packages[package][base_type] = msgspec
def is_registered(self, full_msg_type):
"""
:param full_msg_type: Fully resolve message type
:param default_package: default package namespace to resolve
in. May be ignored by special types (e.g. time/duration).
:returns: ``True`` if :class:`MsgSpec` instance has been loaded for the requested type.
"""
full_msg_type = bare_msg_type(full_msg_type)
package, base_type = package_resource_name(full_msg_type)
if package in self._registered_packages:
return base_type in self._registered_packages[package]
else:
return False
def get_registered(self, full_msg_type):
"""
:raises: :exc:`KeyError` If not registered
"""
full_msg_type = bare_msg_type(full_msg_type)
if self.is_registered(full_msg_type):
package, base_type = package_resource_name(full_msg_type)
return self._registered_packages[package][base_type]
else:
raise KeyError(full_msg_type)
def __str__(self):
return str(self._registered_packages)
def load_srv_from_string(msg_context, text, full_name):
"""
Load :class:`SrvSpec` from the .srv file.
:param msg_context: :class:`MsgContext` instance to load request/response messages into.
:param text: .msg text , ``str``
:param package_name: context to use for msg type name, i.e. the package name,
or '' to use local naming convention. ``str``
:returns: :class:`SrvSpec` instance
:raises :exc:`InvalidMsgSpec` If syntax errors or other problems are detected in file
"""
text_in = StringIO()
text_out = StringIO()
accum = text_in
for l in text.split('\n'):
l = l.split(COMMENTCHAR)[0].strip() #strip comments
if l.startswith(IODELIM): #lenient, by request
accum = text_out
else:
accum.write(l+'\n')
# create separate MsgSpec objects for each half of file
msg_in = load_msg_from_string(msg_context, text_in.getvalue(), '%sRequest'%(full_name))
msg_out = load_msg_from_string(msg_context, text_out.getvalue(), '%sResponse'%(full_name))
return SrvSpec(msg_in, msg_out, text, full_name)
def load_srv_from_file(msg_context, file_path, full_name):
"""
Convert the .srv representation in the file to a :class:`SrvSpec` instance.
:param msg_context: :class:`MsgContext` instance to load request/response messages into.
:param file_name: name of file to load from, ``str``
:returns: :class:`SrvSpec` instance
:raise: :exc:`InvalidMsgSpec` If syntax errors or other problems are detected in file
"""
log("Load spec from %s %s\n"%(file_path, full_name))
with open(file_path, 'r') as f:
text = f.read()
spec = load_srv_from_string(msg_context, text, full_name)
msg_context.set_file('%sRequest'%(full_name), file_path)
msg_context.set_file('%sResponse'%(full_name), file_path)
return spec