We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
unregister() method to message_filter.Subscriber on python
Do you have plan to implement this?
Or there is other way to achieve this?


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `utilities/message_filters/src/message_filters/__init__.py`
Content:
```
1 # Copyright (c) 2009, Willow Garage, Inc.
2 # All rights reserved.
3 #
4 # Redistribution and use in source and binary forms, with or without
5 # modification, are permitted provided that the following conditions are met:
6 #
7 #     * Redistributions of source code must retain the above copyright
8 #       notice, this list of conditions and the following disclaimer.
9 #     * Redistributions in binary form must reproduce the above copyright
10 #       notice, this list of conditions and the following disclaimer in the
11 #       documentation and/or other materials provided with the distribution.
12 #     * Neither the name of the Willow Garage, Inc. nor the names of its
13 #       contributors may be used to endorse or promote products derived from
14 #       this software without specific prior written permission.
15 #
16 # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
17 # AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
18 # IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
19 # ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
20 # LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
21 # CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
22 # SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
23 # INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
24 # CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
25 # ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
26 # POSSIBILITY OF SUCH DAMAGE.
27 
28 """
29 Message Filter Objects
30 ======================
31 """
32 
33 import itertools
34 import threading
35 import rospy
36 
37 class SimpleFilter:
38 
39     def __init__(self):
40         self.callbacks = {}
41 
42     def registerCallback(self, cb, *args):
43         """
44         Register a callback function `cb` to be called when this filter
45         has output.
46         The filter calls the function ``cb`` with a filter-dependent list of arguments,
47         followed by the call-supplied arguments ``args``.
48         """
49 
50         conn = len(self.callbacks)
51         self.callbacks[conn] = (cb, args)
52         return conn
53 
54     def signalMessage(self, *msg):
55         for (cb, args) in self.callbacks.values():
56             cb(*(msg + args))
57 
58 class Subscriber(SimpleFilter):
59     
60     """
61     ROS subscription filter.  Identical arguments as :class:`rospy.Subscriber`.
62 
63     This class acts as a highest-level filter, simply passing messages
64     from a ROS subscription through to the filters which have connected
65     to it.
66     """
67     def __init__(self, *args, **kwargs):
68         SimpleFilter.__init__(self)
69         self.topic = args[0]
70         kwargs['callback'] = self.callback
71         self.sub = rospy.Subscriber(*args, **kwargs)
72 
73     def callback(self, msg):
74         self.signalMessage(msg)
75 
76     def getTopic(self):
77         return self.topic
78 
79 class Cache(SimpleFilter):
80 
81     """
82     Stores a time history of messages.
83 
84     Given a stream of messages, the most recent ``cache_size`` messages
85     are cached in a ring buffer, from which time intervals of the cache
86     can then be retrieved by the client.
87     """
88 
89     def __init__(self, f, cache_size = 1):
90         SimpleFilter.__init__(self)
91         self.connectInput(f)
92         self.cache_size = cache_size
93         # Array to store messages
94         self.cache_msgs = []
95         # Array to store msgs times, auxiliary structure to facilitate
96         # sorted insertion
97         self.cache_times = []
98 
99     def connectInput(self, f):
100         self.incoming_connection = f.registerCallback(self.add)
101 
102     def add(self, msg):
103         # Cannot use message filters with non-stamped messages
104         if not hasattr(msg, 'header') or not hasattr(msg.header, 'stamp'):
105             rospy.logwarn("Cannot use message filters with non-stamped messages")
106             return
107 
108         # Insert sorted
109         stamp = msg.header.stamp
110         self.cache_times.append(stamp)
111         self.cache_msgs.append(msg)
112 
113         # Implement a ring buffer, discard older if oversized
114         if (len(self.cache_msgs) > self.cache_size):
115             del self.cache_msgs[0]
116             del self.cache_times[0]
117 
118         # Signal new input
119         self.signalMessage(msg)
120 
121     def getInterval(self, from_stamp, to_stamp):
122         """Query the current cache content between from_stamp to to_stamp."""
123         assert from_stamp <= to_stamp
124         return [m for m in self.cache_msgs
125                 if m.header.stamp >= from_stamp and m.header.stamp <= to_stamp]
126 
127     def getElemAfterTime(self, stamp):
128         """Return the oldest element after or equal the passed time stamp."""
129         newer = [m for m in self.cache_msgs if m.header.stamp >= stamp]
130         if not newer:
131             return None
132         return newer[0]
133 
134     def getElemBeforeTime(self, stamp):
135         """Return the newest element before or equal the passed time stamp."""
136         older = [m for m in self.cache_msgs if m.header.stamp <= stamp]
137         if not older:
138             return None
139         return older[-1]
140 
141     def getLastestTime(self):
142         """Return the newest recorded timestamp."""
143         if not self.cache_times:
144             return None
145         return self.cache_times[-1]
146 
147     def getOldestTime(self):
148         """Return the oldest recorded timestamp."""
149         if not self.cache_times:
150             return None
151         return self.cache_times[0]
152 
153 
154 class TimeSynchronizer(SimpleFilter):
155 
156     """
157     Synchronizes messages by their timestamps.
158 
159     :class:`TimeSynchronizer` synchronizes incoming message filters by the
160     timestamps contained in their messages' headers. TimeSynchronizer
161     listens on multiple input message filters ``fs``, and invokes the callback
162     when it has a collection of messages with matching timestamps.
163 
164     The signature of the callback function is::
165 
166         def callback(msg1, ... msgN):
167 
168     where N is the number of input message filters, and each message is
169     the output of the corresponding filter in ``fs``.
170     The required ``queue size`` parameter specifies how many sets of
171     messages it should store from each input filter (by timestamp)
172     while waiting for messages to arrive and complete their "set".
173     """
174 
175     def __init__(self, fs, queue_size):
176         SimpleFilter.__init__(self)
177         self.connectInput(fs)
178         self.queue_size = queue_size
179         self.lock = threading.Lock()
180 
181     def connectInput(self, fs):
182         self.queues = [{} for f in fs]
183         self.input_connections = [f.registerCallback(self.add, q) for (f, q) in zip(fs, self.queues)]
184 
185     def add(self, msg, my_queue):
186         self.lock.acquire()
187         my_queue[msg.header.stamp] = msg
188         while len(my_queue) > self.queue_size:
189             del my_queue[min(my_queue)]
190         # common is the set of timestamps that occur in all queues
191         common = reduce(set.intersection, [set(q) for q in self.queues])
192         for t in sorted(common):
193             # msgs is list of msgs (one from each queue) with stamp t
194             msgs = [q[t] for q in self.queues]
195             self.signalMessage(*msgs)
196             for q in self.queues:
197                 del q[t]
198         self.lock.release()
199 
200 class ApproximateTimeSynchronizer(TimeSynchronizer):
201 
202     """
203     Approximately synchronizes messages by their timestamps.
204 
205     :class:`ApproximateTimeSynchronizer` synchronizes incoming message filters by the
206     timestamps contained in their messages' headers. The API is the same as TimeSynchronizer
207     except for an extra `slop` parameter in the constructor that defines the delay (in seconds)
208     with which messages can be synchronized
209     """
210 
211     def __init__(self, fs, queue_size, slop):
212         TimeSynchronizer.__init__(self, fs, queue_size)
213         self.slop = rospy.Duration.from_sec(slop)
214 
215     def add(self, msg, my_queue):
216         self.lock.acquire()
217         my_queue[msg.header.stamp] = msg
218         while len(my_queue) > self.queue_size:
219             del my_queue[min(my_queue)]
220         for vv in itertools.product(*[list(q.keys()) for q in self.queues]):
221             qt = list(zip(self.queues, vv))
222             if ( ((max(vv) - min(vv)) < self.slop) and
223                 (len([1 for q,t in qt if t not in q]) == 0) ):
224                 msgs = [q[t] for q,t in qt]
225                 self.signalMessage(*msgs)
226                 for q,t in qt:
227                     del q[t]
228         self.lock.release()
229
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/utilities/message_filters/src/message_filters/__init__.py b/utilities/message_filters/src/message_filters/__init__.py
--- a/utilities/message_filters/src/message_filters/__init__.py
+++ b/utilities/message_filters/src/message_filters/__init__.py
@@ -76,6 +76,9 @@
     def getTopic(self):
         return self.topic
 
+    def unregister(self):
+        self.sub.unregister()
+
 class Cache(SimpleFilter):
 
     """

{"golden_diff": "diff --git a/utilities/message_filters/src/message_filters/__init__.py b/utilities/message_filters/src/message_filters/__init__.py\n--- a/utilities/message_filters/src/message_filters/__init__.py\n+++ b/utilities/message_filters/src/message_filters/__init__.py\n@@ -76,6 +76,9 @@\n     def getTopic(self):\n         return self.topic\n \n+    def unregister(self):\n+        self.sub.unregister()\n+\n class Cache(SimpleFilter):\n \n     \"\"\"\n", "issue": "unregister() method to message_filter.Subscriber on python\nDo you have plan to implement this?\nOr there is other way to achieve this?\n\n", "before_files": [{"content": "# Copyright (c) 2009, Willow Garage, Inc.\n# All rights reserved.\n#\n# Redistribution and use in source and binary forms, with or without\n# modification, are permitted provided that the following conditions are met:\n#\n#     * Redistributions of source code must retain the above copyright\n#       notice, this list of conditions and the following disclaimer.\n#     * Redistributions in binary form must reproduce the above copyright\n#       notice, this list of conditions and the following disclaimer in the\n#       documentation and/or other materials provided with the distribution.\n#     * Neither the name of the Willow Garage, Inc. nor the names of its\n#       contributors may be used to endorse or promote products derived from\n#       this software without specific prior written permission.\n#\n# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS \"AS IS\"\n# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE\n# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE\n# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE\n# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR\n# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF\n# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS\n# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN\n# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)\n# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE\n# POSSIBILITY OF SUCH DAMAGE.\n\n\"\"\"\nMessage Filter Objects\n======================\n\"\"\"\n\nimport itertools\nimport threading\nimport rospy\n\nclass SimpleFilter:\n\n    def __init__(self):\n        self.callbacks = {}\n\n    def registerCallback(self, cb, *args):\n        \"\"\"\n        Register a callback function `cb` to be called when this filter\n        has output.\n        The filter calls the function ``cb`` with a filter-dependent list of arguments,\n        followed by the call-supplied arguments ``args``.\n        \"\"\"\n\n        conn = len(self.callbacks)\n        self.callbacks[conn] = (cb, args)\n        return conn\n\n    def signalMessage(self, *msg):\n        for (cb, args) in self.callbacks.values():\n            cb(*(msg + args))\n\nclass Subscriber(SimpleFilter):\n    \n    \"\"\"\n    ROS subscription filter.  Identical arguments as :class:`rospy.Subscriber`.\n\n    This class acts as a highest-level filter, simply passing messages\n    from a ROS subscription through to the filters which have connected\n    to it.\n    \"\"\"\n    def __init__(self, *args, **kwargs):\n        SimpleFilter.__init__(self)\n        self.topic = args[0]\n        kwargs['callback'] = self.callback\n        self.sub = rospy.Subscriber(*args, **kwargs)\n\n    def callback(self, msg):\n        self.signalMessage(msg)\n\n    def getTopic(self):\n        return self.topic\n\nclass Cache(SimpleFilter):\n\n    \"\"\"\n    Stores a time history of messages.\n\n    Given a stream of messages, the most recent ``cache_size`` messages\n    are cached in a ring buffer, from which time intervals of the cache\n    can then be retrieved by the client.\n    \"\"\"\n\n    def __init__(self, f, cache_size = 1):\n        SimpleFilter.__init__(self)\n        self.connectInput(f)\n        self.cache_size = cache_size\n        # Array to store messages\n        self.cache_msgs = []\n        # Array to store msgs times, auxiliary structure to facilitate\n        # sorted insertion\n        self.cache_times = []\n\n    def connectInput(self, f):\n        self.incoming_connection = f.registerCallback(self.add)\n\n    def add(self, msg):\n        # Cannot use message filters with non-stamped messages\n        if not hasattr(msg, 'header') or not hasattr(msg.header, 'stamp'):\n            rospy.logwarn(\"Cannot use message filters with non-stamped messages\")\n            return\n\n        # Insert sorted\n        stamp = msg.header.stamp\n        self.cache_times.append(stamp)\n        self.cache_msgs.append(msg)\n\n        # Implement a ring buffer, discard older if oversized\n        if (len(self.cache_msgs) > self.cache_size):\n            del self.cache_msgs[0]\n            del self.cache_times[0]\n\n        # Signal new input\n        self.signalMessage(msg)\n\n    def getInterval(self, from_stamp, to_stamp):\n        \"\"\"Query the current cache content between from_stamp to to_stamp.\"\"\"\n        assert from_stamp <= to_stamp\n        return [m for m in self.cache_msgs\n                if m.header.stamp >= from_stamp and m.header.stamp <= to_stamp]\n\n    def getElemAfterTime(self, stamp):\n        \"\"\"Return the oldest element after or equal the passed time stamp.\"\"\"\n        newer = [m for m in self.cache_msgs if m.header.stamp >= stamp]\n        if not newer:\n            return None\n        return newer[0]\n\n    def getElemBeforeTime(self, stamp):\n        \"\"\"Return the newest element before or equal the passed time stamp.\"\"\"\n        older = [m for m in self.cache_msgs if m.header.stamp <= stamp]\n        if not older:\n            return None\n        return older[-1]\n\n    def getLastestTime(self):\n        \"\"\"Return the newest recorded timestamp.\"\"\"\n        if not self.cache_times:\n            return None\n        return self.cache_times[-1]\n\n    def getOldestTime(self):\n        \"\"\"Return the oldest recorded timestamp.\"\"\"\n        if not self.cache_times:\n            return None\n        return self.cache_times[0]\n\n\nclass TimeSynchronizer(SimpleFilter):\n\n    \"\"\"\n    Synchronizes messages by their timestamps.\n\n    :class:`TimeSynchronizer` synchronizes incoming message filters by the\n    timestamps contained in their messages' headers. TimeSynchronizer\n    listens on multiple input message filters ``fs``, and invokes the callback\n    when it has a collection of messages with matching timestamps.\n\n    The signature of the callback function is::\n\n        def callback(msg1, ... msgN):\n\n    where N is the number of input message filters, and each message is\n    the output of the corresponding filter in ``fs``.\n    The required ``queue size`` parameter specifies how many sets of\n    messages it should store from each input filter (by timestamp)\n    while waiting for messages to arrive and complete their \"set\".\n    \"\"\"\n\n    def __init__(self, fs, queue_size):\n        SimpleFilter.__init__(self)\n        self.connectInput(fs)\n        self.queue_size = queue_size\n        self.lock = threading.Lock()\n\n    def connectInput(self, fs):\n        self.queues = [{} for f in fs]\n        self.input_connections = [f.registerCallback(self.add, q) for (f, q) in zip(fs, self.queues)]\n\n    def add(self, msg, my_queue):\n        self.lock.acquire()\n        my_queue[msg.header.stamp] = msg\n        while len(my_queue) > self.queue_size:\n            del my_queue[min(my_queue)]\n        # common is the set of timestamps that occur in all queues\n        common = reduce(set.intersection, [set(q) for q in self.queues])\n        for t in sorted(common):\n            # msgs is list of msgs (one from each queue) with stamp t\n            msgs = [q[t] for q in self.queues]\n            self.signalMessage(*msgs)\n            for q in self.queues:\n                del q[t]\n        self.lock.release()\n\nclass ApproximateTimeSynchronizer(TimeSynchronizer):\n\n    \"\"\"\n    Approximately synchronizes messages by their timestamps.\n\n    :class:`ApproximateTimeSynchronizer` synchronizes incoming message filters by the\n    timestamps contained in their messages' headers. The API is the same as TimeSynchronizer\n    except for an extra `slop` parameter in the constructor that defines the delay (in seconds)\n    with which messages can be synchronized\n    \"\"\"\n\n    def __init__(self, fs, queue_size, slop):\n        TimeSynchronizer.__init__(self, fs, queue_size)\n        self.slop = rospy.Duration.from_sec(slop)\n\n    def add(self, msg, my_queue):\n        self.lock.acquire()\n        my_queue[msg.header.stamp] = msg\n        while len(my_queue) > self.queue_size:\n            del my_queue[min(my_queue)]\n        for vv in itertools.product(*[list(q.keys()) for q in self.queues]):\n            qt = list(zip(self.queues, vv))\n            if ( ((max(vv) - min(vv)) < self.slop) and\n                (len([1 for q,t in qt if t not in q]) == 0) ):\n                msgs = [q[t] for q,t in qt]\n                self.signalMessage(*msgs)\n                for q,t in qt:\n                    del q[t]\n        self.lock.release()\n", "path": "utilities/message_filters/src/message_filters/__init__.py"}], "after_files": [{"content": "# Copyright (c) 2009, Willow Garage, Inc.\n# All rights reserved.\n#\n# Redistribution and use in source and binary forms, with or without\n# modification, are permitted provided that the following conditions are met:\n#\n#     * Redistributions of source code must retain the above copyright\n#       notice, this list of conditions and the following disclaimer.\n#     * Redistributions in binary form must reproduce the above copyright\n#       notice, this list of conditions and the following disclaimer in the\n#       documentation and/or other materials provided with the distribution.\n#     * Neither the name of the Willow Garage, Inc. nor the names of its\n#       contributors may be used to endorse or promote products derived from\n#       this software without specific prior written permission.\n#\n# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS \"AS IS\"\n# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE\n# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE\n# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE\n# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR\n# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF\n# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS\n# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN\n# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)\n# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE\n# POSSIBILITY OF SUCH DAMAGE.\n\n\"\"\"\nMessage Filter Objects\n======================\n\"\"\"\n\nimport itertools\nimport threading\nimport rospy\n\nclass SimpleFilter:\n\n    def __init__(self):\n        self.callbacks = {}\n\n    def registerCallback(self, cb, *args):\n        \"\"\"\n        Register a callback function `cb` to be called when this filter\n        has output.\n        The filter calls the function ``cb`` with a filter-dependent list of arguments,\n        followed by the call-supplied arguments ``args``.\n        \"\"\"\n\n        conn = len(self.callbacks)\n        self.callbacks[conn] = (cb, args)\n        return conn\n\n    def signalMessage(self, *msg):\n        for (cb, args) in self.callbacks.values():\n            cb(*(msg + args))\n\nclass Subscriber(SimpleFilter):\n    \n    \"\"\"\n    ROS subscription filter.  Identical arguments as :class:`rospy.Subscriber`.\n\n    This class acts as a highest-level filter, simply passing messages\n    from a ROS subscription through to the filters which have connected\n    to it.\n    \"\"\"\n    def __init__(self, *args, **kwargs):\n        SimpleFilter.__init__(self)\n        self.topic = args[0]\n        kwargs['callback'] = self.callback\n        self.sub = rospy.Subscriber(*args, **kwargs)\n\n    def callback(self, msg):\n        self.signalMessage(msg)\n\n    def getTopic(self):\n        return self.topic\n\n    def unregister(self):\n        self.sub.unregister()\n\nclass Cache(SimpleFilter):\n\n    \"\"\"\n    Stores a time history of messages.\n\n    Given a stream of messages, the most recent ``cache_size`` messages\n    are cached in a ring buffer, from which time intervals of the cache\n    can then be retrieved by the client.\n    \"\"\"\n\n    def __init__(self, f, cache_size = 1):\n        SimpleFilter.__init__(self)\n        self.connectInput(f)\n        self.cache_size = cache_size\n        # Array to store messages\n        self.cache_msgs = []\n        # Array to store msgs times, auxiliary structure to facilitate\n        # sorted insertion\n        self.cache_times = []\n\n    def connectInput(self, f):\n        self.incoming_connection = f.registerCallback(self.add)\n\n    def add(self, msg):\n        # Cannot use message filters with non-stamped messages\n        if not hasattr(msg, 'header') or not hasattr(msg.header, 'stamp'):\n            rospy.logwarn(\"Cannot use message filters with non-stamped messages\")\n            return\n\n        # Insert sorted\n        stamp = msg.header.stamp\n        self.cache_times.append(stamp)\n        self.cache_msgs.append(msg)\n\n        # Implement a ring buffer, discard older if oversized\n        if (len(self.cache_msgs) > self.cache_size):\n            del self.cache_msgs[0]\n            del self.cache_times[0]\n\n        # Signal new input\n        self.signalMessage(msg)\n\n    def getInterval(self, from_stamp, to_stamp):\n        \"\"\"Query the current cache content between from_stamp to to_stamp.\"\"\"\n        assert from_stamp <= to_stamp\n        return [m for m in self.cache_msgs\n                if m.header.stamp >= from_stamp and m.header.stamp <= to_stamp]\n\n    def getElemAfterTime(self, stamp):\n        \"\"\"Return the oldest element after or equal the passed time stamp.\"\"\"\n        newer = [m for m in self.cache_msgs if m.header.stamp >= stamp]\n        if not newer:\n            return None\n        return newer[0]\n\n    def getElemBeforeTime(self, stamp):\n        \"\"\"Return the newest element before or equal the passed time stamp.\"\"\"\n        older = [m for m in self.cache_msgs if m.header.stamp <= stamp]\n        if not older:\n            return None\n        return older[-1]\n\n    def getLastestTime(self):\n        \"\"\"Return the newest recorded timestamp.\"\"\"\n        if not self.cache_times:\n            return None\n        return self.cache_times[-1]\n\n    def getOldestTime(self):\n        \"\"\"Return the oldest recorded timestamp.\"\"\"\n        if not self.cache_times:\n            return None\n        return self.cache_times[0]\n\n\nclass TimeSynchronizer(SimpleFilter):\n\n    \"\"\"\n    Synchronizes messages by their timestamps.\n\n    :class:`TimeSynchronizer` synchronizes incoming message filters by the\n    timestamps contained in their messages' headers. TimeSynchronizer\n    listens on multiple input message filters ``fs``, and invokes the callback\n    when it has a collection of messages with matching timestamps.\n\n    The signature of the callback function is::\n\n        def callback(msg1, ... msgN):\n\n    where N is the number of input message filters, and each message is\n    the output of the corresponding filter in ``fs``.\n    The required ``queue size`` parameter specifies how many sets of\n    messages it should store from each input filter (by timestamp)\n    while waiting for messages to arrive and complete their \"set\".\n    \"\"\"\n\n    def __init__(self, fs, queue_size):\n        SimpleFilter.__init__(self)\n        self.connectInput(fs)\n        self.queue_size = queue_size\n        self.lock = threading.Lock()\n\n    def connectInput(self, fs):\n        self.queues = [{} for f in fs]\n        self.input_connections = [f.registerCallback(self.add, q) for (f, q) in zip(fs, self.queues)]\n\n    def add(self, msg, my_queue):\n        self.lock.acquire()\n        my_queue[msg.header.stamp] = msg\n        while len(my_queue) > self.queue_size:\n            del my_queue[min(my_queue)]\n        # common is the set of timestamps that occur in all queues\n        common = reduce(set.intersection, [set(q) for q in self.queues])\n        for t in sorted(common):\n            # msgs is list of msgs (one from each queue) with stamp t\n            msgs = [q[t] for q in self.queues]\n            self.signalMessage(*msgs)\n            for q in self.queues:\n                del q[t]\n        self.lock.release()\n\nclass ApproximateTimeSynchronizer(TimeSynchronizer):\n\n    \"\"\"\n    Approximately synchronizes messages by their timestamps.\n\n    :class:`ApproximateTimeSynchronizer` synchronizes incoming message filters by the\n    timestamps contained in their messages' headers. The API is the same as TimeSynchronizer\n    except for an extra `slop` parameter in the constructor that defines the delay (in seconds)\n    with which messages can be synchronized\n    \"\"\"\n\n    def __init__(self, fs, queue_size, slop):\n        TimeSynchronizer.__init__(self, fs, queue_size)\n        self.slop = rospy.Duration.from_sec(slop)\n\n    def add(self, msg, my_queue):\n        self.lock.acquire()\n        my_queue[msg.header.stamp] = msg\n        while len(my_queue) > self.queue_size:\n            del my_queue[min(my_queue)]\n        for vv in itertools.product(*[list(q.keys()) for q in self.queues]):\n            qt = list(zip(self.queues, vv))\n            if ( ((max(vv) - min(vv)) < self.slop) and\n                (len([1 for q,t in qt if t not in q]) == 0) ):\n                msgs = [q[t] for q,t in qt]\n                self.signalMessage(*msgs)\n                for q,t in qt:\n                    del q[t]\n        self.lock.release()\n", "path": "utilities/message_filters/src/message_filters/__init__.py"}]}