-
Notifications
You must be signed in to change notification settings - Fork 170
/
lookup.py
656 lines (554 loc) · 28.2 KB
/
lookup.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
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
# Copyright (c) 2011, 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 the 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.
# Author Tully Foote/tfoote@willowgarage.com, Ken Conley/kwc@willowgarage.com
from __future__ import print_function
import sys
import yaml
from collections import defaultdict
from rospkg import RosPack, RosStack, ResourceNotFound
from .core import RosdepInternalError, InvalidData, rd_debug
from .model import RosdepDatabase
from .rospkg_loader import RosPkgLoader
from .dependency_graph import DependencyGraph
from .sources_list import SourcesListLoader
from . import catkin_packages
class RosdepDefinition(object):
"""
Single rosdep dependency definition. This data is stored as the
raw dictionary definition for the dependency.
See REP 111, 'Multiple Package Manager Support for Rosdep' for a
discussion of this raw format.
"""
def __init__(self, rosdep_key, data, origin='<dynamic>'):
"""
:param rosdep_key: key/name of rosdep dependency
:param data: raw rosdep data for a single rosdep dependency, ``dict``
:param origin: string that indicates where data originates from (e.g. filename)
"""
self.rosdep_key = rosdep_key
if not isinstance(data, dict):
raise InvalidData('rosdep data for [%s] must be a dictionary' % (self.rosdep_key), origin=origin)
self.data = data
self.origin = origin
def reverse_merge(self, new_data, origin='<dynamic>', verbose=False):
"""
Merge two definitions together, with existing rules taking precendence.
Definitions are merged at the os_name level, meaning that if two rules
exist with the same os_name, the first one wins.
:param data: raw rosdep data for a single rosdep dependency, ``dict``
:param origin: string that indicates where this new data comes from (e.g. filename)
"""
for os_name, rules in new_data.items():
if os_name not in self.data:
if verbose:
print('[%s] adding rules for os [%s] to [%s]' % (origin, os_name, self.rosdep_key), file=sys.stderr)
self.data[os_name] = rules
elif verbose:
print('[%s] ignoring [%s] for os [%s], already loaded' % (origin, self.rosdep_key, os_name), file=sys.stderr)
def get_rule_for_platform(self, os_name, os_version, installer_keys, default_installer_key):
"""
Get installer_key and rule for the specified rule. See REP 111 for precedence rules.
:param os_name: OS name to get rule for
:param os_version: OS version to get rule for
:param installer_keys: Keys of installers for platform, ``[str]``
:param default_installer_key: Default installer key for platform, ``[str]``
:returns: (installer_key, rosdep_args_dict), ``(str, dict)``
:raises: :exc:`ResolutionError` If no rule is available
:raises: :exc:`InvalidData` If rule data is not valid
"""
rosdep_key = self.rosdep_key
data = self.data
if type(data) != dict:
raise InvalidData('rosdep value for [%s] must be a dictionary' % (self.rosdep_key), origin=self.origin)
if os_name not in data:
raise ResolutionError(rosdep_key, data, os_name, os_version, 'No definition of [%s] for OS [%s]' % (rosdep_key, os_name))
data = data[os_name]
return_key = default_installer_key
# REP 111: rosdep first interprets the key as a
# PACKAGE_MANAGER. If this test fails, it will be interpreted
# as an OS_VERSION_CODENAME.
if type(data) == dict:
for installer_key in installer_keys:
if installer_key in data:
data = data[installer_key]
return_key = installer_key
break
else:
# data must be a dictionary, string, or list
if type(data) == dict:
# check for
# hardy:
# apt:
# stuff
# we've already checked for PACKAGE_MANAGER_KEY, so
# version key must be present here for data to be valid
# dictionary value.
# if the os_version is not defined and there is no wildcard
if os_version not in data and '*' not in data:
raise ResolutionError(rosdep_key, self.data, os_name, os_version, 'No definition of [%s] for OS version [%s]' % (rosdep_key, os_version))
# if the os_version has the value None
if os_version in data and data[os_version] is None:
raise ResolutionError(rosdep_key, self.data, os_name, os_version, '[%s] defined as "not available" for OS version [%s]' % (rosdep_key, os_version))
# if os version is not defined (and there is a wildcard) fallback to the wildcard
if os_version not in data:
os_version = '*'
data = data[os_version]
if type(data) == dict:
for installer_key in installer_keys:
if installer_key in data:
data = data[installer_key]
return_key = installer_key
break
# Check if the rule is null
if data is None:
raise ResolutionError(rosdep_key, self.data, os_name, os_version, '[%s] defined as "not available" for OS version [%s]' % (rosdep_key, os_version))
if type(data) not in (dict, list, type('str')):
raise InvalidData('rosdep OS definition for [%s:%s] must be a dictionary, string, or list: %s' % (self.rosdep_key, os_name, data), origin=self.origin)
return return_key, data
def __str__(self):
return '%s:\n%s' % (self.origin, yaml.dump(self.data, default_flow_style=False))
class ResolutionError(Exception):
def __init__(self, rosdep_key, rosdep_data, os_name, os_version, message):
self.rosdep_key = rosdep_key
self.rosdep_data = rosdep_data
self.os_name = os_name
self.os_version = os_version
super(ResolutionError, self).__init__(message)
def __str__(self):
if self.rosdep_data:
pretty_data = yaml.dump(self.rosdep_data, default_flow_style=False)
else:
pretty_data = '<no data>'
return """%s
\trosdep key : %s
\tOS name : %s
\tOS version : %s
\tData:\n%s""" % (self.args[0], self.rosdep_key, self.os_name, self.os_version, pretty_data.replace('\n', '\n\t\t'))
class RosdepView(object):
"""
View of :class:`RosdepDatabase`. Unlike :class:`RosdepDatabase`,
which stores :class:`RosdepDatabaseEntry` data for all stacks, a
view merges entries for a particular stack. This view can then be
queried to lookup and resolve individual rosdep dependencies.
"""
def __init__(self, name):
self.name = name
self.rosdep_defs = {} # {str: RosdepDefinition}
def __str__(self):
return '\n'.join(['%s: %s' % val for val in self.rosdep_defs.items()])
def lookup(self, rosdep_name):
"""
:returns: :class:`RosdepDefinition`
:raises: :exc:`KeyError` If *rosdep_name* is not declared
"""
return self.rosdep_defs[rosdep_name]
def keys(self):
"""
:returns: list of rosdep names in this view
"""
return self.rosdep_defs.keys()
def merge(self, update_entry, override=False, verbose=False):
"""
Merge rosdep database update into main database. Merge rules
are first entry to declare a key wins. There are no
conflicts. This rule logic is modelled after the apt sources
list.
:param override: Ignore first-one-wins rules and instead
always use rules from update_entry
"""
if verbose:
print('view[%s]: merging from cache of [%s]' % (self.name, update_entry.origin))
db = self.rosdep_defs
for dep_name, dep_data in update_entry.rosdep_data.items():
# convert data into RosdepDefinition model
update_definition = RosdepDefinition(dep_name, dep_data, update_entry.origin)
# First rule wins or override, no rule-merging.
if override or dep_name not in db:
db[dep_name] = update_definition
elif dep_name in db:
db[dep_name].reverse_merge(dep_data, update_entry.origin, verbose=verbose)
def prune_catkin_packages(rosdep_keys, verbose=False):
workspace_pkgs = catkin_packages.get_workspace_packages()
if not workspace_pkgs:
return rosdep_keys
for i, rosdep_key in reversed(list(enumerate(rosdep_keys))):
if rosdep_key in workspace_pkgs:
# If workspace packages listed (--catkin-workspace)
# and if the rosdep_key is a package in that
# workspace, then skip it rather than resolve it
if verbose:
print("rosdep key '{0}'".format(rosdep_key) +
' is in the catkin workspace, skipping.',
file=sys.stderr)
del rosdep_keys[i]
return rosdep_keys
def prune_skipped_packages(rosdep_keys, skipped_keys, verbose=False):
if not skipped_keys:
return rosdep_keys
for i, rosdep_key in reversed(list(enumerate(rosdep_keys))):
if rosdep_key in skipped_keys:
# If the key is in the list of keys to explicitly skip, skip it
if verbose:
print("rosdep key '{0}'".format(rosdep_key) +
' was listed in the skipped packages, skipping.',
file=sys.stderr)
del rosdep_keys[i]
return rosdep_keys
class RosdepLookup(object):
"""
Lookup rosdep definitions. Provides API for most
non-install-related commands for rosdep.
:class:`RosdepLookup` caches data as it is loaded, so changes made
on the filesystem will not be reflected if the rosdep information
has already been loaded.
"""
def __init__(self, rosdep_db, loader):
"""
:param loader: Loader to use for loading rosdep data by stack
name, ``RosdepLoader``
:param rosdep_db: Database to load definitions into, :class:`RosdepDatabase`
"""
self.rosdep_db = rosdep_db
self.loader = loader
self._view_cache = {} # {str: {RosdepView}}
self._resolve_cache = {} # {str : (os_name, os_version, installer_key, resolution, dependencies)}
# some APIs that deal with the entire environment save errors
# in to self.errors instead of raising them in order to be
# robust to single-stack faults.
self.errors = []
# flag for turning on printing to console
self.verbose = False
self.skipped_keys = []
def get_loader(self):
return self.loader
def get_errors(self):
"""
Retrieve error state for API calls that do not directly report
error state. This is the case for APIs like
:meth:`RosdepLookup.where_defined` that are meant to be
fault-tolerant to single-stack failures.
:returns: List of exceptions, ``[Exception]``
"""
return self.errors[:]
def get_rosdeps(self, resource_name, implicit=True):
"""
Get rosdeps that *resource_name* (e.g. package) requires.
:param implicit: If ``True``, include implicit rosdep
dependencies. Default: ``True``.
:returns: list of rosdep names, ``[str]``
"""
return self.loader.get_rosdeps(resource_name, implicit=implicit)
def get_resources_that_need(self, rosdep_name):
"""
:param rosdep_name: name of rosdep dependency
:returns: list of package names that require rosdep, ``[str]``
"""
return [k for k in self.loader.get_loadable_resources() if rosdep_name in self.get_rosdeps(k, implicit=False)]
@staticmethod
def create_from_rospkg(rospack=None, rosstack=None,
sources_loader=None,
verbose=False, dependency_types=None):
"""
Create :class:`RosdepLookup` based on current ROS package
environment.
:param rospack: (optional) Override :class:`rospkg.RosPack`
instance used to crawl ROS packages.
:param rosstack: (optional) Override :class:`rospkg.RosStack`
instance used to crawl ROS stacks.
:param sources_loader: (optional) Override SourcesLoader used
for managing sources.list data sources.
:param dependency_types: (optional) List of dependency types.
Allowed: {'build', 'buildtool', 'build_export', 'buildtool_export', 'exec', 'test', 'doc'}
"""
# initialize the loader
if rospack is None:
rospack = RosPack()
if rosstack is None:
rosstack = RosStack()
if sources_loader is None:
sources_loader = SourcesListLoader.create_default(verbose=verbose)
if dependency_types is None:
dependency_types = []
rosdep_db = RosdepDatabase()
# Use sources list to initialize rosdep_db. Underlay has no
# notion of specific resources, and its view keys are just the
# individual sources it can load from. SourcesListLoader
# cannot do delayed evaluation of OS setting due to matcher.
underlay_key = SourcesListLoader.ALL_VIEW_KEY
# Create the rospkg loader on top of the underlay
loader = RosPkgLoader(rospack=rospack, rosstack=rosstack,
underlay_key=underlay_key, dependency_types=dependency_types)
# create our actual instance
lookup = RosdepLookup(rosdep_db, loader)
# load in the underlay
lookup._load_all_views(loader=sources_loader)
# use dependencies to implement precedence
view_dependencies = sources_loader.get_loadable_views()
rosdep_db.set_view_data(underlay_key, {}, view_dependencies, underlay_key)
return lookup
def resolve_all(self, resources, installer_context, implicit=False):
"""
Resolve all the rosdep dependencies for *resources* using *installer_context*.
:param resources: list of resources (e.g. packages), ``[str]``
:param installer_context: :class:`InstallerContext`
:param implicit: Install implicit (recursive) dependencies of
resources. Default ``False``.
:returns: (resolutions, errors), ``([(str, [str])], {str: ResolutionError})``. resolutions provides
an ordered list of resolution tuples. A resolution tuple's first element is the installer
key (e.g.: apt or homebrew) and the second element is a list of opaque resolution values for that
installer. errors maps package names to an :exc:`ResolutionError` or :exc:`KeyError` exception.
:raises: :exc:`RosdepInternalError` if unexpected error in constructing dependency graph
:raises: :exc:`InvalidData` if a cycle occurs in constructing dependency graph
"""
depend_graph = DependencyGraph()
errors = {}
# TODO: resolutions dictionary should be replaced with resolution model instead of mapping (undefined) keys.
for resource_name in resources:
try:
rosdep_keys = self.get_rosdeps(resource_name, implicit=implicit)
if self.verbose:
print('resolve_all: resource [%s] requires rosdep keys [%s]' % (resource_name, ', '.join(rosdep_keys)), file=sys.stderr)
rosdep_keys = prune_catkin_packages(rosdep_keys, self.verbose)
rosdep_keys = prune_skipped_packages(rosdep_keys, self.skipped_keys, self.verbose)
for rosdep_key in rosdep_keys:
try:
installer_key, resolution, dependencies = \
self.resolve(rosdep_key, resource_name, installer_context)
depend_graph[rosdep_key]['installer_key'] = installer_key
depend_graph[rosdep_key]['install_keys'] = list(resolution)
depend_graph[rosdep_key]['dependencies'] = list(dependencies)
while dependencies:
depend_rosdep_key = dependencies.pop()
# prevent infinite loop
if depend_rosdep_key in depend_graph:
continue
installer_key, resolution, more_dependencies = \
self.resolve(depend_rosdep_key, resource_name, installer_context)
dependencies.extend(more_dependencies)
depend_graph[depend_rosdep_key]['installer_key'] = installer_key
depend_graph[depend_rosdep_key]['install_keys'] = list(resolution)
depend_graph[depend_rosdep_key]['dependencies'] = list(more_dependencies)
except ResolutionError as e:
errors[resource_name] = e
except ResourceNotFound as e:
errors[resource_name] = e
try:
# TODO: I really don't like AssertionErrors here; this should be modeled as 'CyclicGraphError'
# or something more explicit. No need to continue if this API errors.
resolutions_flat = depend_graph.get_ordered_dependency_list()
except AssertionError as e:
raise InvalidData('cycle in dependency graph detected: %s' % (e))
except KeyError as e:
raise RosdepInternalError(e)
return resolutions_flat, errors
def resolve(self, rosdep_key, resource_name, installer_context):
"""
Resolve a :class:`RosdepDefinition` for a particular
os/version spec.
:param resource_name: resource (e.g. ROS package) to resolve key within
:param rosdep_key: rosdep key to resolve
:param os_name: OS name to use for resolution
:param os_version: OS name to use for resolution
:returns: *(installer_key, resolution, dependencies)*, ``(str,
[opaque], [str])``. *resolution* are the system
dependencies for the specified installer. The value is an
opaque list and meant to be interpreted by the
installer. *dependencies* is a list of rosdep keys that the
definition depends on.
:raises: :exc:`ResolutionError` If *rosdep_key* cannot be resolved for *resource_name* in *installer_context*
:raises: :exc:`rospkg.ResourceNotFound` if *resource_name* cannot be located
"""
os_name, os_version = installer_context.get_os_name_and_version()
view = self.get_rosdep_view_for_resource(resource_name)
if view is None:
raise ResolutionError(rosdep_key, None, os_name, os_version, '[%s] does not have a rosdep view' % (resource_name))
try:
# print("KEYS", view.rosdep_defs.keys())
definition = view.lookup(rosdep_key)
except KeyError:
rd_debug(view)
raise ResolutionError(rosdep_key, None, os_name, os_version, 'Cannot locate rosdep definition for [%s]' % (rosdep_key))
# check cache: the main motivation for the cache is that
# source rosdeps are expensive to resolve
if rosdep_key in self._resolve_cache:
cache_value = self._resolve_cache[rosdep_key]
cache_os_name = cache_value[0]
cache_os_version = cache_value[1]
cache_view_name = cache_value[2]
if (
cache_os_name == os_name and
cache_os_version == os_version and
cache_view_name == view.name
):
return cache_value[3:]
# get the rosdep data for the platform
try:
installer_keys = installer_context.get_os_installer_keys(os_name)
default_key = installer_context.get_default_os_installer_key(os_name)
except KeyError:
raise ResolutionError(rosdep_key, definition.data, os_name, os_version, 'Unsupported OS [%s]' % (os_name))
installer_key, rosdep_args_dict = definition.get_rule_for_platform(os_name, os_version, installer_keys, default_key)
# resolve the rosdep data for the platform
try:
installer = installer_context.get_installer(installer_key)
except KeyError:
raise ResolutionError(rosdep_key, definition.data, os_name, os_version, 'Unsupported installer [%s]' % (installer_key))
resolution = installer.resolve(rosdep_args_dict)
dependencies = installer.get_depends(rosdep_args_dict)
# cache value
# the dependencies list is copied to prevent mutation before next cache hit
self._resolve_cache[rosdep_key] = os_name, os_version, view.name, installer_key, resolution, list(dependencies)
return installer_key, resolution, dependencies
def _load_all_views(self, loader):
"""
Load all available view keys. In general, this is equivalent
to loading all stacks on the package path. If
:exc:`InvalidData` errors occur while loading a view,
they will be saved in the *errors* field.
:param loader: override self.loader
:raises: :exc:`RosdepInternalError`
"""
for resource_name in loader.get_loadable_views():
try:
self._load_view_dependencies(resource_name, loader)
except ResourceNotFound as e:
self.errors.append(e)
except InvalidData as e:
self.errors.append(e)
def _load_view_dependencies(self, view_key, loader):
"""
Initialize internal :exc:`RosdepDatabase` on demand. Not
thread-safe.
:param view_key: name of view to load dependencies for.
:raises: :exc:`rospkg.ResourceNotFound` If view cannot be located
:raises: :exc:`InvalidData` if view's data is invaid
:raises: :exc:`RosdepInternalError`
"""
rd_debug('_load_view_dependencies[%s]' % (view_key))
db = self.rosdep_db
if db.is_loaded(view_key):
return
try:
loader.load_view(view_key, db, verbose=self.verbose)
entry = db.get_view_data(view_key)
rd_debug('_load_view_dependencies[%s]: %s' % (view_key, entry.view_dependencies))
for d in entry.view_dependencies:
self._load_view_dependencies(d, loader)
except InvalidData:
# mark view as loaded: as we are caching, the valid
# behavior is to not attempt loading this view ever
# again.
db.mark_loaded(view_key)
# re-raise
raise
except KeyError as e:
raise RosdepInternalError(e)
def create_rosdep_view(self, view_name, view_keys, verbose=False):
"""
:param view_name: name of view to create
:param view_keys: order list of view names to merge, first one wins
:param verbose: print debugging output
"""
# Create view and initialize with dbs from all of the
# dependencies.
view = RosdepView(view_name)
db = self.rosdep_db
for view_key in view_keys:
db_entry = db.get_view_data(view_key)
view.merge(db_entry, verbose=verbose)
if verbose:
print('View [%s], merged views:\n' % (view_name) + '\n'.join([' * %s' % view_key for view_key in view_keys]), file=sys.stderr)
return view
def get_rosdep_view_for_resource(self, resource_name, verbose=False):
"""
Get a :class:`RosdepView` for a specific ROS resource *resource_name*.
Views can be queries to resolve rosdep keys to
definitions.
:param resource_name: Name of ROS resource (e.g. stack,
package) to create view for, ``str``.
:returns: :class:`RosdepView` for specific ROS resource
*resource_name*, or ``None`` if no view is associated with this resource.
:raises: :exc:`RosdepConflict` if view cannot be created due
to conflict rosdep definitions.
:raises: :exc:`rospkg.ResourceNotFound` if *view_key* cannot be located
:raises: :exc:`RosdepInternalError`
"""
view_key = self.loader.get_view_key(resource_name)
if not view_key:
# NOTE: this may not be the right behavior and this happens
# for packages that are not in a stack.
return None
return self.get_rosdep_view(view_key, verbose=verbose)
def get_rosdep_view(self, view_key, verbose=False):
"""
Get a :class:`RosdepView` associated with *view_key*. Views
can be queries to resolve rosdep keys to definitions.
:param view_key: Name of rosdep view (e.g. ROS stack name), ``str``
:raises: :exc:`RosdepConflict` if view cannot be created due
to conflict rosdep definitions.
:raises: :exc:`rospkg.ResourceNotFound` if *view_key* cannot be located
:raises: :exc:`RosdepInternalError`
"""
if view_key in self._view_cache:
return self._view_cache[view_key]
# lazy-init
self._load_view_dependencies(view_key, self.loader)
# use dependencies to create view
try:
dependencies = self.rosdep_db.get_view_dependencies(view_key)
except KeyError as e:
# convert to ResourceNotFound. This should be decoupled
# in the future
raise ResourceNotFound(str(e.args[0]))
# load views in order
view = self.create_rosdep_view(view_key, dependencies + [view_key], verbose=verbose)
self._view_cache[view_key] = view
return view
def get_views_that_define(self, rosdep_name):
"""
Locate all views that directly define *rosdep_name*. A
side-effect of this method is that all available rosdep files
in the configuration will be loaded into memory.
Error state from single-stack failures
(e.g. :exc:`InvalidData`, :exc:`ResourceNotFound`) are
not propagated. Caller must check
:meth:`RosdepLookup.get_errors` to check for single-stack
error state. Error state does not reset -- it accumulates.
:param rosdep_name: name of rosdep to lookup
:returns: list of (stack_name, origin) where rosdep is defined.
:raises: :exc:`RosdepInternalError`
"""
# TODOXXX: change this to return errors object so that caller cannot ignore
self._load_all_views(self.loader)
db = self.rosdep_db
retval = []
for view_name in db.get_view_names():
entry = db.get_view_data(view_name)
# not much abstraction in the entry object
if rosdep_name in entry.rosdep_data:
retval.append((view_name, entry.origin))
return retval