pylint

Author: Florian Weisshardt

.pylint.config

@@ -0,0 +1,16 @@
+[REPORTS]
+reports=no
+
+[MESSAGES CONTROL]
+disable= line-too-long,
+  missing-docstring,
+  old-style-class,
+  no-self-use,
+  too-many-locals,
+  too-many-branches,
+  too-few-public-methods,
+  too-many-public-methods,
+  too-many-instance-attributes,
+  duplicate-code,
+  invalid-name,
+  no-name-in-module

atf_core/scripts/analyse_all.py

@@ -1,5 +1,4 @@
 #!/usr/bin/env python
-import rospy
 import rospkg
 import sys
 import os
@@ -12,23 +11,23 @@
     print "analysing all in package '" + pkg + "'"
     pkg_path = r.get_path(pkg)
     path_to_test_files = os.path.join(pkg_path, "test_generated/analysing")
-    
+
     filenames = []
     for (dirpath, dirnames, list_of_files) in os.walk(path_to_test_files):
         for f in list_of_files:
             if f.endswith(".test"):
                 filenames.append(f)
 
     print "found " + str(len(filenames)) + " files: " + str(filenames)
-    
+
     # analyse all
     counter = 1
     for f in filenames:
         print "\n--> analysing " + str(counter) + "/" + str(len(filenames)) + " (" + str(f) + ")"
         command = "rostest " + pkg + " " + f
         subprocess.call(command, shell=True)
         counter += 1
-    
+
     # merge
     command = "rostest " + pkg + " " + "merging.test"
     subprocess.call(command, shell=True)

atf_core/scripts/generate_tests.py

@@ -23,7 +23,7 @@ def __init__(self, arguments):
         self.package_name = arguments[1]
         self.package_path = arguments[2]
         self.generation_config = self.load_yaml(self.package_path + "/config/test_generation_config.yaml")
-        
+
         # required parameters
         try:
             self.test_suite_file = os.path.join(self.package_path, self.generation_config["test_suite_file"])
@@ -44,7 +44,7 @@ def __init__(self, arguments):
             print error_message
             self.print_output = "ATF: Test generation failed! " + error_message
             sys.exit(1)
-        
+
         # optional parameters
         try:
             if type(self.generation_config["upload_data"]) is bool:
@@ -63,10 +63,13 @@ def __init__(self, arguments):
 
         self.test_list = {}
 
-        # Empty folders
         self.test_generated_path = os.path.join(self.package_path, "test_generated")
         self.test_generated_recording_path = os.path.join(self.test_generated_path, "recording")
         self.test_generated_analysing_path = os.path.join(self.test_generated_path, "analysing")
+        self.create_folders()
+
+    def create_folders(self):
+        # Empty folders
         if os.path.exists(self.test_generated_path):
             shutil.rmtree(self.test_generated_path)
         os.makedirs(self.test_generated_recording_path)
@@ -95,7 +98,7 @@ def generate_tests(self):
         test = em.test
         node = em.node
         param = em.param
-        rosparam = em.rosparam
+        #rosparam = em.rosparam
 
         for item in self.test_list:
             robot_config = self.load_yaml(os.path.join(self.package_path, self.generation_config["robot_config_path"], self.test_list[item]["robot"], "robot_config.yaml"))
@@ -128,7 +131,7 @@ def generate_tests(self):
 
             test_record.append(node(param(name="/test_config_file", value="$(find " + self.package_name + ")/" + self.generation_config["test_config_file"]),
                                     param(name="/bagfile_output", value=self.bagfile_output),
-                                    name="atf_recorder", pkg="atf_recorder", type="recorder_core.py", output="screen")) 
+                                    name="atf_recorder", pkg="atf_recorder", type="recorder_core.py", output="screen"))
 
             for params in robot_config["additional_parameter"]:
                 test_record.append(param(name=str(params["name"]), value=str(params["value"])))
@@ -160,7 +163,7 @@ def generate_tests(self):
             )
 
             xmlstr = minidom.parseString(ElementTree.tostring(test_analyse)).toprettyxml(indent="    ")
-            filepath = os.path.join(self.test_generated_analysing_path, "analysing_" + item) + ".test" 
+            filepath = os.path.join(self.test_generated_analysing_path, "analysing_" + item) + ".test"
             with open(filepath, "w") as f:
                 f.write(xmlstr)
 
@@ -175,14 +178,14 @@ def generate_tests(self):
                       'time-limit': "10"})
             )
             xmlstr = minidom.parseString(ElementTree.tostring(test_merge)).toprettyxml(indent="    ")
-            filepath = os.path.join(self.test_generated_path, "merging.test") 
+            filepath = os.path.join(self.test_generated_path, "merging.test")
             with open(filepath, "w") as f:
                 f.write(xmlstr)
