From 371d70f87ee0280d58ff7693ee4ad20c9b644293 Mon Sep 17 00:00:00 2001 From: Dominik Jain Date: Thu, 30 Apr 2026 11:43:23 +0200 Subject: [PATCH] Refactoring: Introduce abstraction GroupedDiagnostics, removing duplication --- src/serena/tools/symbol_tools.py | 68 +++-------------- src/serena/tools/tools_base.py | 2 +- src/serena/util/ls_diagnostics.py | 121 +++++++++++++++--------------- 3 files changed, 71 insertions(+), 120 deletions(-) diff --git a/src/serena/tools/symbol_tools.py b/src/serena/tools/symbol_tools.py index 7f9c17f5..d2626ae0 100644 --- a/src/serena/tools/symbol_tools.py +++ b/src/serena/tools/symbol_tools.py @@ -18,44 +18,8 @@ from serena.tools import ( ToolMarkerSymbolicRead, ) from serena.tools.tools_base import ToolMarkerOptional -from solidlsp import ls_types +from serena.util.ls_diagnostics import GroupedDiagnostics from solidlsp.ls_types import SymbolKind -from solidlsp.lsp_protocol_handler.lsp_types import DiagnosticSeverity - -FILE_LEVEL_DIAGNOSTIC_BUCKET = "" - - -def _diagnostic_severity_name(severity: int | None) -> str: - if severity is None: - return "Unknown" - try: - return DiagnosticSeverity(severity).name - except ValueError: - return f"Severity_{severity}" - - -def _diagnostic_output_dict(diagnostic: ls_types.Diagnostic) -> dict[str, Any]: - result: dict[str, Any] = { - "message": diagnostic["message"], - "range": diagnostic["range"], - } - if "code" in diagnostic: - result["code"] = diagnostic["code"] - if "source" in diagnostic: - result["source"] = diagnostic["source"] - return result - - -def _add_grouped_diagnostic( - grouped_result: dict[str, dict[str, dict[str, list[dict[str, Any]]]]], - relative_path: str, - severity_name: str, - name_path: str, - diagnostic: ls_types.Diagnostic, -) -> None: - grouped_result.setdefault(relative_path, {}).setdefault(severity_name, {}).setdefault(name_path, []).append( - _diagnostic_output_dict(diagnostic) - ) def _offset_to_line_and_column(text: str, offset: int) -> tuple[int, int]: @@ -663,7 +627,7 @@ class GetDiagnosticsForFileTool(Tool, ToolMarkerSymbolicRead): Gets diagnostics for a file, optionally restricted to a line range, grouped by file, severity, and containing symbol. """ - _ENABLE_DIAGNOSTICS: bool = True + FILE_LEVEL_DIAGNOSTIC_BUCKET = "" def apply( self, @@ -693,10 +657,10 @@ class GetDiagnosticsForFileTool(Tool, ToolMarkerSymbolicRead): min_severity=min_severity, ) - grouped_result: dict[str, dict[str, dict[str, list[dict[str, Any]]]]] = {} + grouped_diagnostics = GroupedDiagnostics() for diagnostic in diagnostics: diag_range = diagnostic["range"]["start"] - name_path = FILE_LEVEL_DIAGNOSTIC_BUCKET + name_path = self.FILE_LEVEL_DIAGNOSTIC_BUCKET owner_symbol = symbol_retriever.find_diagnostic_owner_symbol( relative_file_path=relative_path, line=diag_range["line"], @@ -704,15 +668,9 @@ class GetDiagnosticsForFileTool(Tool, ToolMarkerSymbolicRead): ) if owner_symbol is not None: name_path = owner_symbol.get_name_path() - _add_grouped_diagnostic( - grouped_result, - relative_path=relative_path, - severity_name=_diagnostic_severity_name(diagnostic.get("severity")), - name_path=name_path, - diagnostic=diagnostic, - ) + grouped_diagnostics.add(relative_path, name_path, diagnostic) - result = self._to_json(grouped_result) + result = self._to_json(grouped_diagnostics.get_dict()) return self._limit_length(result, max_answer_chars) @@ -721,8 +679,6 @@ class GetDiagnosticsForSymbolTool(Tool, ToolMarkerSymbolicRead, ToolMarkerOption Gets diagnostics for a symbol and, optionally, for symbols that reference it. """ - _ENABLE_DIAGNOSTICS: bool = True - def apply( self, name_path: str, @@ -752,22 +708,16 @@ class GetDiagnosticsForSymbolTool(Tool, ToolMarkerSymbolicRead, ToolMarkerOption min_severity=min_severity, ) - grouped_result: dict[str, dict[str, dict[str, list[dict[str, Any]]]]] = {} + grouped_diagnostics = GroupedDiagnostics() for symbol, diagnostics in diagnostics_by_symbol.items(): relative_path = symbol.relative_path if relative_path is None: continue symbol_name_path = symbol.get_name_path() for diagnostic in diagnostics: - _add_grouped_diagnostic( - grouped_result, - relative_path=relative_path, - severity_name=_diagnostic_severity_name(diagnostic.get("severity")), - name_path=symbol_name_path, - diagnostic=diagnostic, - ) + grouped_diagnostics.add(relative_path, symbol_name_path, diagnostic) - result = self._to_json(grouped_result) + result = self._to_json(grouped_diagnostics.get_dict()) return self._limit_length(result, max_answer_chars) diff --git a/src/serena/tools/tools_base.py b/src/serena/tools/tools_base.py index 4348deb9..565a4055 100644 --- a/src/serena/tools/tools_base.py +++ b/src/serena/tools/tools_base.py @@ -448,7 +448,7 @@ class EditingToolWithDiagnostics(Tool, ToolMarkerCanEdit): assert self._symbol_retriever is not None diagnostics_diff = DiagnosticsDiff(self._before_edit_diagnostics_snapshot, self._edited_files, self._symbol_retriever) - grouped_diagnostics = diagnostics_diff.get_grouped_diagnostics() + grouped_diagnostics = diagnostics_diff.get_grouped_diagnostics().get_dict() if not grouped_diagnostics: return base_result diff --git a/src/serena/util/ls_diagnostics.py b/src/serena/util/ls_diagnostics.py index 6df5ce34..13b505ea 100644 --- a/src/serena/util/ls_diagnostics.py +++ b/src/serena/util/ls_diagnostics.py @@ -88,6 +88,62 @@ class PublishedDiagnosticsSnapshot: self.warning_identities_by_before_path = warning_identities_by_before_path +class GroupedDiagnostics: + def __init__(self) -> None: + self._grouped_diagnostics: dict[str, dict[str, dict[str, list[dict[str, Any]]]]] = {} + + def add(self, relative_path: str, name_path: str, diagnostic: ls_types.Diagnostic) -> None: + severity_name = self._diagnostic_severity_name(diagnostic.get("severity")) + self._grouped_diagnostics.setdefault(relative_path, {}).setdefault(severity_name, {}).setdefault(name_path, []).append( + self._diagnostic_output_dict(diagnostic) + ) + + def get_dict(self) -> dict[str, dict[str, dict[str, list[dict[str, Any]]]]]: + """ + Returns a nested dictionary of the form: + { + relative_file_path: { + severity_name: { + name_path: [ + diagnostic_dict, + ... + ], + ... + }, + ... + }, + ... + } + where: + - relative_file_path is the relative path of the file containing the diagnostic + - severity_name is the name of the diagnostic severity (e.g. "Warning", "Error") + - name_path is the name path of the symbol that owns the diagnostic (or "" if no owner symbol was found) + - diagnostic_dict is a dictionary containing the diagnostic's message, range, and optionally code and source + """ + return self._grouped_diagnostics + + @staticmethod + def _diagnostic_severity_name(severity: int | None) -> str: + if severity is None: + return "Unknown" + try: + return DiagnosticSeverity(severity).name + except ValueError: + return f"Severity_{severity}" + + @staticmethod + def _diagnostic_output_dict(diagnostic: ls_types.Diagnostic) -> dict[str, Any]: + result: dict[str, Any] = { + "message": diagnostic["message"], + "range": diagnostic["range"], + } + if "code" in diagnostic: + result["code"] = diagnostic["code"] + if "source" in diagnostic: + result["source"] = diagnostic["source"] + return result + + class DiagnosticsDiff: def __init__( self, @@ -95,7 +151,7 @@ class DiagnosticsDiff: edited_files: Iterable[EditedFilePath], symbol_retriever: "LanguageServerSymbolRetriever", ): - grouped_result: dict[str, dict[str, dict[str, list[dict[str, Any]]]]] = {} + grouped_diagnostics = GroupedDiagnostics() for edited_file_path in edited_files: try: @@ -139,64 +195,9 @@ class DiagnosticsDiff: column=diagnostic_start["character"], ) name_path = owner_symbol.get_name_path() if owner_symbol is not None else "" - self._add_grouped_diagnostic(grouped_result, edited_file_path.after_relative_path, name_path, diagnostic) + grouped_diagnostics.add(edited_file_path.after_relative_path, name_path, diagnostic) - self._grouped_result = grouped_result + self._grouped_diagnostics = grouped_diagnostics - def get_grouped_diagnostics(self) -> dict[str, dict[str, dict[str, list[dict[str, Any]]]]]: - """ - Returns a nested dictionary of the form: - { - relative_file_path: { - severity_name: { - name_path: [ - diagnostic_dict, - ... - ], - ... - }, - ... - }, - ... - } - where: - - relative_file_path is the relative path of the file containing the diagnostic - - severity_name is the name of the diagnostic severity (e.g. "Warning", "Error") - - name_path is the name path of the symbol that owns the diagnostic (or "" if no owner symbol was found) - - diagnostic_dict is a dictionary containing the diagnostic's message, range, and optionally code and source - """ - return self._grouped_result - - @classmethod - def _add_grouped_diagnostic( - cls, - grouped_result: dict[str, dict[str, dict[str, list[dict[str, Any]]]]], - relative_path: str, - name_path: str, - diagnostic: ls_types.Diagnostic, - ) -> None: - severity_name = cls._diagnostic_severity_name(diagnostic.get("severity")) - grouped_result.setdefault(relative_path, {}).setdefault(severity_name, {}).setdefault(name_path, []).append( - cls._diagnostic_output_dict(diagnostic) - ) - - @staticmethod - def _diagnostic_severity_name(severity: int | None) -> str: - if severity is None: - return "Unknown" - try: - return DiagnosticSeverity(severity).name - except ValueError: - return f"Severity_{severity}" - - @staticmethod - def _diagnostic_output_dict(diagnostic: ls_types.Diagnostic) -> dict[str, Any]: - result: dict[str, Any] = { - "message": diagnostic["message"], - "range": diagnostic["range"], - } - if "code" in diagnostic: - result["code"] = diagnostic["code"] - if "source" in diagnostic: - result["source"] = diagnostic["source"] - return result + def get_grouped_diagnostics(self) -> GroupedDiagnostics: + return self._grouped_diagnostics