Add linter to all source files

Author: javizqh

manager/manager/lint/linter.py

@@ -1,7 +1,8 @@
+import glob
 import re
 import os
 import subprocess
-
+import tempfile
 
 class Lint:
 
@@ -103,3 +104,51 @@ def evaluate_code(
             return final_result.strip()
         except Exception as ex:
             print(ex)
+
+    def evaluate_source_code(self, files):
+        all_files = []
+        for f in files:
+            glob_files = glob.glob(os.path.join("/workspace/code", f))
+            for g in glob_files:
+                all_files.append(g)
+
+        output = []
+
+        linter_env = os.environ.copy()
+        linter_env["PYTHONPATH"] = f"{linter_env['PYTHONPATH']}:/workspace/code"
+
+        try:
+
+            for file in all_files:
+                if file.endswith(".py"):
+                    with open(file) as f:
+                        python_code = f.read()
+
+                    # Create temp file
+                    code_file = tempfile.NamedTemporaryFile(delete=False)
+                    code_file.write(python_code.encode())
+                    code_file.seek(0)
+                    code_file.close()
+
+                    options = f"{code_file.name} --enable=similarities --disable=C0114,C0116"
+
+                    # Run pylint using subprocess
+                    result = subprocess.run(
+                        ["pylint"] + options.split(), capture_output=True, text=True, env=linter_env
+                    )
+
+                    # Process pylint exit
+                    stdout = result.stdout
+
+                    # Clean temp files
+                    if os.path.exists(code_file.name):
+                        os.remove(code_file.name)
+
+                    raw_output = stdout + "\n"
+                    cleaned_result = self.clean_pylint_output(raw_output)
+                    final_result = self.append_rating_if_missing(cleaned_result)
+                    output.append(final_result.strip())
+
+            return output
+        except Exception as ex:
+            return ex

manager/manager/manager.py

@@ -354,36 +354,6 @@ def on_prepare_tools(self, event):
         self.tools_launcher.run(self.consumer)
         LogManager.logger.info("Tools transition finished")
 
-    def add_frequency_control(self, code):
-        frequency_control_code_imports = """
-import time
-from datetime import datetime
-ideal_cycle = 20
-"""
-        code = frequency_control_code_imports + code
-        infinite_loop = re.search(
-            r"[^ ]while\s*\(\s*True\s*\)\s*:|[^ ]while\s*True\s*:|[^ ]while\s*1\s*:|[^ ]while\s*\(\s*1\s*\)\s*:",
-            code,
-        )
-        frequency_control_code_pre = """
-    start_time_internal_freq_control = datetime.now()
-            """
-        code = (
-            code[: infinite_loop.end()]
-            + frequency_control_code_pre
-            + code[infinite_loop.end() :]
-        )
-        frequency_control_code_post = """
-    finish_time_internal_freq_control = datetime.now()
-    dt = finish_time_internal_freq_control - start_time_internal_freq_control
-    ms = (dt.days * 24 * 60 * 60 + dt.seconds) * 1000 + dt.microseconds / 1000.0
-
-    if (ms < ideal_cycle):
-        time.sleep((ideal_cycle - ms) / 1000.0)
-"""
-        code = code + frequency_control_code_post
-        return code
-
     def on_style_check_application(self, event):
         """
         Handles the 'style_check' event, does not change the state and returns the current state.
@@ -616,33 +586,6 @@ def find_docker_console():
             # raise Exception("No active console other than /dev/pts/0")
             return consoles
 
-        def prepare_RA_code(code_path):
-            f = open(code_path, "r")
-            code = f.read()
-            f.close()
-
-            # Make code backwards compatible
-            code = code.replace("from GUI import GUI", "import GUI")
-            code = code.replace("from HAL import HAL", "import HAL")
-
-            # Create executable app
-            errors = self.linter.evaluate_code(code, self.ros_version)
-            if errors == "":
-
-                # code = self.add_frequency_control(code)
-                # f = open(code_path, "w")
-                # f.write(code)
-                # f.close()
-                pass
-
-            else:
-                console_path = find_docker_console()
-                for i in console_path:
-                    with open(i, "w") as console:
-                        console.write(errors + "\n\n")
-
-                raise Exception(errors)
-
         # Kill already running code
         try:
             proc = psutil.Process(self.application_process.pid)
@@ -659,6 +602,7 @@ def prepare_RA_code(code_path):
         # Extract app config
         app_cfg = event.kwargs.get("data", {})
         entrypoint = app_cfg["entrypoint"]
+        to_lint = app_cfg["linter"]
 
         # Unzip the app
         if app_cfg["code"].startswith("data:"):
@@ -668,17 +612,27 @@ def prepare_RA_code(code_path):
         zip_ref = zipfile.ZipFile("/workspace/code/app.zip", "r")
         zip_ref.extractall("/workspace/code")
         zip_ref.close()
-        print("Here")
 
         if not os.path.isfile(entrypoint):
             LogManager.logger.info("User code not found")
             raise Exception("User code not found")
 
-        try:
-            # if entrypoint == "/workspace/code/academy.py":
-            #     # TODO: temporal
-            #     prepare_RA_code(entrypoint)
+        # Pass the linter
+        errors = self.linter.evaluate_source_code(to_lint)
+        failed_linter = False
+
+        for error in errors:
+            if error != "":
+                failed_linter = True
+                console_path = find_docker_console()
+                for i in console_path:
+                    with open(i, "w") as console:
+                        console.write(errors + "\n\n")
+
+        if failed_linter:
+            raise Exception(errors)
 
+        try:
             fds = os.listdir("/dev/pts/")
             console_fd = str(max(map(int, fds[:-1])))