more Python 3 compatibility

test/test_rospy/test/unit/test_genmsg_py.py

@@ -95,8 +95,8 @@ def test_test_rospy_TestFixedArray(self):
         self.assertEquals([0., 0., 0.], m.f64_3)        
         self.assertEquals([0], m.i8_1)
         self.assertEquals([0, 0, 0], m.i8_3)        
-        self.assertEquals(chr(0), m.u8_1)
-        self.assertEquals(chr(0)*3, m.u8_3)        
+        self.assertEquals(chr(0).encode(), m.u8_1)
+        self.assertEquals((chr(0)*3).encode(), m.u8_3)
         self.assertEquals([0], m.i32_1)
         self.assertEquals([0, 0, 0], m.i32_3)        
         self.assertEquals([0], m.u32_1)
@@ -358,7 +358,7 @@ def test_std_msgs_MultiArray(self):
         # test. the buff was with the uint8[] type consistency
         buff = StringIO()
         self.assertEquals(UInt8MultiArray(),UInt8MultiArray())
-        self.assertEquals('',UInt8MultiArray().data)        
+        self.assertEquals(b'', UInt8MultiArray().data)
         UInt8MultiArray().serialize(buff)
         self.assertEquals(UInt8MultiArray(layout=MultiArrayLayout()),UInt8MultiArray())
         UInt8MultiArray(layout=MultiArrayLayout()).serialize(buff)

tools/rosgraph/src/rosgraph/roslogging.py

@@ -49,7 +49,7 @@
 class LoggingException(Exception): pass
 
 class RospyLogger(logging.getLoggerClass()):
-    def findCaller(self, dummy=False): # Dummy second arg to match Python3 function declaration
+    def findCaller(self, stack_info=False):
         """
         Find the stack frame of the caller so that we can note the source
         file name, line number, and function name with class name if possible.

tools/rosgraph/test/test_roslogging_user_logger.py

@@ -51,12 +51,16 @@
 
 # set user defined custom logger
 class UserCustomLogger(logging.Logger):
-    def findCaller(self):
+    def findCaller(self, stack_info=False):
         """Returns static caller.
 
         This method is being overwritten in rosgraph.roslogging.
         """
-        return '<filename>', '<lineno>', '<func_name>'
+        if sys.version_info > (3, 2):
+            # Dummy last argument to match Python3 return type
+            return '<filename>', '<lineno>', '<func_name>', None
+        else:
+            return '<filename>', '<lineno>', '<func_name>'
 
     def _log(self, level, msg, args, exc_info=None, extra=None):
         """Write log with ROS_IP.

tools/roslaunch/test/unit/test_xmlloader.py

@@ -218,7 +218,7 @@ def test_params(self):
         p = [p for p in mock.params if p.key == '/configfile'][0]
         self.assertEquals(contents, p.value, 1)
         p = [p for p in mock.params if p.key == '/binaryfile'][0]
-        self.assertEquals(Binary(contents), p.value, 1)
+        self.assertEquals(Binary(contents.encode()), p.value, 1)
 
         f = open(os.path.join(get_example_path(), 'example.launch'))
         try:

tools/rosmsg/src/rosmsg/__init__.py

@@ -240,7 +240,7 @@ def create_names_filter(names):
     """
     returns a function to use as filter that returns all objects slots except those with names in list.
     """
-    return lambda obj : filter(lambda slotname : not slotname in names, obj.__slots__)
+    return lambda obj : list(filter(lambda slotname : not slotname in names, obj.__slots__))
 
 
 def init_rosmsg_proto():

tools/rosmsg/test/test_rosmsg_command_line.py

@@ -54,24 +54,24 @@ def test_cmd_help(self):
 
         for cmd in ['rosmsg', 'rossrv']:
             glob_cmd=[os.path.join(_SCRIPT_FOLDER, cmd)]
-            output = Popen(glob_cmd, stdout=PIPE, env=self.new_environ).communicate()[0]
+            output = Popen(glob_cmd, stdout=PIPE, env=self.new_environ).communicate()[0].decode()
             self.assert_('Commands' in output)
