sort diagnostics and select the closest one on navigation

plugin/core/diagnostics_storage.py

@@ -1,16 +1,24 @@
 from .protocol import Diagnostic, DiagnosticSeverity, DocumentUri
-from .typing import Callable, Iterator, List, Tuple, TypeVar
+from .typing import Callable, Iterator, List, Literal, Optional, Tuple, TypeVar
 from .url import parse_uri
 from .views import diagnostic_severity
 from collections import OrderedDict
 import functools
+import operator
+import sys
 
 ParsedUri = Tuple[str, str]
+SortOrder = Literal['asc', 'desc']
 T = TypeVar('T')
 
-
 # NOTE: OrderedDict can only be properly typed in Python >=3.8.
-class DiagnosticsStorage(OrderedDict):
+if sys.version_info >= (3, 8, 0):
+    DiagnosticsStorageItems = OrderedDict[ParsedUri, List[Diagnostic]]
+else:
+    DiagnosticsStorageItems = OrderedDict
+
+
+class DiagnosticsStorage(DiagnosticsStorageItems):
     # From the specs:
     #
     #   When a file changes it is the server’s responsibility to re-compute
@@ -36,7 +44,10 @@ def add_diagnostics_async(self, document_uri: DocumentUri, diagnostics: List[Dia
         self.move_to_end(uri)  # maintain incoming order
 
     def filter_map_diagnostics_async(
-        self, pred: Callable[[Diagnostic], bool], f: Callable[[ParsedUri, Diagnostic], T]
+        self,
+        pred: Callable[[Diagnostic], bool],
+        f: Callable[[ParsedUri, Diagnostic], T],
+        sort_order: Optional[SortOrder] = None
     ) -> Iterator[Tuple[ParsedUri, List[T]]]:
         """
         Yields `(uri, results)` items with `results` being a list of `f(diagnostic)` for each
@@ -45,19 +56,25 @@ def filter_map_diagnostics_async(
         not more than once. Items and results are ordered as they came in from the server.
         """
         for uri, diagnostics in self.items():
-            results = list(filter(None, map(functools.partial(f, uri), filter(pred, diagnostics))))  # type: List[T]
+            if sort_order:
+                self._sort_by_location(diagnostics, sort_order)
+            results = list(filter(None, map(functools.partial(f, uri), filter(pred, diagnostics))))
             if results:
                 yield uri, results
 
-    def filter_map_diagnostics_flat_async(self, pred: Callable[[Diagnostic], bool],
-                                          f: Callable[[ParsedUri, Diagnostic], T]) -> Iterator[Tuple[ParsedUri, T]]:
+    def filter_map_diagnostics_flat_async(
+        self,
+        pred: Callable[[Diagnostic], bool],
+        f: Callable[[ParsedUri, Diagnostic], T],
+        sort_order: Optional[SortOrder] = None
+    ) -> Iterator[Tuple[ParsedUri, T]]:
         """
         Flattened variant of `filter_map_diagnostics_async()`. Yields `(uri, result)` items for each
         of the `result`s per `uri` instead. Each `uri` can be yielded more than once. Items are
         grouped by `uri` and each `uri` group is guaranteed to appear not more than once. Items are
         ordered as they came in from the server.
         """
-        for uri, results in self.filter_map_diagnostics_async(pred, f):
+        for uri, results in self.filter_map_diagnostics_async(pred, f, sort_order):
             for result in results:
                 yield uri, result
 
@@ -70,17 +87,31 @@ def sum_total_errors_and_warnings_async(self) -> Tuple[int, int]:
             sum(map(severity_count(DiagnosticSeverity.Warning), self.values())),
         )
 
-    def diagnostics_by_document_uri(self, document_uri: DocumentUri) -> List[Diagnostic]:
+    def diagnostics_by_document_uri(
+        self,
+        document_uri: DocumentUri,
+        sort_order: Optional[SortOrder] = None
+    ) -> List[Diagnostic]:
         """
         Returns possibly empty list of diagnostic for `document_uri`.
         """
-        return self.get(parse_uri(document_uri), [])
+        diagnostics = self.get(parse_uri(document_uri), [])
+        if sort_order:
+            self._sort_by_location(diagnostics, sort_order)
+        return diagnostics
 
-    def diagnostics_by_parsed_uri(self, uri: ParsedUri) -> List[Diagnostic]:
+    def diagnostics_by_parsed_uri(self, uri: ParsedUri, sort_order: Optional[SortOrder] = None) -> List[Diagnostic]:
         """
         Returns possibly empty list of diagnostic for `uri`.
         """
-        return self.get(uri, [])
+        diagnostics = self.get(uri, [])
+        if sort_order:
+            self._sort_by_location(diagnostics, sort_order)
+        return diagnostics
+
+    def _sort_by_location(self, diagnostics: List[Diagnostic], sort_order: SortOrder) -> None:
+        diagnostics.sort(key=lambda d: operator.itemgetter('line', 'character')(d['range']['start']),
+                         reverse=sort_order == 'desc')
 
 
 def severity_count(severity: int) -> Callable[[List[Diagnostic]], int]:

plugin/core/windows.py

@@ -457,7 +457,9 @@ def update_diagnostics_panel_async(self) -> None:
         )  # type: OrderedDict[str, List[Tuple[str, Optional[int], Optional[str], Optional[str]]]]
         for session in self._sessions:
             for (_, path), contribution in session.diagnostics.filter_map_diagnostics_async(
-                    is_severity_included(max_severity), lambda _, diagnostic: format_diagnostic_for_panel(diagnostic)):
+                    is_severity_included(max_severity),
+                    lambda _, diagnostic: format_diagnostic_for_panel(diagnostic),
+                    sort_order='asc'):
                 seen = path in contributions
                 contributions.setdefault(path, []).extend(contribution)
                 if not seen:

