Refactoring: Introduce abstraction GroupedDiagnostics, removing duplication

This commit is contained in:
Dominik Jain authored and Dominik Jain committed 2026-04-30 15:31:27 +02:00
1 parent 5ab912ca0b
commit 371d70f87e
3 files changed
+71 -120

No files matched your search

+9 -59
View File
@@ -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 = "<file>"
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 = "<file>"
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)
+1 -1
View File
@@ -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
+61 -60
View File
@@ -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 "<file>" 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 "<file>"
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 "<file>" 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