message_loader_thread.py
Go to the documentation of this file.
1 # Software License Agreement (BSD License)
2 #
3 # Copyright (c) 2012, Willow Garage, Inc.
4 # All rights reserved.
5 #
6 # Redistribution and use in source and binary forms, with or without
7 # modification, are permitted provided that the following conditions
8 # are met:
9 #
10 # * Redistributions of source code must retain the above copyright
11 # notice, this list of conditions and the following disclaimer.
12 # * Redistributions in binary form must reproduce the above
13 # copyright notice, this list of conditions and the following
14 # disclaimer in the documentation and/or other materials provided
15 # with the distribution.
16 # * Neither the name of Willow Garage, Inc. nor the names of its
17 # contributors may be used to endorse or promote products derived
18 # from this software without specific prior written permission.
19 #
20 # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
21 # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
22 # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
23 # FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
24 # COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
25 # INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
26 # BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
27 # LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
28 # CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
29 # LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
30 # ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
31 # POSSIBILITY OF SUCH DAMAGE.
32 
33 import threading
34 
35 
36 class MessageLoaderThread(threading.Thread):
37 
38  """
39  Waits for a new playhead position on the given topic, then loads the message at that position and notifies the view threads.
40 
41  One thread per topic. Maintains a cache of recently loaded messages.
42  """
43 
44  def __init__(self, timeline, topic):
45  threading.Thread.__init__(self)
46 
47  self.timeline = timeline
48  self.topic = topic
49 
51 
53  self._message_cache = {}
55 
56  self._stop_flag = False
57 
58  self.setDaemon(True)
59  self.start()
60 
61  def reset(self):
62  self.bag_playhead_position = None
63 
64  def run(self):
65  while not self._stop_flag:
66  # Wait for a new entry
67  cv = self.timeline._playhead_positions_cvs[self.topic]
68  with cv:
69  while (self.topic not in self.timeline._playhead_positions) or (self.bag_playhead_position == self.timeline._playhead_positions[self.topic]):
70  cv.wait()
71  if self._stop_flag:
72  return
73  bag, playhead_position = self.timeline._playhead_positions[self.topic]
74 
75  self.bag_playhead_position = (bag, playhead_position)
76 
77  # Don't bother loading the message if there are no listeners
78  if not self.timeline.has_listeners(self.topic):
79  continue
80 
81  # Load the message
82  if playhead_position is None:
83  msg_data = None
84  else:
85  msg_data = self._get_message(bag, playhead_position)
86 
87  # Inform the views
88  messages_cv = self.timeline._messages_cvs[self.topic]
89  with messages_cv:
90  self.timeline._messages[self.topic] = (bag, msg_data)
91  messages_cv.notify_all() # notify all views that a message is loaded
92 
93  def _get_message(self, bag, position):
94  key = '%s%s' % (bag.filename, str(position))
95  if key in self._message_cache:
96  return self._message_cache[key]
97 
98  msg_data = self.timeline.read_message(bag, position)
99 
100  self._message_cache[key] = msg_data
101  self._message_cache_keys.append(key)
102 
103  if len(self._message_cache) > self._message_cache_capacity:
104  oldest_key = self._message_cache_keys[0]
105  del self._message_cache[oldest_key]
106  self._message_cache_keys.remove(oldest_key)
107 
108  return msg_data
109 
110  def stop(self):
111  self._stop_flag = True
112  cv = self.timeline._playhead_positions_cvs[self.topic]
113  with cv:
114  cv.notify_all()
rqt_bag.message_loader_thread.MessageLoaderThread.topic
topic
Definition: message_loader_thread.py:48
rqt_bag.message_loader_thread.MessageLoaderThread._message_cache_keys
_message_cache_keys
Definition: message_loader_thread.py:54
rqt_bag.message_loader_thread.MessageLoaderThread._stop_flag
_stop_flag
Definition: message_loader_thread.py:56
rqt_bag.message_loader_thread.MessageLoaderThread.stop
def stop(self)
Definition: message_loader_thread.py:110
rqt_bag.message_loader_thread.MessageLoaderThread.run
def run(self)
Definition: message_loader_thread.py:64
rqt_bag.message_loader_thread.MessageLoaderThread._message_cache_capacity
_message_cache_capacity
Definition: message_loader_thread.py:52
rqt_bag.message_loader_thread.MessageLoaderThread._get_message
def _get_message(self, bag, position)
Definition: message_loader_thread.py:93
rqt_bag.message_loader_thread.MessageLoaderThread
Definition: message_loader_thread.py:36
rqt_bag.message_loader_thread.MessageLoaderThread.timeline
timeline
Definition: message_loader_thread.py:47
rqt_bag.message_loader_thread.MessageLoaderThread.__init__
def __init__(self, timeline, topic)
Definition: message_loader_thread.py:44
rqt_bag.message_loader_thread.MessageLoaderThread._message_cache
_message_cache
Definition: message_loader_thread.py:53
rqt_bag.message_loader_thread.MessageLoaderThread.bag_playhead_position
bag_playhead_position
Definition: message_loader_thread.py:50
rqt_bag.message_loader_thread.MessageLoaderThread.reset
def reset(self)
Definition: message_loader_thread.py:61


rqt_bag
Author(s): Dirk Thomas , Aaron Blasdel , Austin Hendrix , Tim Field
autogenerated on Thu Mar 2 2023 03:43:15