-            output = Popen(glob_cmd+['-h'], stdout=PIPE, env=self.new_environ).communicate()[0]
+            output = Popen(glob_cmd+['-h'], stdout=PIPE, env=self.new_environ).communicate()[0].decode()
             self.assert_('Commands' in output)
             self.assert_('Traceback' not in output)
             for c in sub:
                 self.assert_("%s %s"%(cmd, c) in output, "%s %s"%(cmd, c) + " not in "+ output + " of " + str(glob_cmd))
 
             for c in sub:
-                output = Popen(glob_cmd + [c, '-h'], stdout=PIPE, env=self.new_environ).communicate()[0]
+                output = Popen(glob_cmd + [c, '-h'], stdout=PIPE, env=self.new_environ).communicate()[0].decode()
                 self.assert_('Usage' in output)
                 self.assert_("%s %s"%(cmd, c) in output, output)
 
     def test_cmd_packages(self):
         # - single line
-        output1 = Popen(['rosmsg', 'packages', '-s'], stdout=PIPE).communicate()[0]
+        output1 = Popen(['rosmsg', 'packages', '-s'], stdout=PIPE).communicate()[0].decode()
         # - multi-line
-        output2 = Popen(['rosmsg', 'packages'], stdout=PIPE).communicate()[0]
+        output2 = Popen(['rosmsg', 'packages'], stdout=PIPE).communicate()[0].decode()
         l1 = [x for x in output1.split() if x]
         l2 = [x.strip() for x in output2.split('\n') if x.strip()]
         self.assertEquals(l1, l2)
@@ -80,8 +80,8 @@ def test_cmd_packages(self):
         for p in ['std_srvs', 'rosmsg']:
             self.assert_(p not in l1)
 
-        output1 = Popen(['rossrv', 'packages', '-s'], stdout=PIPE).communicate()[0]
-        output2 = Popen(['rossrv', 'packages'], stdout=PIPE).communicate()[0]
+        output1 = Popen(['rossrv', 'packages', '-s'], stdout=PIPE).communicate()[0].decode()
+        output2 = Popen(['rossrv', 'packages'], stdout=PIPE).communicate()[0].decode()
         l1 = [x for x in output1.split() if x]
         l2 = [x.strip() for x in output2.split('\n') if x.strip()]
         self.assertEquals(l1, l2)
@@ -92,14 +92,14 @@ def test_cmd_packages(self):
 
     def test_cmd_list(self):
         # - multi-line
-        output1 = Popen([os.path.join(_SCRIPT_FOLDER,'rosmsg'), 'list'], stdout=PIPE).communicate()[0]
+        output1 = Popen([os.path.join(_SCRIPT_FOLDER,'rosmsg'), 'list'], stdout=PIPE).communicate()[0].decode()
         l1 = [x.strip() for x in output1.split('\n') if x.strip()]
         for p in ['std_msgs/String', 'test_rosmaster/Floats']:
             self.assert_(p in l1)
         for p in ['std_srvs/Empty', 'roscpp/Empty']:
             self.assert_(p not in l1)
 
-        output1 = Popen([os.path.join(_SCRIPT_FOLDER,'rossrv'), 'list'], stdout=PIPE).communicate()[0]
+        output1 = Popen([os.path.join(_SCRIPT_FOLDER,'rossrv'), 'list'], stdout=PIPE).communicate()[0].decode()
         l1 = [x.strip() for x in output1.split('\n') if x.strip()]
         for p in ['std_srvs/Empty', 'roscpp/Empty']:
             self.assert_(p in l1)
@@ -109,9 +109,9 @@ def test_cmd_list(self):
     def test_cmd_package(self):
         # this test is obviously very brittle, but should stabilize as the tests stabilize
         # - single line output
