prompt: We are currently solving the following issue within our repository. Here is the issue text:
  --- BEGIN ISSUE ---
  message_filters: Incorporate python approximate time synchronizer
  https://github.com/ros-perception/image_pipeline/issues/76
  
  
  --- 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 threading
  34 import rospy
  35 
  36 class SimpleFilter:
  37 
  38     def __init__(self):
  39         self.callbacks = {}
  40 
  41     def registerCallback(self, cb, *args):
  42         """
  43         Register a callback function `cb` to be called when this filter
  44         has output.
  45         The filter calls the function ``cb`` with a filter-dependent list of arguments,
  46         followed by the call-supplied arguments ``args``.
  47         """
  48 
  49         conn = len(self.callbacks)
  50         self.callbacks[conn] = (cb, args)
  51         return conn
  52 
  53     def signalMessage(self, *msg):
  54         for (cb, args) in self.callbacks.values():
  55             cb(*(msg + args))
  56 
  57 class Subscriber(SimpleFilter):
  58     
  59     """
  60     ROS subscription filter.  Identical arguments as :class:`rospy.Subscriber`.
  61 
  62     This class acts as a highest-level filter, simply passing messages
  63     from a ROS subscription through to the filters which have connected
  64     to it.
  65     """
  66     def __init__(self, *args, **kwargs):
  67         SimpleFilter.__init__(self)
  68         self.topic = args[0]
  69         kwargs['callback'] = self.callback
  70         self.sub = rospy.Subscriber(*args, **kwargs)
  71 
  72     def callback(self, msg):
  73         self.signalMessage(msg)
  74 
  75     def getTopic(self):
  76         return self.topic
  77 
  78 class Cache(SimpleFilter):
  79 
  80     """
  81     Stores a time history of messages.
  82 
  83     Given a stream of messages, the most recent ``cache_size`` messages
  84     are cached in a ring buffer, from which time intervals of the cache
  85     can then be retrieved by the client.
  86     """
  87 
  88     def __init__(self, f, cache_size = 1):
  89         SimpleFilter.__init__(self)
  90         self.connectInput(f)
  91         self.cache_size = cache_size
  92 
  93     def connectInput(self, f):
  94         self.incoming_connection = f.registerCallback(self.add)
  95 
  96     def add(self, msg):
  97         # Add msg to cache... XXX TODO
  98 
  99         self.signalMessage(msg)
  100 
  101 class TimeSynchronizer(SimpleFilter):
  102 
  103     """
  104     Synchronizes messages by their timestamps.
  105 
  106     :class:`TimeSynchronizer` synchronizes incoming message filters by the
  107     timestamps contained in their messages' headers. TimeSynchronizer
  108     listens on multiple input message filters ``fs``, and invokes the callback
  109     when it has a collection of messages with matching timestamps.
  110 
  111     The signature of the callback function is::
  112 
  113         def callback(msg1, ... msgN):
  114 
  115     where N is the number of input message filters, and each message is
  116     the output of the corresponding filter in ``fs``.
  117     The required ``queue size`` parameter specifies how many sets of
  118     messages it should store from each input filter (by timestamp)
  119     while waiting for messages to arrive and complete their "set".
  120     """
  121 
  122     def __init__(self, fs, queue_size):
  123         SimpleFilter.__init__(self)
  124         self.connectInput(fs)
  125         self.queue_size = queue_size
  126         self.lock = threading.Lock()
  127 
  128     def connectInput(self, fs):
  129         self.queues = [{} for f in fs]
  130         self.input_connections = [f.registerCallback(self.add, q) for (f, q) in zip(fs, self.queues)]
  131 
  132     def add(self, msg, my_queue):
  133         self.lock.acquire()
  134         my_queue[msg.header.stamp] = msg
  135         while len(my_queue) > self.queue_size:
  136             del my_queue[min(my_queue)]
  137         # common is the set of timestamps that occur in all queues
  138         common = reduce(set.intersection, [set(q) for q in self.queues])
  139         for t in sorted(common):
  140             # msgs is list of msgs (one from each queue) with stamp t
  141             msgs = [q[t] for q in self.queues]
  142             self.signalMessage(*msgs)
  143             for q in self.queues:
  144                 del q[t]
  145         self.lock.release()
  146
  ```
  --- 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()
  
  ```
  
golden_diff: 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
  @@ -30,6 +30,7 @@
   ======================
   """
   
  +import itertools
   import threading
   import rospy
   
  @@ -143,3 +144,33 @@
               for q in self.queues:
                   del q[t]
           self.lock.release()
  +
  +class ApproximateTimeSynchronizer(TimeSynchronizer):
  +
  +    """
  +    Approximately synchronizes messages by their timestamps.
  +
  +    :class:`ApproximateTimeSynchronizer` synchronizes incoming message filters by the
  +    timestamps contained in their messages' headers. The API is the same as TimeSynchronizer
  +    except for an extra `slop` parameter in the constructor that defines the delay (in seconds)
  +    with which messages can be synchronized
  +    """
  +
  +    def __init__(self, fs, queue_size, slop):
  +        TimeSynchronizer.__init__(self, fs, queue_size)
  +        self.slop = rospy.Duration.from_sec(slop)
  +
  +    def add(self, msg, my_queue):
  +        self.lock.acquire()
  +        my_queue[msg.header.stamp] = msg
  +        while len(my_queue) > self.queue_size:
  +            del my_queue[min(my_queue)]
  +        for vv in itertools.product(*[list(q.keys()) for q in self.queues]):
  +            qt = list(zip(self.queues, vv))
  +            if ( ((max(vv) - min(vv)) < self.slop) and
  +                (len([1 for q,t in qt if t not in q]) == 0) ):
  +                msgs = [q[t] for q,t in qt]
  +                self.signalMessage(*msgs)
  +                for q,t in qt:
  +                    del q[t]
  +        self.lock.release()
  
verification_info: {"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@@ -30,6 +30,7 @@\n ======================\n \"\"\"\n \n+import itertools\n import threading\n import rospy\n \n@@ -143,3 +144,33 @@\n             for q in self.queues:\n                 del q[t]\n         self.lock.release()\n+\n+class 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", "issue": "message_filters: Incorporate python approximate time synchronizer\nhttps://github.com/ros-perception/image_pipeline/issues/76\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 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\n    def connectInput(self, f):\n        self.incoming_connection = f.registerCallback(self.add)\n\n    def add(self, msg):\n        # Add msg to cache... XXX TODO\n\n        self.signalMessage(msg)\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", "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\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\n    def connectInput(self, f):\n        self.incoming_connection = f.registerCallback(self.add)\n\n    def add(self, msg):\n        # Add msg to cache... XXX TODO\n\n        self.signalMessage(msg)\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"}]}