-            
+
             # Uploading
             test_upload = launch()
             if self.upload_data:
-                test_upload.append(    
+                test_upload.append(
                     test({'test-name': "uploading_data", 'pkg': "atf_core", 'type': "test_dropbox_uploader.py",
                           'time-limit': str(self.time_limit_uploading), 'args': "-f " + os.path.join(self.package_path, "config/.dropbox_uploader_config") + " upload " + self.bagfile_output + " " + os.path.join(self.package_name, "data")}))
 

atf_core/scripts/merger.py

@@ -6,8 +6,6 @@
 import os
 import copy
 import json
-import yaml
-import numpy
 
 class Merger():
     def __init__(self):
@@ -28,7 +26,6 @@ def merge(self):
                 #print "subtests=", subtests
                 test_data_merged = {}
                 for subtest in subtests:
-                    subtest_data = subtests
                     #print "subtest=", subtest
                     subtest_data = self.load_data(os.path.join(self.json_output, subtest + ".json"))
                     #print "subtest_data=", subtest_data
@@ -41,7 +38,7 @@ def merge(self):
                             for metric_data in metric_data_list:
                                 #print "metric_data=", metric_data
                                 #print "metric_data['data']=", metric_data['data']
-                                
+
                                 # check if entry exists
                                 if testblock_name not in test_data_merged:
                                     # create new testblock entry
@@ -59,7 +56,7 @@ def merge(self):
                                 else:
                                     # entry already exists
                                     #print "entry for metric '" + metric_name + "' in testblock '" + testblock_name + "' already exists"
-                                    
+
                                     # check if merging is possible, if not: append
                                     is_in, element_number = self.is_in_metric_data_list(copy.deepcopy(metric_data), copy.deepcopy(test_data_merged[testblock_name][metric_name]))
                                     if is_in:
@@ -79,7 +76,7 @@ def merge(self):
                                 #print "test_data_merged=", test_data_merged
 
                 #print "test_data_merged before average=", test_data_merged
-                
+
                 # calculate min/max/average
                 for testblock_name, testblock_data in test_data_merged.items():
                     #print "testblock_data=", testblock_data
@@ -90,10 +87,10 @@ def merge(self):
                             #print "test_data_merged[testblock_name][metric_name][i]['data']['values']=", test_data_merged[testblock_name][metric_name][i]['data']['values']
                             test_data_merged[testblock_name][metric_name][i]['data']['min'] = min(test_data_merged[testblock_name][metric_name][i]['data']['values'])
                             test_data_merged[testblock_name][metric_name][i]['data']['max'] = max(test_data_merged[testblock_name][metric_name][i]['data']['values'])
-                            test_data_merged[testblock_name][metric_name][i]['data']['average'] = round(sum(test_data_merged[testblock_name][metric_name][i]['data']['values'])/len(test_data_merged[testblock_name][metric_name][i]['data']['values']),3)
-                            
+                            test_data_merged[testblock_name][metric_name][i]['data']['average'] = round(sum(test_data_merged[testblock_name][metric_name][i]['data']['values'])/len(test_data_merged[testblock_name][metric_name][i]['data']['values']), 3)
+
                 #print "test_data_merged after average=", test_data_merged
-                
+
                 # write to file
                 filename = self.json_output + test_name + ".json"
                 stream = file(filename, 'w')
@@ -107,14 +104,14 @@ def merge(self):
 
     def is_in_metric_data_list(self, data, data_list):
         counter = 0
-        for d in data_list:
-            is_same = self.is_same_metric_data_without_data(copy.deepcopy(data), copy.deepcopy(d))
+        for dat in data_list:
+            is_same = self.is_same_metric_data(copy.deepcopy(data), copy.deepcopy(dat))
             if is_same:
                 return True, counter
             counter += 1
         return False, None
 
-    def is_same_metric_data_without_data(self, data1, data2):
+    def is_same_metric_data(self, data1, data2):
         data1.pop('data')
         data1.pop('groundtruth_result')
         data2.pop('data')
@@ -130,7 +127,7 @@ def load_data(self, filename):
             return doc
 
 class TestMerging(unittest.TestCase):
-    def test_MergingResults(self):
+    def test_merging_results(self):
         merger = Merger()
         merger.merge()
         self.assertTrue(merger.result, merger.merging_error_message)

atf_core/scripts/record_all.py

@@ -1,5 +1,4 @@
 #!/usr/bin/env python
-import rospy
 import rospkg
 import sys
 import os
@@ -12,15 +11,15 @@
     print "recording all in package '" + pkg + "'"
     pkg_path = r.get_path(pkg)
     path_to_test_files = os.path.join(pkg_path, "test_generated/recording")
-    
+
     filenames = []
     for (dirpath, dirnames, list_of_files) in os.walk(path_to_test_files):
         for f in list_of_files:
             if f.endswith(".test"):
                 filenames.append(f)
 
     print "found " + str(len(filenames)) + " files: " + str(filenames)
-    
+
     # record all
     counter = 1
     for f in filenames:

atf_core/src/atf_core/atf.py

@@ -6,7 +6,7 @@
 import shutil
 import os
 
-from atf_msgs.msg import *
+from atf_msgs.msg import TestStatus, Status, TestblockStatus
 from copy import copy
 
 
@@ -121,7 +121,7 @@ def export_to_file(self):
 
         self.test_status_publisher.publish(test_status)
 
-        shutil.copy(os.path.join(rosparam.get_param("analysing/test_generated_path"), "test_list.json") , rosparam.get_param("/analysing/result_json_output"))
+        shutil.copy(os.path.join(rosparam.get_param("analysing/test_generated_path"), "test_list.json"), rosparam.get_param("/analysing/result_json_output"))
 
         filename = rosparam.get_param("/analysing/result_json_output") + self.test_name + ".json"
         stream = file(filename, 'w')

atf_core/src/atf_core/state_machine.py

@@ -39,9 +39,9 @@ def run(self):
         try:
             self.handler = self.handlers[self.startState]
         except:
-            raise "InitializationError", "Must call .set_start() before .run()"
+            raise SMError("Must call .set_start() before .run()")
         if not self.endStates:
-            raise "InitializationError", "At least one state must be an end_state"
+            raise SMError("At least one state must be an end_state")
 
         rospy.loginfo("SM for testblock '" + self.name + "' is running")
 
@@ -54,3 +54,5 @@ def run(self):
                 self.handler = self.handlers[new_state]
             rospy.loginfo("SM '" + self.name + "' in state " + str(self.get_current_state()))
         rospy.loginfo("SM '" + self.name + "' finished with state " + str(self.get_current_state()))
+class SMError(Exception):
+    pass

atf_metrics/src/atf_metrics/calculate_distance_to_obstacles.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python
 import rospy
 from atf_msgs.msg import ObstacleDistance
-from types import *
+from types import ListType
 
 
 class CalculateDistanceToObstaclesParamHandler:

atf_metrics/src/atf_metrics/calculate_interface.py

@@ -15,7 +15,7 @@ def parse_parameter(self, testblock_name, params):
         :param params: Parameter
         """
         metrics = []
-        
+
         if type(params) is not list:
             rospy.logerr("metric config not a list")
             return False

atf_metrics/src/atf_metrics/calculate_path_length.py

@@ -76,14 +76,13 @@ def purge(self):
     def record_tf(self, event):
         if self.active:
             try:
-
                 self.listener.waitForTransform(self.root_frame,
                                                self.measured_frame,
                                                rospy.Time(0),
                                                rospy.Duration.from_sec(2 / self.tf_sampling_freq))
                 (trans, rot) = self.listener.lookupTransform(self.root_frame, self.measured_frame, rospy.Time(0))
 
-            except (tf.Exception, tf.LookupException, tf.ConnectivityException, Exception), e:
+            except (tf.Exception, tf.LookupException, tf.ConnectivityException), e:
                 #rospy.logwarn(e)
                 pass
             else:

atf_metrics/src/atf_metrics/calculate_resources.py

@@ -2,7 +2,7 @@
 import numpy
 import rospy
 
-from atf_msgs.msg import *
+from atf_msgs.msg import Resources, IO, Network 
 
 
 class CalculateResourcesParamHandler:

atf_metrics/src/atf_metrics/calculate_time.py

@@ -20,7 +20,7 @@ def parse_parameter(self, testblock_name, params):
         if params == None:
             metrics.append(CalculateTime(None, None))
             return metrics
-        
+
         if type(params) is not list:
             rospy.logerr("metric config not a list")
             return False

atf_recorder/scripts/recorder_core.py

@@ -10,8 +10,8 @@
 import atf_recorder_plugins
 
 from threading import Lock
-from atf_msgs.msg import *
-from atf_recorder.srv import *
+from atf_msgs.msg import Trigger, TestStatus
+from atf_recorder.srv import RecorderCommand, RecorderCommandResponse
 from atf_recorder import BagfileWriter
 
 
@@ -58,7 +58,7 @@ def __init__(self):
         #    num_subscriber = ob_sub.get_num_connections()
 
         self.subscriber = []
-        self.topics= self.get_topics()
+        self.topics = self.get_topics()
         rospy.Timer(rospy.Duration(0.5), self.create_subscriber_callback)
         rospy.sleep(1) #wait for subscribers to get active (rospy bug?)
 
@@ -144,11 +144,10 @@ def create_subscriber_callback(self, event):
             if topic not in self.subscriber:
                 try:
                     msg_class, _, _ = rostopic.get_topic_class(topic)
-                    msg = rospy.wait_for_message(topic, msg_class)
                     rospy.Subscriber(topic, msg_class, self.global_topic_callback, callback_args=topic)
                     self.subscriber.append(topic)
                 except Exception as e:
-                    #print e 
+                    #print e
                     pass
 
     def command_callback(self, msg):

atf_recorder/src/atf_recorder/__init__.py

@@ -1,2 +1,2 @@
-from recording_manager import RecordingManager
-from recorder_tools import BagfileWriter
+from atf_recorder.recording_manager import RecordingManager
+from atf_recorder.recorder_tools import BagfileWriter

atf_recorder/src/atf_recorder/recorder_tools.py

@@ -1,7 +1,4 @@
 #!/usr/bin/env python
-import rospy
-
-
 class BagfileWriter:
     def __init__(self, bagfile, write_lock):
         """

atf_recorder/src/atf_recorder/recording_manager.py

@@ -1,8 +1,8 @@
 #!/usr/bin/env python
 import rospy
 
-from atf_recorder.srv import *
-from atf_msgs.msg import *
+from atf_recorder.srv import RecorderCommand
+from atf_msgs.msg import Trigger
 
 
 class RecordingManager:

atf_recorder_plugins/src/atf_recorder_plugins/plugin_interface.py

@@ -3,32 +3,32 @@
 import rosgraph
 import rosservice
 import socket
-import copy 
 from atf_recorder import BagfileWriter
 from rosapi.srv import Nodes, Topics, Publishers, Subscribers
 from atf_msgs.msg import Api, NodeApi, InterfaceItem
+from rosnode import ROSNodeIOException
 
 class RecordInterface:
     def __init__(self, write_lock, bag_file):
         self.rosapi_service_nodes = rospy.ServiceProxy('/rosapi/nodes', Nodes)
         self.rosapi_service_topics = rospy.ServiceProxy('/rosapi/topics', Topics)
         self.rosapi_service_publishers = rospy.ServiceProxy('/rosapi/publishers', Publishers)
         self.rosapi_service_subscribers = rospy.ServiceProxy('/rosapi/subscribers', Subscribers)
-        
+
         self.master = rosgraph.Master("/rosnode")
 
         self.BfW = BagfileWriter(bag_file, write_lock)
 
     def trigger_callback(self, msg):
         #print "msg=", msg
 
-        try: 
-            publishers, subscribers, services = self.master.getSystemState() 
+        try:
+            publishers, subscribers, services = self.master.getSystemState()
             #pub_topics = self.master.getPublishedTopics('/subscriber1')
             topic_types = self.master.getTopicTypes()
             service_types = self.get_service_types(services)
-        except socket.error: 
-            raise ROSNodeIOException("Unable to communicate with master!") 
+        except socket.error:
+            raise ROSNodeIOException("Unable to communicate with master!")
 
         #print "publishers=", publishers
         #print "subscribers=", subscribers
@@ -41,7 +41,7 @@ def trigger_callback(self, msg):
         self.add_api(api_dict, "subscribers", subscribers, topic_types)
         self.add_api(api_dict, "services", services, service_types)
         #TODO actions
-        
+
         #print "api_dict=\n", api_dict
         api = self.dict_to_msg(api_dict)
         #print "api=\n", api

atf_recorder_plugins/src/atf_recorder_plugins/plugin_resources.py

@@ -8,7 +8,7 @@
 from copy import copy
 from re import findall
 from subprocess import check_output, CalledProcessError
-from atf_msgs.msg import *
+from atf_msgs.msg import Resources, IO, Network
 from atf_recorder import BagfileWriter
 
 

atf_status_server/scripts/atf_status_server_node.py

@@ -3,8 +3,8 @@
 import yaml
 import rosparam
 
-from atf_msgs.msg import *
-from atf_status_server.srv import *
+from atf_msgs.msg import TestStatus, TestblockStatus
+from atf_status_server.srv import GetTestStatus, GetTestStatusResponse
 from threading import Lock
 
 

atf_test/scripts/application.py

@@ -1,11 +1,7 @@
 #!/usr/bin/python
 import unittest
-import yaml
-from subprocess import call
-
 import rospy
 import rostest
-import rostopic
 import tf
 import math
 from atf_recorder import RecordingManager
@@ -16,7 +12,7 @@ def __init__(self):
         self.testblock_small = RecordingManager('testblock_small')
         self.testblock_large = RecordingManager('testblock_large')
         self.testblock_all = RecordingManager('testblock_all')
-        
+
         # native app code
         self.pub_freq = 100.0 # Hz
         self.br = tf.TransformBroadcaster()