-        output1 = Popen(['rosmsg', 'package', '-s', 'test_rosmaster'], stdout=PIPE).communicate()[0]
+        output1 = Popen(['rosmsg', 'package', '-s', 'test_rosmaster'], stdout=PIPE).communicate()[0].decode()
         # - multi-line output
-        output2 = Popen(['rosmsg', 'package', 'test_rosmaster'], stdout=PIPE).communicate()[0]
+        output2 = Popen(['rosmsg', 'package', 'test_rosmaster'], stdout=PIPE).communicate()[0].decode()
         l = set([x for x in output1.split() if x])        
         l2 = set([x.strip() for x in output2.split('\n') if x.strip()])
         self.assertEquals(l, l2)
@@ -121,8 +121,8 @@ def test_cmd_package(self):
                   'test_rosmaster/RosmsgC']:
             self.assertTrue(m in l, l)
 
-        output = Popen(['rossrv', 'package', '-s', 'test_rosmaster'], stdout=PIPE).communicate()[0]
-        output2 = Popen(['rossrv', 'package','test_rosmaster'], stdout=PIPE).communicate()[0]        
+        output = Popen(['rossrv', 'package', '-s', 'test_rosmaster'], stdout=PIPE).communicate()[0].decode()
+        output2 = Popen(['rossrv', 'package','test_rosmaster'], stdout=PIPE).communicate()[0].decode()
         l = set([x for x in output.split() if x])
         l2 = set([x.strip() for x in output2.split('\n') if x.strip()])
         self.assertEquals(l, l2)
@@ -132,14 +132,14 @@ def test_cmd_package(self):
 
     ## test that the rosmsg/rossrv show command works
     def test_cmd_show(self):
-        output = Popen(['rosmsg', 'show', 'std_msgs/String'], stdout=PIPE).communicate()[0]
+        output = Popen(['rosmsg', 'show', 'std_msgs/String'], stdout=PIPE).communicate()[0].decode()
         self.assertEquals('string data', output.strip())
 
-        output = Popen(['rossrv', 'show', 'std_srvs/Empty'], stdout=PIPE).communicate()[0]
+        output = Popen(['rossrv', 'show', 'std_srvs/Empty'], stdout=PIPE).communicate()[0].decode()
         self.assertEquals('---', output.strip())
-        output = Popen(['rossrv', 'show', 'std_srvs/Empty'], stdout=PIPE).communicate()[0]
+        output = Popen(['rossrv', 'show', 'std_srvs/Empty'], stdout=PIPE).communicate()[0].decode()
         self.assertEquals('---', output.strip())
-        output = Popen(['rossrv', 'show', 'test_rosmaster/AddTwoInts'], stdout=PIPE).communicate()[0]
+        output = Popen(['rossrv', 'show', 'test_rosmaster/AddTwoInts'], stdout=PIPE).communicate()[0].decode()
         self.assertEquals('int64 a\nint64 b\n---\nint64 sum', output.strip())
 
         # test against test_rosmsg package
@@ -159,20 +159,20 @@ def test_cmd_show(self):
             text = text+'\n' # running command adds one new line
             text_raw = text_raw+'\n'
             type_ =test_message_package+'/'+t
-            output = Popen(['rosmsg', 'show', type_], stdout=PIPE).communicate()[0]
+            output = Popen(['rosmsg', 'show', type_], stdout=PIPE).communicate()[0].decode()
             self.assertEquals(text, output)
-            output = Popen(['rosmsg', 'show', '-r',type_], stdout=PIPE).communicate()[0]
+            output = Popen(['rosmsg', 'show', '-r',type_], stdout=PIPE).communicate()[0].decode()
             self.assertEquals(text_raw, output)
-            output = Popen(['rosmsg', 'show', '--raw', type_], stdout=PIPE).communicate()[0]
+            output = Popen(['rosmsg', 'show', '--raw', type_], stdout=PIPE).communicate()[0].decode()
             self.assertEquals(text_raw, output)
 
             # test as search
             type_ = t
             text = "[test_rosmaster/%s]:\n%s"%(t, text)
             text_raw = "[test_rosmaster/%s]:\n%s"%(t, text_raw)