plugin/goto_diagnostic.py

@@ -7,6 +7,7 @@
 from .core.protocol import DiagnosticSeverity
 from .core.protocol import DocumentUri
 from .core.protocol import Location
+from .core.protocol import Point
 from .core.registry import windows
 from .core.sessions import Session
 from .core.settings import userprefs
@@ -15,11 +16,13 @@
 from .core.url import parse_uri, unparse_uri
 from .core.views import DIAGNOSTIC_KINDS
 from .core.views import diagnostic_severity
+from .core.views import first_selection_region
 from .core.views import format_diagnostic_for_html
 from .core.views import format_diagnostic_source_and_code
 from .core.views import format_severity
 from .core.views import get_uri_and_position_from_location
 from .core.views import MissingUriError
+from .core.views import point_to_offset
 from .core.views import to_encoded_filename
 from .core.views import uri_from_view
 from abc import ABCMeta
@@ -32,6 +35,9 @@
 import sublime_plugin
 
 
+SessionIndex = int
+SelectedIndex = int
+
 PREVIEW_PANE_CSS = """
     .diagnostics {padding: 0.5em}
     .diagnostics a {color: var(--bluish)}
@@ -93,8 +99,9 @@ def input_description(self) -> str:
         return "Goto Diagnostic"
 
 
-ListItemsReturn = Union[List[str], Tuple[List[str], int], List[Tuple[str, Any]], Tuple[List[Tuple[str, Any]], int],
-                        List[sublime.ListInputItem], Tuple[List[sublime.ListInputItem], int]]
+ListItemsReturn = Union[List[str], Tuple[List[str], SelectedIndex],
+                        List[Tuple[str, Any]], Tuple[List[Tuple[str, Any]], SelectedIndex],
+                        List[sublime.ListInputItem], Tuple[List[sublime.ListInputItem], SelectedIndex]]
 
 
 class PreselectedListInputHandler(sublime_plugin.ListInputHandler, metaclass=ABCMeta):
@@ -145,7 +152,7 @@ def __init__(self, window: sublime.Window, view: sublime.View, initial_value: Op
     def name(self) -> str:
         return "uri"
 
-    def get_list_items(self) -> Tuple[List[sublime.ListInputItem], int]:
+    def get_list_items(self) -> Tuple[List[sublime.ListInputItem], SelectedIndex]:
         max_severity = userprefs().diagnostics_panel_include_severity_level
         # collect severities and location of first diagnostic per uri
         severities_per_path = OrderedDict()  # type: OrderedDict[ParsedUri, List[DiagnosticSeverity]]
@@ -221,10 +228,11 @@ def _project_path(self, parsed_uri: ParsedUri) -> str:
         return path
 
 
-class DiagnosticInputHandler(sublime_plugin.ListInputHandler):
+class DiagnosticInputHandler(PreselectedListInputHandler):
     _preview = None  # type: Optional[sublime.View]
 
     def __init__(self, window: sublime.Window, view: sublime.View, uri: DocumentUri) -> None:
+        super().__init__(window, initial_value=None)
         self.window = window
         self.view = view
         self.sessions = list(get_sessions(window))
@@ -233,29 +241,40 @@ def __init__(self, window: sublime.Window, view: sublime.View, uri: DocumentUri)
     def name(self) -> str:
         return "diagnostic"
 
-    def list_items(self) -> List[sublime.ListInputItem]:
-        list_items = []  # type: List[sublime.ListInputItem]
+    def get_list_items(self) -> Tuple[List[sublime.ListInputItem], SelectedIndex]:
         max_severity = userprefs().diagnostics_panel_include_severity_level
+        diagnostics = []  # type: List[Tuple[SessionIndex, Diagnostic]]
         for i, session in enumerate(self.sessions):
-            for diagnostic in filter(is_severity_included(max_severity),
-                                     session.diagnostics.diagnostics_by_parsed_uri(self.parsed_uri)):
-                lines = diagnostic["message"].splitlines()
-                first_line = lines[0] if lines else ""
-                if len(lines) > 1:
-                    first_line += " …"
-                text = "{}: {}".format(format_severity(diagnostic_severity(diagnostic)), first_line)
-                annotation = format_diagnostic_source_and_code(diagnostic)
-                kind = DIAGNOSTIC_KINDS[diagnostic_severity(diagnostic)]
-                list_items.append(sublime.ListInputItem(text, (i, diagnostic), annotation=annotation, kind=kind))
-        return list_items
+            for diagnostic in filter(is_severity_included(max_severity), session.diagnostics.diagnostics_by_parsed_uri(
+                    self.parsed_uri, sort_order='asc')):
+                diagnostics.append((i, diagnostic))
+        selected_index = 0
+        selection_region = first_selection_region(self.view)
+        selection_offset = selection_region.b if selection_region is not None else 0
+        list_items = []  # type: List[sublime.ListInputItem]
+        for i, diagnostic_tuple in enumerate(diagnostics):
+            diagnostic = diagnostic_tuple[1]
+            lines = diagnostic["message"].splitlines()
+            first_line = lines[0] if lines else ""
+            if len(lines) > 1:
+                first_line += " …"
+            text = "{}: {}".format(format_severity(diagnostic_severity(diagnostic)), first_line)
+            annotation = format_diagnostic_source_and_code(diagnostic)
+            kind = DIAGNOSTIC_KINDS[diagnostic_severity(diagnostic)]
+            list_items.append(sublime.ListInputItem(text, diagnostic_tuple, annotation=annotation, kind=kind))
+            # Pick as a selected index if before or equal the first selection point.
+            range_start_offset = point_to_offset(Point.from_lsp(diagnostic['range']['start']), self.view)
+            if range_start_offset <= selection_offset:
+                selected_index = i
+        return (list_items, selected_index)
 
     def placeholder(self) -> str:
         return "Select diagnostic"
 
     def next_input(self, args: dict) -> Optional[sublime_plugin.CommandInputHandler]:
         return None if args.get("diagnostic") else sublime_plugin.BackInputHandler()  # type: ignore
 
-    def confirm(self, value: Optional[Tuple[int, Diagnostic]]) -> None:
+    def confirm(self, value: Optional[Tuple[SessionIndex, Diagnostic]]) -> None:
         if not value:
             return
         i, diagnostic = value
@@ -272,7 +291,7 @@ def cancel(self) -> None:
             self._preview.close()
         self.window.focus_view(self.view)
 
-    def preview(self, value: Optional[Tuple[int, Diagnostic]]) -> Union[str, sublime.Html]:
+    def preview(self, value: Optional[Tuple[SessionIndex, Diagnostic]]) -> Union[str, sublime.Html]:
         if not value:
             return ""
         i, diagnostic = value