feat: enhance diagnostics and analysis features with progress indicators in verbose mode and performance improvements

Author: d-biehl

.vscode/launch.json

@@ -40,6 +40,13 @@
 			"processId": "${command:pickProcess}",
 			"justMyCode": false
 		},
+		{
+			"name": "Python: Attach using Process",
+			"type": "debugpy",
+			"request": "attach",
+			"processId": 41720,
+			"justMyCode": false
+		},
 		{
 			"name": "Python: Create Cmd Line Doc",
 			"type": "debugpy",

packages/analyze/src/robotcode/analyze/code/cli.py

@@ -1,3 +1,5 @@
+import functools
+import time
 from enum import Flag
 from pathlib import Path
 from textwrap import indent
@@ -36,43 +38,73 @@ class ReturnCode(Flag):
     HINTS = 8
 
 
-class Statistic:
+class ResultCollector:
     def __init__(self, exit_code_mask: ExitCodeMask) -> None:
         self.exit_code_mask = exit_code_mask
         self._folders: Set[WorkspaceFolder] = set()
         self._files: Set[TextDocument] = set()
-        self._diagnostics: List[Union[DocumentDiagnosticReport, FolderDiagnosticReport]] = []
+        self.diagnostics: List[Union[DocumentDiagnosticReport, FolderDiagnosticReport]] = []
+        self._start_time = time.time()
+        self._end_time = self._start_time
+
+    @staticmethod
+    def _format_duration(seconds: float) -> str:
+        total_seconds = max(0.0, seconds)
+
+        hours, remainder = divmod(total_seconds, 3600.0)
+        minutes, seconds_remainder = divmod(remainder, 60.0)
+
+        hours_int = int(hours)
+        minutes_int = int(minutes)
+
+        if hours_int > 0:
+            return f"{hours_int}h {minutes_int}m {seconds_remainder:.2f}s"
+        if minutes_int > 0:
+            return f"{minutes_int}m {seconds_remainder:.2f}s"
+        return f"{seconds_remainder:.2f}s"
+
+    def start(self) -> float:
+        self._start_time = time.time()
+        return self._start_time
+
+    def stop(self) -> float:
+        self._end_time = time.time()
+        return self._end_time
 
     @property
+    def elapsed(self) -> float:
+        return self._end_time - self._start_time
+
+    @functools.cached_property
     def errors(self) -> int:
         return sum(
-            len([i for i in e.items if i.severity == DiagnosticSeverity.ERROR]) for e in self._diagnostics if e.items
+            len([i for i in e.items if i.severity == DiagnosticSeverity.ERROR]) for e in self.diagnostics if e.items
         )
 
-    @property
+    @functools.cached_property
     def warnings(self) -> int:
         return sum(
-            len([i for i in e.items if i.severity == DiagnosticSeverity.WARNING]) for e in self._diagnostics if e.items
+            len([i for i in e.items if i.severity == DiagnosticSeverity.WARNING]) for e in self.diagnostics if e.items
         )
 
-    @property
+    @functools.cached_property
     def infos(self) -> int:
         return sum(
             len([i for i in e.items if i.severity == DiagnosticSeverity.INFORMATION])
-            for e in self._diagnostics
+            for e in self.diagnostics
             if e.items
         )
 
-    @property
+    @functools.cached_property
     def hints(self) -> int:
         return sum(
-            len([i for i in e.items if i.severity == DiagnosticSeverity.HINT]) for e in self._diagnostics if e.items
+            len([i for i in e.items if i.severity == DiagnosticSeverity.HINT]) for e in self.diagnostics if e.items
         )
 
     def add_diagnostics_report(
         self, diagnostics_report: Union[DocumentDiagnosticReport, FolderDiagnosticReport]
     ) -> None:
-        self._diagnostics.append(diagnostics_report)
+        self.diagnostics.append(diagnostics_report)
 
         if isinstance(diagnostics_report, FolderDiagnosticReport):
             self._folders.add(diagnostics_report.folder)
@@ -83,6 +115,7 @@ def __str__(self) -> str:
         return (
             f"Files: {len(self._files)}, Errors: {self.errors}, Warnings: {self.warnings}, "
             f"Infos: {self.infos}, Hints: {self.hints}"
+            f" (in {self._format_duration(self.elapsed)})"
         )
 
     def calculate_return_code(self) -> ReturnCode:
@@ -363,32 +396,37 @@ def code(
         app.verbose(f"Using analyzer_config: {analyzer_config}")
         app.verbose(f"Using exit code mask: {mask}")
 
-        statistics = Statistic(mask)
-        for e in CodeAnalyzer(
-            app=app,
-            analysis_config=analyzer_config.to_workspace_analysis_config(),
-            robot_profile=robot_profile,
-            root_folder=root_folder,
-        ).run(paths=paths, filter=filter):
-            statistics.add_diagnostics_report(e)
-
-            if isinstance(e, FolderDiagnosticReport):
-                if e.items:
-                    _print_diagnostics(app, root_folder, e.items, e.folder.uri.to_path())
-            elif isinstance(e, DocumentDiagnosticReport):
-                doc_path = (
-                    e.document.uri.to_path().relative_to(root_folder) if root_folder else e.document.uri.to_path()
-                )
-                if e.items:
-                    _print_diagnostics(app, root_folder, e.items, doc_path)
+        result_collector = ResultCollector(mask)
+        result_collector.start()
+        try:
+            for e in CodeAnalyzer(
+                app=app,
+                analysis_config=analyzer_config.to_workspace_analysis_config(),
+                robot_profile=robot_profile,
+                root_folder=root_folder,
+            ).run(paths=paths, filter=filter):
+                result_collector.add_diagnostics_report(e)
+
+            for e in result_collector.diagnostics:
+                if isinstance(e, FolderDiagnosticReport):
+                    if e.items:
+                        _print_diagnostics(app, root_folder, e.items, e.folder.uri.to_path())
+                elif isinstance(e, DocumentDiagnosticReport):
+                    doc_path = (
+                        e.document.uri.to_path().relative_to(root_folder) if root_folder else e.document.uri.to_path()
+                    )
+                    if e.items:
+                        _print_diagnostics(app, root_folder, e.items, doc_path)
+        finally:
+            result_collector.stop()
 
-        statistics_str = str(statistics)
-        if statistics.errors > 0:
+        statistics_str = str(result_collector)
+        if result_collector.errors > 0:
             statistics_str = click.style(statistics_str, fg="red")
 
         app.echo(statistics_str)
 
-        app.exit(statistics.calculate_return_code().value)
+        app.exit(result_collector.calculate_return_code().value)
 
     except (TypeError, ValueError) as e:
         raise click.ClickException(str(e)) from e

packages/analyze/src/robotcode/analyze/code/code_analyzer.py

@@ -98,35 +98,35 @@ def run(
 
             documents = self.collect_documents(folder, paths=paths, filter=filter)
 
-            self.app.verbose(f"Analyzing {len(documents)} documents")
-            for document in documents:
-                analyze_result = self.diagnostics.analyze_document(document)
-                if analyze_result is not None:
-                    diagnostics = []
-                    for item in analyze_result:
-                        if item is None:
-                            continue
-                        elif isinstance(item, BaseException):
-                            self.app.error(f"Error analyzing {document.uri.to_path()}: {item}")
-                        else:
-                            diagnostics.extend(item)
-
-                    yield DocumentDiagnosticReport(document, diagnostics)
-
-            self.app.verbose(f"Collect Diagnostics for {len(documents)} documents")
-            for document in documents:
-                analyze_result = self.diagnostics.collect_diagnostics(document)
-                if analyze_result is not None:
-                    diagnostics = []
-                    for item in analyze_result:
-                        if item is None:
-                            continue
-                        elif isinstance(item, BaseException):
-                            self.app.error(f"Error collecting diagnostics for {document.uri.to_path()}: {item}")
-                        else:
-                            diagnostics.extend(item)
-
-                    yield DocumentDiagnosticReport(document, diagnostics)
+            with self.app.progressbar(documents, label="Analyzing Documents") as progressbar:
+                for document in progressbar:
+                    analyze_result = self.diagnostics.analyze_document(document)
+                    if analyze_result is not None:
+                        diagnostics = []
+                        for item in analyze_result:
+                            if item is None:
+                                continue
+                            elif isinstance(item, BaseException):
+                                self.app.error(f"Error analyzing {document.uri.to_path()}: {item}")
+                            else:
+                                diagnostics.extend(item)
+
+                        yield DocumentDiagnosticReport(document, diagnostics)
+
+            with self.app.progressbar(documents, label="Collecting Diagnostics") as progressbar:
+                for document in progressbar:
+                    analyze_result = self.diagnostics.collect_diagnostics(document)
+                    if analyze_result is not None:
+                        diagnostics = []
+                        for item in analyze_result:
+                            if item is None:
+                                continue
+                            elif isinstance(item, BaseException):
+                                self.app.error(f"Error collecting diagnostics for {document.uri.to_path()}: {item}")
+                            else:
+                                diagnostics.extend(item)
+
+                        yield DocumentDiagnosticReport(document, diagnostics)
 
     def collect_documents(
         self, folder: WorkspaceFolder, paths: Iterable[Path] = {}, filter: Iterable[str] = {}
@@ -139,8 +139,10 @@ def collect_documents(
         documents: List[TextDocument] = []
 
         self.app.verbose(f"Collecting files in workspace folder '{folder.uri.to_path()}'")
+        file_counter = 0
         for handler in self.language_handlers:
             for file in handler.collect_workspace_folder_files(folder):
+                file_counter += 1
                 try:
                     document = self.workspace.documents.get_or_open_document(file)
                     document_path = normalized_path(document.uri.to_path()).as_posix()
@@ -158,4 +160,7 @@ def collect_documents(
                     self.app.error(f"Error reading {file}: {e}")
                     continue
 
+        self.app.verbose(f"{file_counter} files in workspace folder '{folder.uri.to_path()}'")
+        self.app.verbose(f"Collected {len(documents)} files for analyzing in workspace folder '{folder.uri.to_path()}'")
+
         return documents

packages/analyze/src/robotcode/analyze/code/diagnostics_context.py

@@ -17,29 +17,29 @@ def document_analyzers(sender, document: TextDocument) -> Optional[List[Diagnost
     def folder_analyzers(sender, folder: WorkspaceFolder) -> Optional[List[Diagnostic]]: ...
 
     @event
-    def collectors(sender, document: TextDocument) -> Optional[List[Diagnostic]]: ...
+    def document_collectors(sender, document: TextDocument) -> Optional[List[Diagnostic]]: ...
 
     def analyze_folder(self, folder: WorkspaceFolder) -> List[Union[List[Diagnostic], BaseException, None]]:
         return self.folder_analyzers(
             self,
             folder,
-            return_exceptions=True,
+            return_exceptions=False,
         )
 
     def analyze_document(self, document: TextDocument) -> List[Union[List[Diagnostic], BaseException, None]]:
         return self.document_analyzers(
             self,
             document,
             callback_filter=language_id_filter(document),
-            return_exceptions=True,
+            return_exceptions=False,
         )
 
     def collect_diagnostics(self, document: TextDocument) -> List[Union[List[Diagnostic], BaseException, None]]:
-        return self.collectors(
+        return self.document_collectors(
             self,
             document,
             callback_filter=language_id_filter(document),
-            return_exceptions=True,
+            return_exceptions=False,
         )