-            output = Popen(['rosmsg', 'show', type_], stdout=PIPE).communicate()[0]
+            output = Popen(['rosmsg', 'show', type_], stdout=PIPE).communicate()[0].decode()
             self.assertEquals(text, output)
             output = Popen(['rosmsg', 'show', '-r',type_], stdout=PIPE, stderr=PIPE).communicate()
-            self.assertEquals(text_raw, output[0], "Failed: %s"%(str(output)))
-            output = Popen(['rosmsg', 'show', '--raw', type_], stdout=PIPE).communicate()[0]
+            self.assertEquals(text_raw, output[0].decode(), "Failed: %s"%(str(output)))
+            output = Popen(['rosmsg', 'show', '--raw', type_], stdout=PIPE).communicate()[0].decode()
             self.assertEquals(text_raw, output)

tools/rosmsg/test/test_rosmsgproto_command_line.py

@@ -67,37 +67,37 @@ def testFail(self):
         cmd.extend(["msg", "foo123barxyz"])
         call = subprocess.Popen(cmd, stdout = subprocess.PIPE, stderr = subprocess.PIPE, env = self.new_environ)
         (output, erroutput) = call.communicate()
-        self.assertEqual('', output)
-        self.assertTrue('Unknown message name foo123barxyz' in erroutput)
+        self.assertEqual(b'', output)
+        self.assertTrue('Unknown message name foo123barxyz' in erroutput.decode())
 
     def testSilentFail(self):
         cmd = copy.copy(ROSMSGPROTO_FN)
         cmd.extend(["msg", "-s", "foo123barxyz"])
         call = subprocess.Popen(cmd, stdout = subprocess.PIPE, stderr = subprocess.PIPE, env = self.new_environ)
         (output, erroutput) = call.communicate()
-        self.assertEqual('', output)
-        self.assertEqual('', erroutput)
+        self.assertEqual(b'', output)
+        self.assertEqual(b'', erroutput)
 
     def testSilentFailCpp(self):
         cmd = copy.copy(ROSMSGPROTO_FN)
         cmd.extend(["msg", "-s", "foo123barxyz::bar"])
         call = subprocess.Popen(cmd, stdout = subprocess.PIPE, stderr = subprocess.PIPE, env = self.new_environ)
         (output, erroutput) = call.communicate()
-        self.assertEqual('', output)
-        self.assertEqual('', erroutput)
+        self.assertEqual(b'', output)
+        self.assertEqual(b'', erroutput)
 
     def testSilentFailDot(self):
         cmd = copy.copy(ROSMSGPROTO_FN)
         cmd.extend(["msg", "-s", "foo123barxyz.bar"])
         call = subprocess.Popen(cmd, stdout = subprocess.PIPE, stderr = subprocess.PIPE, env = self.new_environ)
         (output, erroutput) = call.communicate()
-        self.assertEqual('', output)
-        self.assertEqual('', erroutput)
+        self.assertEqual(b'', output)
+        self.assertEqual(b'', erroutput)
 
     def testSilentFailMode(self):
         cmd = copy.copy(ROSMSGPROTO_FN)
         cmd.extend(["msgfoobar", "-s", "foo123barxyz.bar"])
         call = subprocess.Popen(cmd, stdout = subprocess.PIPE, stderr = subprocess.PIPE, env = self.new_environ)
         (output, erroutput) = call.communicate()
-        self.assertEqual('', output)
-        self.assertEqual('', erroutput)
+        self.assertEqual(b'', output)
+        self.assertEqual(b'', erroutput)

tools/rostopic/test/test_rostopic_command_line_offline.py

@@ -48,9 +48,9 @@ def test_cmd_help(self):
         cmd = 'rostopic'
 
         sub = ['bw', 'echo', 'hz', 'delay', 'info', 'list', 'pub', 'type','find']
-        output = Popen([cmd], stdout=PIPE).communicate()[0]
+        output = Popen([cmd], stdout=PIPE).communicate()[0].decode()
         self.assert_('Commands' in output)
