refactor: move diagnostics helper code to robotcode.robot package

Author: d-biehl

package.json

@@ -1,6 +1,6 @@
 {
   "name": "robotcode",
-  "displayName": "Robot Code - Robot Framework Support",
+  "displayName": "RobotCode - Robot Framework Support",
   "description": "Robot Framework IntelliSense, linting, test execution and debugging, code formatting, refactoring, and many more",
   "icon": "images/icon.png",
   "publisher": "d-biehl",
@@ -541,7 +541,7 @@
             "items": {
               "type": "string"
             },
-            "markdownDescription": "Specifies the library names that should not be cached. This is useful if you have a dynamic or hybrid library that has different keywords depending on the arguments. You can specify a glob pattern that matches the library name or the source file. \n\nExamples:\n- `**/mylibfolder/mylib.py`\n- `MyLib`\n- `mylib.subpackage.subpackage` \n\nFor robot framework internal libraries, you have to specify the full module name like `robot.libraries.Remote`.\n\nIf you change this setting, you may need to run the command `Robot Code: Clear Cache and Restart Language Servers`.",
+            "markdownDescription": "Specifies the library names that should not be cached. This is useful if you have a dynamic or hybrid library that has different keywords depending on the arguments. You can specify a glob pattern that matches the library name or the source file. \n\nExamples:\n- `**/mylibfolder/mylib.py`\n- `MyLib`\n- `mylib.subpackage.subpackage` \n\nFor robot framework internal libraries, you have to specify the full module name like `robot.libraries.Remote`.\n\nIf you change this setting, you may need to run the command `RobotCode: Clear Cache and Restart Language Servers`.",
             "scope": "resource"
           },
           "robotcode.analysis.cache.ignoredVariables": {
@@ -550,7 +550,7 @@
             "items": {
               "type": "string"
             },
-            "markdownDescription": "Specifies the variable files that should not be cached. This is useful if you have a dynamic or hybrid variable files that has different variables depending on the arguments. You can specify a glob pattern that matches the variable module name or the source file. \n\nExamples:\n- `**/variables/myvars.py`\n- `MyVariables`\n- `myvars.subpackage.subpackage` \n\nIf you change this setting, you may need to run the command `Robot Code: Clear Cache and Restart Language Servers`.",
+            "markdownDescription": "Specifies the variable files that should not be cached. This is useful if you have a dynamic or hybrid variable files that has different variables depending on the arguments. You can specify a glob pattern that matches the variable module name or the source file. \n\nExamples:\n- `**/variables/myvars.py`\n- `MyVariables`\n- `myvars.subpackage.subpackage` \n\nIf you change this setting, you may need to run the command `RobotCode: Clear Cache and Restart Language Servers`.",
             "scope": "resource"
           },
           "robotcode.run.openOutputAfterRun": {

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

@@ -30,7 +30,7 @@ def analyze(app: Application, paths: Tuple[str]) -> Union[str, int, None]:
     config_files, root_folder, _ = get_config_files(paths, app.config.config_files, verbose_callback=app.verbose)
 
     try:
-        analizer_config = load_config_from_path(
+        analyzer_config = load_config_from_path(
             AnalyzerConfig,
             *config_files,
             tool_name="robotcode-analyze",
@@ -44,7 +44,7 @@ def analyze(app: Application, paths: Tuple[str]) -> Union[str, int, None]:
     except (TypeError, ValueError) as e:
         raise click.ClickException(str(e)) from e
 
-    app.print_data(analizer_config)
+    app.print_data(analyzer_config)
     app.print_data(robot_config)
 
     return 0

packages/core/src/robotcode/core/documents_manager.py

@@ -0,0 +1,204 @@
+import os
+import re
+import threading
+from pathlib import Path
+from typing import (
+    Any,
+    Callable,
+    Dict,
+    Final,
+    Iterator,
+    List,
+    Optional,
+    Union,
+)
+
+from .event import event
+from .language import LanguageDefinition, language_id_filter
+from .lsp.types import DocumentUri
+from .text_document import TextDocument
+from .uri import Uri
+from .utils.logging import LoggingDescriptor
+
+
+class CantReadDocumentError(Exception):
+    pass
+
+
+class DocumentsManager:
+    _logger: Final = LoggingDescriptor()
+
+    def __init__(self, languages: List[LanguageDefinition]) -> None:
+        self.languages = languages
+
+        self._documents: Dict[DocumentUri, TextDocument] = {}
+        self._lock = threading.RLock()
+
+    @property
+    def documents(self) -> List[TextDocument]:
+        return list(self._documents.values())
+
+    __NORMALIZE_LINE_ENDINGS: Final = re.compile(r"(\r?\n)")
+
+    @classmethod
+    def _normalize_line_endings(cls, text: str) -> str:
+        return cls.__NORMALIZE_LINE_ENDINGS.sub("\n", text)
+
+    def read_document_text(self, uri: Uri, language_id: Union[str, Callable[[Any], bool], None]) -> str:
+        for e in self.on_read_document_text(
+            self,
+            uri,
+            callback_filter=language_id_filter(language_id) if isinstance(language_id, str) else language_id,
+        ):
+            if isinstance(e, BaseException):
+                raise RuntimeError(f"Can't read document text from {uri}: {e}") from e
+
+            if e is not None:
+                return self._normalize_line_endings(e)
+
+        raise FileNotFoundError(str(uri))
+
+    def detect_language_id(self, path_or_uri: Union[str, "os.PathLike[Any]", Uri]) -> str:
+        path = path_or_uri.to_path() if isinstance(path_or_uri, Uri) else Path(path_or_uri)
+
+        for lang in self.languages:
+            suffix = path.suffix
+            if lang.extensions_ignore_case:
+                suffix = suffix.lower()
+            if suffix in lang.extensions:
+                return lang.id
+
+        return "unknown"
+
+    @_logger.call
+    def get_or_open_document(
+        self,
+        path: Union[str, "os.PathLike[Any]"],
+        language_id: Optional[str] = None,
+        version: Optional[int] = None,
+    ) -> TextDocument:
+        uri = Uri.from_path(path).normalized()
+
+        result = self.get(uri)
+        if result is not None:
+            return result
+
+        try:
+            return self._append_document(
+                document_uri=DocumentUri(uri),
+                language_id=language_id or self.detect_language_id(path),
+                text=self.read_document_text(uri, language_id),
+                version=version,
+            )
+        except (SystemExit, KeyboardInterrupt):
+            raise
+        except BaseException as e:
+            raise CantReadDocumentError(f"Error reading document '{path}': {e!s}") from e
+
+    @event
+    def on_read_document_text(sender, uri: Uri) -> Optional[str]:
+        ...
+
+    @event
+    def did_create_uri(sender, uri: DocumentUri) -> None:
+        ...
+
+    @event
+    def did_create(sender, document: TextDocument) -> None:
+        ...
+
+    @event
+    def did_open(sender, document: TextDocument) -> None:
+        ...
+
+    @event
+    def did_close(sender, document: TextDocument, full_close: bool) -> None:
+        ...
+
+    @event
+    def did_change(sender, document: TextDocument) -> None:
+        ...
+
+    @event
+    def did_save(sender, document: TextDocument) -> None:
+        ...
+
+    def get(self, _uri: Union[DocumentUri, Uri]) -> Optional[TextDocument]:
+        with self._lock:
+            return self._documents.get(
+                str(Uri(_uri).normalized() if not isinstance(_uri, Uri) else _uri),
+                None,
+            )
+
+    def __len__(self) -> int:
+        return self._documents.__len__()
+
+    def __iter__(self) -> Iterator[DocumentUri]:
+        return self._documents.__iter__()
+
+    @event
+    def on_document_cache_invalidate(sender, document: TextDocument) -> None:
+        ...
+
+    def _on_document_cache_invalidate(self, sender: TextDocument) -> None:
+        self.on_document_cache_invalidate(self, sender)
+
+    @event
+    def on_document_cache_invalidated(sender, document: TextDocument) -> None:
+        ...
+
+    def _on_document_cache_invalidated(self, sender: TextDocument) -> None:
+        self.on_document_cache_invalidated(self, sender)
+
+    def _create_document(
+        self,
+        document_uri: DocumentUri,
+        text: str,
+        language_id: Optional[str] = None,
+        version: Optional[int] = None,
+    ) -> TextDocument:
+        result = TextDocument(
+            document_uri=document_uri,
+            language_id=language_id,
+            text=text,
+            version=version,
+        )
+
+        result.cache_invalidate.add(self._on_document_cache_invalidate)
+        result.cache_invalidated.add(self._on_document_cache_invalidated)
+
+        return result
+
+    def _append_document(
+        self,
+        document_uri: DocumentUri,
+        language_id: str,
+        text: str,
+        version: Optional[int] = None,
+    ) -> TextDocument:
+        with self._lock:
+            document = self._create_document(
+                document_uri=document_uri,
+                language_id=language_id,
+                text=text,
+                version=version,
+            )
+
+            self._documents[document_uri] = document
+
+            return document
+
+    @_logger.call
+    def close_document(self, document: TextDocument, real_close: bool = False) -> None:
+        document._version = None
+
+        if real_close:
+            with self._lock:
+                self._documents.pop(str(document.uri), None)
+
+            document.clear()
+        else:
+            if document.revert(None):
+                self.did_change(self, document, callback_filter=language_id_filter(document))
+
+        self.did_close(self, document, real_close, callback_filter=language_id_filter(document))

packages/core/src/robotcode/core/filewatcher.py

@@ -0,0 +1,77 @@
+from abc import ABC, abstractmethod
+from typing import Any, Callable, List, NamedTuple, Optional, Tuple, Union
+from uuid import uuid4
+
+from robotcode.core.event import event
+from robotcode.core.lsp.types import FileEvent, WatchKind
+
+
+class FileWatcher(NamedTuple):
+    glob_pattern: str
+    kind: Optional[WatchKind] = None
+
+
+class FileWatcherEntry:
+    def __init__(
+        self,
+        id: str,
+        callback: Callable[[Any, List[FileEvent]], None],
+        watchers: List[FileWatcher],
+    ) -> None:
+        self.id = id
+        self.callback = callback
+        self.watchers = watchers
+        self.parent: Optional[FileWatcherEntry] = None
+        self.finalizer: Any = None
+
+    @event
+    def child_callbacks(sender, changes: List[FileEvent]) -> None:
+        ...
+
+    def call_childrens(self, sender: Any, changes: List[FileEvent]) -> None:
+        self.child_callbacks(sender, changes)
+
+    def __str__(self) -> str:
+        return self.id
+
+    def __repr__(self) -> str:
+        return f"{type(self).__qualname__}(id={self.id!r}, watchers={self.watchers!r})"
+
+
+class FileWatcherManagerBase(ABC):
+    def add_file_watcher(
+        self,
+        callback: Callable[[Any, List[FileEvent]], None],
+        glob_pattern: str,
+        kind: Optional[WatchKind] = None,
+    ) -> FileWatcherEntry:
+        return self.add_file_watchers(callback, [(glob_pattern, kind)])
+
+    @abstractmethod
+    def add_file_watchers(
+        self,
+        callback: Callable[[Any, List[FileEvent]], None],
+        watchers: List[Union[FileWatcher, str, Tuple[str, Optional[WatchKind]]]],
+    ) -> FileWatcherEntry:
+        ...
+
+    @abstractmethod
+    def remove_file_watcher_entry(self, entry: FileWatcherEntry) -> None:
+        ...
+
+
+class FileWatcherManagerDummy(FileWatcherManagerBase):
+    def add_file_watchers(
+        self,
+        callback: Callable[[Any, List[FileEvent]], None],
+        watchers: List[Union[FileWatcher, str, Tuple[str, Optional[WatchKind]]]],
+    ) -> FileWatcherEntry:
+        _watchers = [
+            e if isinstance(e, FileWatcher) else FileWatcher(*e) if isinstance(e, tuple) else FileWatcher(e)
+            for e in watchers
+        ]
+
+        return FileWatcherEntry(f"dummy-{uuid4()}", callback, _watchers)
+
+    def remove_file_watcher_entry(self, entry: FileWatcherEntry) -> None:
+        pass

packages/core/src/robotcode/core/language.py

@@ -0,0 +1,40 @@
+from dataclasses import dataclass
+from typing import Any, Callable, List, Optional, TypeVar, Union
+
+from robotcode.core.text_document import TextDocument
+
+_F = TypeVar("_F", bound=Callable[..., Any])
+
+LANGUAGE_ID_ATTR = "__language_id__"
+
+
+def language_id(id: str, *ids: str) -> Callable[[_F], _F]:
+    def decorator(func: _F) -> _F:
+        setattr(func, LANGUAGE_ID_ATTR, {id, *ids})
+        return func
+
+    return decorator
+
+
+def language_id_filter(
+    language_id_or_document: Union[str, TextDocument],
+) -> Callable[[Any], bool]:
+    def filter(c: Any) -> bool:
+        return not hasattr(c, LANGUAGE_ID_ATTR) or (
+            (
+                language_id_or_document.language_id
+                if isinstance(language_id_or_document, TextDocument)
+                else language_id_or_document
+            )
+            in getattr(c, LANGUAGE_ID_ATTR)
+        )
+
+    return filter
+
+
+@dataclass
+class LanguageDefinition:
+    id: str
+    extensions: List[str]
+    extensions_ignore_case: bool = False
+    aliases: Optional[List[str]] = None

packages/core/src/robotcode/core/text_document.py

@@ -209,7 +209,7 @@ def get_lines(self) -> List[str]:
             return self._lines
 
     @event
-    async def cache_invalidate(sender) -> None:
+    def cache_invalidate(sender) -> None:
         ...
 
     @event
@@ -238,7 +238,7 @@ def __remove_cache_entry(self, ref: Any) -> None:
             if ref in self._cache:
                 self._cache.pop(ref)
 
-    def __get_cache_reference(self, entry: Callable[..., Any], /, *, add_remove: bool = True) -> weakref.ref[Any]:
+    def __get_cache_reference(self, entry: Callable[..., Any], /, *, add_remove: bool = True) -> "weakref.ref[Any]":
         if inspect.ismethod(entry):
             return weakref.WeakMethod(entry, self.__remove_cache_entry if add_remove else None)