本文整理汇总了Python中unique_id.toMsg函数的典型用法代码示例。如果您正苦于以下问题：Python toMsg函数的具体用法？Python toMsg怎么用？Python toMsg使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了toMsg函数的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_reject
def test_reject(self):
rq1 = ResourceReply(Request(id=unique_id.toMsg(TEST_UUID),
resources=[TEST_RESOURCE],
status=Request.NEW))
rq1.reject()
self.assertEqual(rq1.msg.status, Request.REJECTED)
rq2 = ResourceReply(Request(id=unique_id.toMsg(TEST_UUID),
resources=[TEST_RESOURCE],
status=Request.WAITING))
rq2.reject()
self.assertEqual(rq2.msg.status, Request.REJECTED)
rq3 = ResourceReply(Request(id=unique_id.toMsg(TEST_UUID),
resources=[TEST_RESOURCE],
status=Request.PREEMPTING))
rq3.reject()
self.assertEqual(rq3.msg.status, Request.REJECTED)
rq4 = ResourceReply(Request(id=unique_id.toMsg(TEST_UUID),
resources=[TEST_RESOURCE],
status=Request.PREEMPTED))
rq4.reject()
self.assertEqual(rq4.msg.status, Request.REJECTED)
rq5 = ResourceReply(Request(id=unique_id.toMsg(TEST_UUID),
resources=[TEST_WILDCARD],
status=Request.GRANTED))
self.assertRaises(TransitionError, rq5.reject)
开发者ID:dykim723，项目名称:rocon_scheduler_requests，代码行数:29，代码来源:test_transitions.py
示例2: test_one_request_set
def test_one_request_set(self):
msg1 = Request(id=unique_id.toMsg(TEST_UUID),
resources=[TEST_WILDCARD],
status=Request.NEW)
rset = RequestSet([msg1], RQR_UUID, contents=ResourceReply)
self.assertEqual(len(rset), 1)
self.assertTrue(TEST_UUID in rset)
self.assertEqual(rset[TEST_UUID].msg, msg1)
self.assertEqual(rset.get(TEST_UUID), rset[TEST_UUID])
self.assertFalse(DIFF_UUID in rset)
self.assertIsNone(rset.get(DIFF_UUID))
self.assertEqual(rset.get(DIFF_UUID, 10), 10)
self.assertTrue(rset == RequestSet([msg1], RQR_UUID))
self.assertTrue(rset == RequestSet([msg1], RQR_UUID,
contents=ResourceReply))
rs2 = copy.deepcopy(rset)
self.assertTrue(rset == rs2)
rset_str = """requester_id: 01234567-89ab-cdef-0123-456789abcdef
requests:
id: 01234567-89ab-cdef-fedc-ba9876543210
priority: 0
resources:
linux.precise.ros.segbot.*/test_rapp
status: 0"""
self.assertEqual(str(rset), rset_str)
sch_msg = SchedulerRequests(requester=unique_id.toMsg(RQR_UUID),
requests=[msg1])
self.assertEqual(rset.to_msg(stamp=rospy.Time()), sch_msg)
开发者ID:dykim723，项目名称:rocon_scheduler_requests，代码行数:30，代码来源:test_transitions.py
示例3: test_constructor
def test_constructor(self):
msg1 = Request(id=unique_id.toMsg(TEST_UUID),
resources=[TEST_WILDCARD],
status=Request.NEW)
rq1 = ResourceRequest(msg1)
self.assertIsNotNone(rq1)
self.assertEqual(str(rq1),
'id: 01234567-89ab-cdef-fedc-ba9876543210\n'
' priority: 0\n'
' resources: \n'
' linux.precise.ros.segbot.*/test_rapp\n'
' status: 0')
self.assertEqual(rq1.msg.status, Request.NEW)
self.assertEqual(rq1.msg.resources, [TEST_WILDCARD])
# why is this broken???
#self.assertEqual(rq1.get_uuid, TEST_UUID)
rq2 = ResourceRequest(Request(id=unique_id.toMsg(DIFF_UUID),
resources=[TEST_RESOURCE],
status=Request.NEW))
self.assertEqual(rq2.msg.status, Request.NEW)
self.assertEqual(rq2.msg.resources, [TEST_RESOURCE])
self.assertEqual(rq2.get_uuid(), DIFF_UUID)
self.assertEqual(str(rq2),
'id: 01234567-cdef-fedc-89ab-ba9876543210\n'
' priority: 0\n'
' resources: \n'
' linux.precise.ros.segbot.roberto/test_rapp\n'
' status: 0')
开发者ID:dykim723，项目名称:rocon_scheduler_requests，代码行数:30，代码来源:test_transitions.py
示例4: test_constructor
def test_constructor(self):
msg1 = Request(id=unique_id.toMsg(TEST_UUID),
resources=[TEST_WILDCARD],
status=Request.NEW)
rq1 = ResourceRequest(msg1)
self.assertIsNotNone(rq1)
self.assertEqual(str(rq1),
"""id: 01234567-89ab-cdef-fedc-ba9876543210
priority: 0
resources:
rocon:/segbot#test_rapp
status: 0""")
self.assertEqual(rq1.msg.status, Request.NEW)
self.assertEqual(rq1.msg.resources, [TEST_WILDCARD])
self.assertEqual(rq1.uuid, TEST_UUID)
rq2 = ResourceRequest(Request(id=unique_id.toMsg(DIFF_UUID),
resources=[TEST_RESOURCE],
status=Request.NEW))
self.assertEqual(rq2.msg.status, Request.NEW)
self.assertEqual(rq2.msg.resources, [TEST_RESOURCE])
self.assertEqual(rq2.uuid, DIFF_UUID)
self.assertEqual(str(rq2),
"""id: 01234567-cdef-fedc-89ab-ba9876543210
priority: 0
resources:
rocon:/segbot/roberto#test_rapp
status: 0""")
开发者ID:jihoonl，项目名称:concert_scheduling，代码行数:28，代码来源:test_transitions.py
示例5: test_canceled_merge_plus_new_request
def test_canceled_merge_plus_new_request(self):
msg1 = Request(id=unique_id.toMsg(TEST_UUID),
resources=[TEST_RESOURCE],
status=Request.CANCELING)
msg2 = Request(id=unique_id.toMsg(DIFF_UUID),
resources=[TEST_WILDCARD],
status=Request.NEW)
rset = RequestSet([msg1, msg2], RQR_UUID)
self.assertEqual(len(rset), 2)
self.assertIn(TEST_UUID, rset)
self.assertIn(DIFF_UUID, rset)
self.assertEqual(rset[DIFF_UUID].msg.status, Request.NEW)
# merge a canceled request: TEST_UUID should be deleted, but
# DIFF_UUID should not
msg3 = Request(id=unique_id.toMsg(TEST_UUID),
resources=[TEST_RESOURCE],
status=Request.CLOSED)
rel_rset = RequestSet([msg3], RQR_UUID)
rset.merge(rel_rset)
self.assertEqual(len(rset), 1)
self.assertNotIn(TEST_UUID, rset)
self.assertIn(DIFF_UUID, rset)
self.assertEqual(rset[DIFF_UUID].msg.status, Request.NEW)
# make a fresh object like the original msg2 for comparison
msg4 = Request(id=unique_id.toMsg(DIFF_UUID),
resources=[TEST_WILDCARD],
status=Request.NEW)
self.assertEqual(rset, RequestSet([msg4], RQR_UUID))
self.assertNotEqual(rset, rel_rset)
开发者ID:jihoonl，项目名称:concert_scheduling，代码行数:31，代码来源:test_transitions.py
示例6: test_abort
def test_abort(self):
rq1 = ResourceReply(Request(id=unique_id.toMsg(TEST_UUID),
resources=[TEST_RESOURCE],
status=Request.NEW))
rq1.abort()
self.assertEqual(rq1.msg.status, Request.ABORTED)
rq2 = ResourceReply(Request(id=unique_id.toMsg(TEST_UUID),
resources=[TEST_RESOURCE],
status=Request.WAITING))
rq2.abort()
self.assertEqual(rq2.msg.status, Request.ABORTED)
rq3 = ResourceReply(Request(id=unique_id.toMsg(TEST_UUID),
resources=[TEST_RESOURCE],
status=Request.PREEMPTING))
rq3.abort()
self.assertEqual(rq3.msg.status, Request.ABORTED)
rq4 = ResourceReply(Request(id=unique_id.toMsg(TEST_UUID),
resources=[TEST_RESOURCE],
status=Request.PREEMPTED))
rq4.abort()
self.assertEqual(rq4.msg.status, Request.ABORTED)
rq5 = ResourceReply(Request(id=unique_id.toMsg(TEST_UUID),
resources=[TEST_WILDCARD],
status=Request.GRANTED))
rq5.abort()
self.assertEqual(rq5.msg.status, Request.ABORTED)
开发者ID:dykim723，项目名称:rocon_scheduler_requests，代码行数:30，代码来源:test_transitions.py
示例7: filter_by
def filter_by(self, world, uuids=[], names=[], types=[], keywords=[], relationships=[]):
'''
Reload annotations collection, filtered by new selection criteria.
@param world: Annotations in this collection belong to this world.
@param uuids: Filter annotations by their uuid.
@param names: Filter annotations by their name.
@param types: Filter annotations by their type.
@param keywords: Filter annotations by their keywords.
@param relationships: Filter annotations by their relationships.
@raise WCFError: If something went wrong.
'''
rospy.loginfo("Getting annotations for world '%s' and additional filter criteria", world)
get_anns_srv = self._get_service_handle('get_annotations', world_canvas_msgs.srv.GetAnnotations)
ids = [unique_id.toMsg(uuid.UUID('urn:uuid:' + id)) for id in uuids]
relationships = [unique_id.toMsg(uuid.UUID('urn:uuid:' + id)) for id in relationships]
response = get_anns_srv(world, ids, names, types, keywords, relationships)
if response.result:
if len(response.annotations) > 0:
rospy.loginfo("%d annotations found", len(response.annotations))
self._annotations = response.annotations
else:
rospy.loginfo("No annotations found for world '%s' with the given search criteria", world)
del self._annotations[:]
# All non-saved information gets invalidated after reload the collection
del self._annots_data[:]
del self._annots_to_delete[:]
self._saved = True
else:
message = "Server reported an error: %s" % response.message
rospy.logerr(message)
raise WCFError(message)
开发者ID:corot，项目名称:world_canvas_libs，代码行数:35，代码来源:annotation_collection.py
示例8: read
def read(filename):
yaml_data = None
with open(filename) as f:
yaml_data = yaml.load(f)
anns_list = []
data_list = []
for t in yaml_data:
ann = Annotation()
ann.timestamp = rospy.Time.now()
ann.world_id = unique_id.toMsg(uuid.UUID('urn:uuid:' + world_id))
ann.data_id = unique_id.toMsg(unique_id.fromRandom())
ann.id = unique_id.toMsg(unique_id.fromRandom())
ann.name = t['name']
ann.type = 'yocs_msgs/Wall'
for i in range(0, random.randint(0,11)):
ann.keywords.append('kw'+str(random.randint(1,11)))
if 'prev_id' in vars():
ann.relationships.append(prev_id)
prev_id = ann.id
ann.shape = 1 # CUBE
ann.color.r = 0.8
ann.color.g = 0.2
ann.color.b = 0.2
ann.color.a = 0.4
ann.size.x = float(t['width'])
ann.size.y = float(t['length'])
ann.size.z = float(t['height'])
ann.pose.header.frame_id = t['frame_id']
ann.pose.header.stamp = rospy.Time.now()
ann.pose.pose.pose = message_converter.convert_dictionary_to_ros_message('geometry_msgs/Pose',t['pose'])
# walls are assumed to lay on the floor, so z coordinate is zero;
# but WCF assumes that the annotation pose is the center of the object
ann.pose.pose.pose.position.z += ann.size.z/2.0
anns_list.append(ann)
object = Wall()
object.name = t['name']
object.length = float(t['length'])
object.width = float(t['width'])
object.height = float(t['height'])
object.pose.header.frame_id = t['frame_id']
object.pose.header.stamp = rospy.Time.now()
object.pose.pose.pose = message_converter.convert_dictionary_to_ros_message('geometry_msgs/Pose',t['pose'])
data = AnnotationData()
data.id = ann.data_id
data.data = pickle.dumps(object)
data_list.append(data)
print ann, object, data
return anns_list, data_list
开发者ID:stonier，项目名称:world_canvas，代码行数:56，代码来源:save_walls.py
示例9: read
def read(file):
yaml_data = None
with open(file) as f:
yaml_data = yaml.load(f)
anns_list = []
data_list = []
for t in yaml_data:
ann = Annotation()
ann.timestamp = rospy.Time.now()
ann.data_id = unique_id.toMsg(unique_id.fromRandom())
ann.id = unique_id.toMsg(unique_id.fromRandom())
ann.world = world
ann.name = t['name']
ann.type = 'yocs_msgs/Table'
for i in range(0, random.randint(0,11)):
ann.keywords.append('kw'+str(random.randint(1,11)))
if 'prev_id' in vars():
ann.relationships.append(prev_id)
prev_id = ann.id
ann.shape = 3 #CYLINDER
ann.color.r = 0.2
ann.color.g = 0.2
ann.color.b = 0.8
ann.color.a = 0.5
ann.size.x = float(t['radius'])*2
ann.size.y = float(t['radius'])*2
ann.size.z = float(t['height'])
ann.pose.header.frame_id = t['frame_id']
ann.pose.header.stamp = rospy.Time.now()
ann.pose.pose.pose = message_converter.convert_dictionary_to_ros_message('geometry_msgs/Pose',t['pose'])
# tables are assumed to lay on the floor, so z coordinate is zero;
# but WCF assumes that the annotation pose is the center of the object
ann.pose.pose.pose.position.z += ann.size.z/2.0
anns_list.append(ann)
object = Table()
object.name = t['name']
object.radius = float(t['radius'])
object.height = float(t['height'])
object.pose.header.frame_id = t['frame_id']
object.pose.header.stamp = rospy.Time.now()
object.pose.pose.pose = message_converter.convert_dictionary_to_ros_message('geometry_msgs/Pose',t['pose'])
data = AnnotationData()
data.id = ann.data_id
data.type = ann.type
data.data = serialize_msg(object)
data_list.append(data)
return anns_list, data_list
开发者ID:corot，项目名称:world_canvas，代码行数:54，代码来源:save_tables.py
示例10: read
def read(filename):
yaml_data = None
with open(filename) as f:
yaml_data = yaml.load(f)
anns_list = []
data_list = []
for t in yaml_data:
ann = Annotation()
ann.timestamp = rospy.Time.now()
ann.world_id = unique_id.toMsg(uuid.UUID('urn:uuid:' + world_id))
ann.data_id = unique_id.toMsg(unique_id.fromRandom())
ann.id = unique_id.toMsg(unique_id.fromRandom())
ann.name = t['name']
ann.type = 'ar_track_alvar/AlvarMarker'
for i in range(0, random.randint(0,11)):
ann.keywords.append('kw'+str(random.randint(1,11)))
# if 'prev_id' in vars():
# ann.relationships.append(prev_id)
# prev_id = ann.id
ann.shape = 1 # CUBE
ann.color.r = 1.0
ann.color.g = 1.0
ann.color.b = 1.0
ann.color.a = 1.0
ann.size.x = 0.18
ann.size.y = 0.18
ann.size.z = 0.01
ann.pose.header.frame_id = t['frame_id']
ann.pose.header.stamp = rospy.Time.now()
ann.pose.pose.pose = message_converter.convert_dictionary_to_ros_message('geometry_msgs/Pose',t['pose'])
anns_list.append(ann)
object = AlvarMarker()
object.id = t['id']
object.confidence = t['confidence']
object.pose.header.frame_id = t['frame_id']
object.pose.header.stamp = rospy.Time.now()
object.pose.pose = message_converter.convert_dictionary_to_ros_message('geometry_msgs/Pose',t['pose'])
data = AnnotationData()
data.id = ann.data_id
data.data = pickle.dumps(object)
data_list.append(data)
print ann, object, data
return anns_list, data_list
开发者ID:stonier，项目名称:world_canvas，代码行数:50，代码来源:save_markers.py
示例11: __call__
def __call__(self, msg, timeout=None, callback=None, error_callback=None):
'''
Initiates and executes the client request to the server. The type of arguments
supplied determines whether to apply blocking or non-blocking behaviour.
If no callback is supplied, the mode is blocking, otherwise non-blocking.
If no timeout is specified, then a return of None indicates that the
operation timed out.
:param msg: the request message
:type msg: <name>Request
:param rospy.Duration timeout: time to wait for data
:param callback: user callback invoked for responses of non-blocking calls
:type callback: method with arguments (uuid_msgs.UniqueID, <name>Response)
:returns: msg/id for blocking calls it is the response message, for non-blocking it is the unique id
:rtype: <name>Response/uuid_msgs.UniqueID or None (if timed out)
'''
pair_request_msg = self.ServicePairRequest()
pair_request_msg.id = unique_id.toMsg(unique_id.fromRandom())
pair_request_msg.request = msg
key = unique_id.toHexString(pair_request_msg.id)
if callback == None and error_callback == None:
self._request_handlers[key] = BlockingRequestHandler(key)
return self._make_blocking_call(self._request_handlers[key], pair_request_msg, timeout)
else:
request_handler = NonBlockingRequestHandler(key, callback, error_callback)
self._request_handlers[key] = request_handler.copy()
self._make_non_blocking_call(request_handler, pair_request_msg, timeout)
return pair_request_msg.id
开发者ID:suryaprakaz，项目名称:rospy_android，代码行数:32，代码来源:service_pair_client.py
示例12: test_empty_merge
def test_empty_merge(self):
msg1 = Request(id=unique_id.toMsg(TEST_UUID),
resources=[TEST_WILDCARD],
status=Request.NEW)
rset = RequestSet([msg1], RQR_UUID)
self.assertEqual(len(rset), 1)
self.assertIn(TEST_UUID, rset)
sch_msg = SchedulerRequests(requester=unique_id.toMsg(RQR_UUID),
requests=[msg1])
self.assertEqual(rset.to_msg(stamp=rospy.Time()), sch_msg)
# merge an empty request set: rset should remain the same
rset.merge(RequestSet([], RQR_UUID, contents=ActiveRequest))
self.assertEqual(len(rset), 1)
self.assertIn(TEST_UUID, rset)
self.assertEqual(rset.to_msg(stamp=rospy.Time()), sch_msg)
开发者ID:jihoonl，项目名称:concert_scheduling，代码行数:16，代码来源:test_transitions.py
示例13: test_allocate_permutation_two_resources
def test_allocate_permutation_two_resources(self):
""" Request a regexp allocation followed by an exact
allocation. Initially the exact resource gets assigned to the
regexp, so the second part of the request fails. The
allocator must try the other permutation for it to succeed.
"""
pool = ResourcePool(KnownResources(resources=[
CurrentStatus(uri=MARVIN_NAME,
rapps={TELEOP_RAPP, EXAMPLE_RAPP}),
CurrentStatus(uri=ROBERTO_NAME,
rapps={TELEOP_RAPP})]))
rq = ActiveRequest(Request(
id=unique_id.toMsg(RQ_UUID),
resources=[Resource(rapp=TELEOP_RAPP, uri=ANY_NAME),
Resource(rapp=EXAMPLE_RAPP, uri=MARVIN_NAME)]))
alloc = pool.allocate(rq)
self.assertTrue(alloc)
self.assertEqual(len(alloc), 2)
self.assertEqual(pool[MARVIN_NAME].status, CurrentStatus.ALLOCATED)
self.assertEqual(pool[MARVIN_NAME].owner, RQ_UUID)
self.assertEqual(pool[ROBERTO_NAME].status, CurrentStatus.ALLOCATED)
self.assertEqual(pool[ROBERTO_NAME].owner, RQ_UUID)
self.assertEqual(alloc[0],
Resource(rapp=TELEOP_RAPP, uri=ROBERTO_NAME))
self.assertEqual(alloc[1],
Resource(rapp=EXAMPLE_RAPP, uri=MARVIN_NAME))
开发者ID:jihoonl，项目名称:concert_scheduling，代码行数:26，代码来源:test_resource_pool.py
示例14: test_validate
def test_validate(self):
rq1 = ResourceRequest(Request(id=unique_id.toMsg(TEST_UUID),
resources=[TEST_RESOURCE],
status=Request.NEW))
self.assertTrue(rq1._validate(Request.GRANTED))
self.assertTrue(rq1._validate(Request.PREEMPTING))
self.assertFalse(rq1._validate(Request.CLOSED))
开发者ID:jihoonl，项目名称:concert_scheduling，代码行数:7，代码来源:test_transitions.py
示例15: current_status
def current_status(self):
""" :returns: ``scheduler_msgs/CurrentStatus`` for this resource. """
msg = CurrentStatus(uri=self.uri, status=self.status, rapps=list(self.rapps))
if self.status == CurrentStatus.ALLOCATED:
msg.owner = unique_id.toMsg(self.owner)
msg.priority = self.priority
return msg
开发者ID:utexas-bwi，项目名称:concert_scheduling，代码行数:7，代码来源:resource_pool.py
示例16: test_matching_allocation_one_resource
def test_matching_allocation_one_resource(self):
pool = ResourcePool(SINGLETON_POOL)
self.assertEqual(len(pool), 1)
self.assertEqual(pool.known_resources(), SINGLETON_POOL)
self.assertFalse(pool.changed)
res = Resource(rapp=TELEOP_RAPP, uri=ANY_NAME)
subset = pool._match_subset(res, {CurrentStatus.AVAILABLE})
self.assertNotIn(MARVIN_NAME, subset)
self.assertIn(ROBERTO_NAME, subset)
self.assertEqual(subset, set([ROBERTO_NAME]))
self.assertEqual(pool.match_list([ROBERTO_RESOURCE],
{CurrentStatus.AVAILABLE}),
[set([ROBERTO_NAME])])
rq = copy.deepcopy(ANY_REQUEST)
alloc = pool.allocate(rq)
self.assertTrue(alloc)
self.assertTrue(pool.changed)
self.assertEqual(alloc[0], ROBERTO_RESOURCE)
self.assertEqual(pool[ROBERTO_NAME].status, CurrentStatus.ALLOCATED)
self.assertEqual(pool[ROBERTO_NAME].owner, RQ_UUID)
self.assertTrue(pool.changed)
self.assertEqual(
pool.known_resources(),
KnownResources(resources=[
CurrentStatus(uri=ROBERTO_NAME, rapps=TEST_RAPPS,
status=CurrentStatus.ALLOCATED,
owner=unique_id.toMsg(RQ_UUID))]))
self.assertFalse(pool.changed)
开发者ID:jihoonl，项目名称:concert_scheduling，代码行数:28，代码来源:test_resource_pool.py
示例17: new_request
def new_request(self, resources, priority=None, uuid=None):
""" Add a new scheduler request.
Call this method for each desired new request, then invoke
:py:meth:`.send_requests` to notify the scheduler.
:param resources: ROCON resources requested
:type resources: list of scheduler_msgs/Resource
:param priority: Scheduling priority of this request. If
``None`` provided, use this requester's priority.
:type priority: int
:param uuid: UUID_ of this request. If ``None`` provided, a
random UUID will be assigned.
:type uuid: :class:`uuid.UUID` or ``None``
:returns: UUID (:class:`uuid.UUID`) assigned.
:raises: :exc:`.WrongRequestError` if request already exists.
"""
if priority is None:
priority = self.priority
if uuid is None:
uuid = unique_id.fromRandom()
if uuid in self.rset:
raise WrongRequestError('UUID already in use.')
msg = Request(id=unique_id.toMsg(uuid),
priority=priority,
resources=resources,
status=Request.NEW)
self.rset[uuid] = ResourceRequest(msg)
return uuid
开发者ID:stonier，项目名称:rocon_scheduler_requests，代码行数:32，代码来源:requester.py
示例18: test_one_request_set
def test_one_request_set(self):
msg1 = Request(id=unique_id.toMsg(TEST_UUID),
resources=[TEST_WILDCARD],
status=Request.NEW)
rset = RequestSet([msg1], RQR_UUID, contents=ActiveRequest)
self.assertEqual(len(rset), 1)
self.assertIn(TEST_UUID, rset)
self.assertEqual(rset[TEST_UUID].msg, msg1)
self.assertEqual(rset.get(TEST_UUID), rset[TEST_UUID])
self.assertNotIn(DIFF_UUID, rset)
self.assertIsNone(rset.get(DIFF_UUID))
self.assertEqual(rset.get(DIFF_UUID, 10), 10)
self.assertEqual(rset, RequestSet([msg1], RQR_UUID))
self.assertEqual(rset, RequestSet([msg1], RQR_UUID,
contents=ActiveRequest))
rset_str = """requester_id: 01234567-89ab-cdef-0123-456789abcdef
requests:
id: 01234567-89ab-cdef-fedc-ba9876543210
priority: 0
resources:
rocon:/segbot#test_rapp
status: 0"""
self.assertEqual(str(rset), rset_str)
sch_msg = SchedulerRequests(requester=unique_id.toMsg(RQR_UUID),
requests=[msg1])
self.assertEqual(rset.to_msg(stamp=rospy.Time()), sch_msg)
# vary message contents:
rs2 = copy.deepcopy(rset)
self.assertEqual(rset, rs2)
rq = rs2.get(TEST_UUID)
rq.msg.priority = 10
self.assertNotEqual(rset, rs2)
rq.msg.priority = 0
self.assertEqual(rset, rs2)
rq.msg.availability = rospy.Time(1000.0)
self.assertNotEqual(rset, rs2)
rq.msg.availability = rospy.Time()
self.assertEqual(rset, rs2)
rq.msg.hold_time = rospy.Duration(1000.0)
self.assertNotEqual(rset, rs2)
rq.msg.hold_time = rospy.Duration()
self.assertEqual(rset, rs2)
rq.msg.resources = [TEST_RESOURCE]
self.assertNotEqual(rset, rs2)
rq.msg.resources = [TEST_WILDCARD]
self.assertEqual(rset, rs2)
开发者ID:jihoonl，项目名称:concert_scheduling，代码行数:47，代码来源:test_transitions.py
示例19: test_closed_merge
def test_closed_merge(self):
msg1 = Request(id=unique_id.toMsg(TEST_UUID),
resources=[TEST_WILDCARD],
status=Request.CLOSED)
rset = RequestSet([msg1], RQR_UUID, contents=ActiveRequest)
self.assertEqual(len(rset), 1)
self.assertIn(TEST_UUID, rset)
sch_msg = SchedulerRequests(requester=unique_id.toMsg(RQR_UUID),
requests=[msg1])
self.assertEqual(rset.to_msg(stamp=rospy.Time()), sch_msg)
# merge an empty request set: TEST_UUID should be deleted
empty_rset = RequestSet([], RQR_UUID)
rset.merge(empty_rset)
self.assertEqual(len(rset), 0)
self.assertNotIn(TEST_UUID, rset)
self.assertNotEqual(rset.to_msg(stamp=rospy.Time()), sch_msg)
self.assertEqual(rset, empty_rset)
开发者ID:jihoonl，项目名称:concert_scheduling，代码行数:18，代码来源:test_transitions.py
示例20: test_insufficient_resources
def test_insufficient_resources(self):
# ask for two when there's only one
pool = ResourcePool(KnownResources(resources=[ROBERTO]))
rq1 = ActiveRequest(Request(
id=unique_id.toMsg(RQ_UUID),
resources=[ANY_RESOURCE, ANY_RESOURCE]))
alloc1 = pool.allocate(rq1)
self.assertFalse(alloc1)
self.assertEqual(pool[ROBERTO_NAME].status, CurrentStatus.AVAILABLE)
self.assertIsNone(pool[ROBERTO_NAME].owner)
开发者ID:jihoonl，项目名称:concert_scheduling，代码行数:10，代码来源:test_resource_pool.py
注：本文中的unique_id.toMsg函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论