-        output = Popen([cmd, '-h'], stdout=PIPE).communicate()[0]
+        output = Popen([cmd, '-h'], stdout=PIPE).communicate()[0].decode()
         self.assert_('Commands' in output)
         # make sure all the commands are in the usage
         for c in sub:
@@ -59,16 +59,16 @@ def test_cmd_help(self):
 
         for c in sub:
             output = Popen([cmd, c, '-h'], stdout=PIPE, stderr=PIPE).communicate()
-            self.assert_("usage:" in output[0].lower(), output)
+            self.assert_("usage:" in output[0].decode().lower(), output)
             # make sure usage refers to the command
-            self.assert_("%s %s"%(cmd, c) in output[0], output)
+            self.assert_("%s %s"%(cmd, c) in output[0].decode(), output)
 
         # test no args on commands that require args
         for c in ['bw', 'echo', 'hz', 'delay', 'info', 'pub', 'type', 'find']:
             output = Popen([cmd, c], stdout=PIPE, stderr=PIPE).communicate()
-            self.assert_("usage:" in output[0].lower() or "usage:" in output[1].lower(), output)
+            self.assert_("usage:" in output[0].decode().lower() or "usage:" in output[1].decode().lower(), output)
             # make sure usage refers to the command
-            self.assert_("%s %s"%(cmd, c) in output[1], output)
+            self.assert_("%s %s"%(cmd, c) in output[1].decode(), output)
 
     def test_offline(self):
         cmd = 'rostopic'
@@ -80,19 +80,19 @@ def test_offline(self):
 
         msg = "ERROR: Unable to communicate with master!\n"
 
-        output = Popen([cmd, 'bw', 'chatter'], **kwds).communicate()
-        self.assert_(output[1].endswith(msg))
-        output = Popen([cmd, 'echo', 'chatter'], **kwds).communicate()
-        self.assert_(output[1].endswith(msg))
-        output = Popen([cmd, 'hz', 'chatter'], **kwds).communicate()
-        self.assert_(output[1].endswith(msg))
-        output = Popen([cmd, 'delay', 'chatter'], **kwds).communicate()
-        self.assert_(output[1].endswith(msg))
-        output = Popen([cmd, 'list'], **kwds).communicate()
-        self.assert_(output[1].endswith(msg))
-        output = Popen([cmd, 'pub', 'chatter', 'std_msgs/String', 'hello'], **kwds).communicate()
-        self.assert_(output[1].endswith(msg))
-        output = Popen([cmd, 'type', 'chatter'], **kwds).communicate()
-        self.assert_(output[1].endswith(msg))
-        output = Popen([cmd, 'type', 'std_msgs/String'], **kwds).communicate()
-        self.assert_(output[1].endswith(msg))
+        output = Popen([cmd, 'bw', 'chatter'], **kwds).communicate()[1].decode()
+        self.assert_(output.endswith(msg))
+        output = Popen([cmd, 'echo', 'chatter'], **kwds).communicate()[1].decode()
+        self.assert_(output.endswith(msg))
+        output = Popen([cmd, 'hz', 'chatter'], **kwds).communicate()[1].decode()
+        self.assert_(output.endswith(msg))
+        output = Popen([cmd, 'delay', 'chatter'], **kwds).communicate()[1].decode()
+        self.assert_(output.endswith(msg))
+        output = Popen([cmd, 'list'], **kwds).communicate()[1].decode()
+        self.assert_(output.endswith(msg))
+        output = Popen([cmd, 'pub', 'chatter', 'std_msgs/String', 'hello'], **kwds).communicate()[1].decode()
+        self.assert_(output.endswith(msg))
+        output = Popen([cmd, 'type', 'chatter'], **kwds).communicate()[1].decode()
+        self.assert_(output.endswith(msg))
+        output = Popen([cmd, 'type', 'std_msgs/String'], **kwds).communicate()[1].decode()
+        self.assert_(output.endswith(msg))