Christoph Brandau
757b885f28
Add PSC (.psc) indexing and share TclOO class metadata across files so class definitions discovered via PSC layers can be used for completions, signature help, inlay hints, and "go to definition". Key behavior changes: - Client file watcher now includes *.psc and .vscode launch paths/tests updated to use the postprocessor test folder; .gitignore updated to ignore that folder. - Server watches .psc changes and refreshes a PSC script index; new tools/tcloo_navigation.py exposes tcloo_definition used by the language server to resolve cross-file class/constructor/method definitions. - Language server uses class_snapshot(document.path) when producing TclOO completions, signature help, and inlay hints so resolved class metadata is available across files. Also includes related docs/changelog updates, minor code formatting cleanups, and added tests for PSC/TclOO behavior.
765 lines
31 KiB
Python
765 lines
31 KiB
Python
import logging
|
|
import os
|
|
import pathlib
|
|
import threading
|
|
import xml.etree.ElementTree as ET
|
|
from typing import List, Optional, Tuple
|
|
|
|
import lsprotocol.types as lsp
|
|
from plugins.poco_plugin import commands
|
|
from pygls import uris
|
|
from pygls.lsp.server import LanguageServer
|
|
from pygls.workspace.text_document import TextDocument
|
|
from tclint.format import FormatterOpts
|
|
from tclint.lexer import TclSyntaxError
|
|
from tclint.violations import Violation
|
|
from tools import checks, parser
|
|
from tools.completion_items import CompletionCollector
|
|
from tools.tcloo_symbols import class_completion_items
|
|
from tools.tcloo_completion import indexed_classes
|
|
from tools.file_sourcing import get_all_psc_files, psc_script_files
|
|
from tools.formatter import NxFormatter as Formatter
|
|
from tools.inlay_hint import InlayHintSignature, build_custom_inlay_signatures
|
|
from tools.navigation import FileSymbolIndex, build_file_symbol_index
|
|
from tools.proc_docs import build_proc_docs
|
|
from tools.variable_index import ProcRange, build_variable_index
|
|
|
|
DIAGNOSTIC_SOURCE = "nx-post-support"
|
|
LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
class TclLanguageServer(LanguageServer):
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self.parser = parser.CustomParser()
|
|
for command in commands:
|
|
self.parser._commands.update(command)
|
|
self.diagnostics = {}
|
|
self.poco_completion: dict = {}
|
|
self.proc_signatures: dict = {}
|
|
self.proc_docs: dict = {}
|
|
self.class_indexes: dict = {}
|
|
self.psc_script_paths: list[str] = []
|
|
self._psc_files: dict[str, list[pathlib.Path]] = {}
|
|
self._psc_lock = threading.RLock()
|
|
self.navigation_indexes: dict[str, FileSymbolIndex] = {}
|
|
self.variable_indexes: dict[
|
|
str,
|
|
tuple[
|
|
int | None,
|
|
tuple[set[str], dict[str, set[str]], list[ProcRange]],
|
|
],
|
|
] = {}
|
|
# Cache: (uri, version) -> (tree, violations)
|
|
self._ast_cache = {}
|
|
self._line_cache: dict[tuple[str, int | None], tuple[str, ...]] = {}
|
|
self._parser_lock = threading.RLock()
|
|
self._index_lock = threading.RLock()
|
|
self._index_tokens: dict[str, int] = {}
|
|
self._index_versions: dict[str, int | None] = {}
|
|
self._committed_index_versions: dict[str, int | None] = {}
|
|
self._next_index_token = 0
|
|
self._diagnostic_tokens: dict[str, int] = {}
|
|
self._next_diagnostic_token = 0
|
|
self._index_generation = 0
|
|
self._workspace_completion_cache: tuple[int, tuple] = (-1, ())
|
|
self._custom_function_names_cache: tuple[int, frozenset[str]] = (
|
|
-1,
|
|
frozenset(),
|
|
)
|
|
self._proc_metadata_cache: dict[
|
|
str, tuple[int, dict[str, list[str]], dict[str, str]]
|
|
] = {}
|
|
self._custom_inlay_cache: dict[
|
|
str, tuple[int, dict[str, InlayHintSignature]]
|
|
] = {}
|
|
self._analysis_lock = threading.RLock()
|
|
self._analysis_timers: dict[str, threading.Timer] = {}
|
|
self._analysis_tokens: dict[str, int] = {}
|
|
self._next_analysis_token = 0
|
|
|
|
def _parse_source(self, source: str):
|
|
self.parser.violations = []
|
|
tree = self.parser.parse(source)
|
|
return tree, list(self.parser.violations)
|
|
|
|
def parse_source(self, source: str):
|
|
"""Parse without retaining an AST, serialized around the shared parser."""
|
|
with self._parser_lock:
|
|
tree, _ = self._parse_source(source)
|
|
return tree
|
|
|
|
def get_tree(self, document: TextDocument):
|
|
key = (document.uri, document.version)
|
|
with self._parser_lock:
|
|
cached = self._ast_cache.get(key)
|
|
if cached is not None:
|
|
return cached[0]
|
|
tree, violations = self._parse_source(document.source)
|
|
self._ast_cache[key] = (tree, violations)
|
|
return tree
|
|
|
|
def get_tree_and_violations(self, document: TextDocument):
|
|
key = (document.uri, document.version)
|
|
with self._parser_lock:
|
|
cached = self._ast_cache.get(key)
|
|
if cached is not None:
|
|
return cached
|
|
tree, violations = self._parse_source(document.source)
|
|
self._ast_cache[key] = (tree, violations)
|
|
return tree, violations
|
|
|
|
def get_lines(self, document: TextDocument) -> tuple[str, ...]:
|
|
"""Return split source lines once per document version."""
|
|
key = (document.uri, document.version)
|
|
with self._parser_lock:
|
|
lines = self._line_cache.get(key)
|
|
if lines is None:
|
|
lines = tuple(document.source.splitlines())
|
|
self._line_cache[key] = lines
|
|
return lines
|
|
|
|
def clear_cache_for_uri(self, uri: str):
|
|
with self._parser_lock:
|
|
to_delete = [key for key in self._ast_cache if key[0] == uri]
|
|
for key in to_delete:
|
|
del self._ast_cache[key]
|
|
for key in list(self._line_cache):
|
|
if key[0] == uri:
|
|
del self._line_cache[key]
|
|
|
|
@staticmethod
|
|
def _normalized_path(path: pathlib.Path | str) -> str:
|
|
return os.path.normcase(os.path.abspath(os.fspath(path)))
|
|
|
|
@classmethod
|
|
def _is_same_or_child(
|
|
cls, candidate: pathlib.Path | str, parent: pathlib.Path | str
|
|
) -> bool:
|
|
candidate_path = cls._normalized_path(candidate)
|
|
parent_path = cls._normalized_path(parent)
|
|
try:
|
|
return os.path.commonpath([candidate_path, parent_path]) == parent_path
|
|
except ValueError:
|
|
return False
|
|
|
|
@classmethod
|
|
def paths_equal(
|
|
cls, first: pathlib.Path | str, second: pathlib.Path | str
|
|
) -> bool:
|
|
return cls._normalized_path(first) == cls._normalized_path(second)
|
|
|
|
def _invalidate_workspace_caches_locked(self) -> None:
|
|
"""Invalidate request-level aggregates after an index mutation."""
|
|
self._index_generation += 1
|
|
self._workspace_completion_cache = (-1, ())
|
|
self._custom_function_names_cache = (-1, frozenset())
|
|
self._proc_metadata_cache.clear()
|
|
self._custom_inlay_cache.clear()
|
|
|
|
def completion_items_snapshot(self) -> tuple:
|
|
"""Return de-duplicated workspace completion items, cached by generation."""
|
|
with self._index_lock:
|
|
generation, items = self._workspace_completion_cache
|
|
if generation == self._index_generation:
|
|
return items
|
|
|
|
merged = []
|
|
seen = set()
|
|
for path_items in self.poco_completion.values():
|
|
for item in path_items:
|
|
key = (item.label, getattr(item, "kind", None))
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
merged.append(item)
|
|
|
|
items = tuple(merged)
|
|
self._workspace_completion_cache = (self._index_generation, items)
|
|
return items
|
|
|
|
def class_snapshot(self, current_path) -> dict:
|
|
"""Share class metadata, without stale definitions from the active file.
|
|
|
|
PSC scripts override other workspace files in their listed load order.
|
|
The request parser adds the current document's declarations last.
|
|
"""
|
|
with self._index_lock:
|
|
indexes = {self._normalized_path(path): classes for path, classes in self.class_indexes.items()}
|
|
paths = sorted(self.class_indexes, key=str.casefold)
|
|
psc_paths = {self._normalized_path(path) for path in self.psc_script_paths}
|
|
paths = [path for path in paths if self._normalized_path(path) not in psc_paths]
|
|
paths.extend(self.psc_script_paths)
|
|
classes = {}
|
|
for path in paths:
|
|
if not self.paths_equal(path, current_path):
|
|
classes.update(indexes.get(self._normalized_path(path), {}))
|
|
return classes
|
|
|
|
def refresh_psc_scripts(self, roots, report=LOGGER.warning):
|
|
"""Index PSC dependencies through the same pipeline as workspace procs."""
|
|
with self._psc_lock:
|
|
discovered = {}
|
|
for root in roots:
|
|
for psc in get_all_psc_files(root):
|
|
try:
|
|
discovered[str(psc)] = psc_script_files(psc)
|
|
except (OSError, ET.ParseError) as error:
|
|
report(f"Could not read PSC {psc}: {error}")
|
|
discovered[str(psc)] = self._psc_files.get(str(psc), [])
|
|
paths = [str(path) for scripts in discovered.values() for path in scripts]
|
|
with self._index_lock:
|
|
previous = set(self.psc_script_paths)
|
|
self.psc_script_paths = paths
|
|
self._psc_files = discovered
|
|
try:
|
|
open_documents = {
|
|
self._normalized_path(document.path): document
|
|
for document in self.workspace.text_documents.values()
|
|
}
|
|
except RuntimeError:
|
|
open_documents = {}
|
|
for removed in previous - set(paths):
|
|
path = pathlib.Path(removed)
|
|
if (not any(self._is_same_or_child(path, root) for root in roots)
|
|
and self._normalized_path(path) not in open_documents):
|
|
self.remove_file_state(path.as_uri())
|
|
for path_string in dict.fromkeys(paths):
|
|
path = pathlib.Path(path_string)
|
|
uri = path.as_uri()
|
|
document = open_documents.get(self._normalized_path(path))
|
|
if document is None and not path.is_file():
|
|
self.remove_file_state(uri)
|
|
report(f"PSC script not found: {path}")
|
|
continue
|
|
try:
|
|
if document is None:
|
|
data = path.read_bytes()
|
|
try:
|
|
source = data.decode("utf-8-sig")
|
|
except UnicodeDecodeError:
|
|
# Older Windows NX layers use the ANSI code page.
|
|
source = data.decode("cp1252")
|
|
document = TextDocument(uri=uri, source=source, language_id="tcl")
|
|
if not self.update_poco_completion_for_file(document, cache_tree=False):
|
|
report(f"Could not index PSC script: {path}")
|
|
except (OSError, UnicodeError) as error:
|
|
report(f"Could not read PSC script {path}: {error}")
|
|
|
|
def completion_items_by_file_snapshot(
|
|
self,
|
|
) -> dict[str, tuple[lsp.CompletionItem, ...]]:
|
|
"""Return completion items grouped by source file for request ranking."""
|
|
with self._index_lock:
|
|
return {
|
|
path: tuple(items) for path, items in self.poco_completion.items()
|
|
}
|
|
|
|
def custom_function_names_snapshot(self) -> frozenset[str]:
|
|
"""Return custom completion labels for semantic highlighting."""
|
|
with self._index_lock:
|
|
generation, names = self._custom_function_names_cache
|
|
if generation == self._index_generation:
|
|
return names
|
|
|
|
names = frozenset(
|
|
item.label
|
|
for path_items in self.poco_completion.values()
|
|
for item in path_items
|
|
if item.kind != lsp.CompletionItemKind.Class
|
|
)
|
|
self._custom_function_names_cache = (self._index_generation, names)
|
|
return names
|
|
|
|
def proc_metadata_snapshot(
|
|
self, current_path: pathlib.Path | str
|
|
) -> tuple[dict[str, list[str]], dict[str, str]]:
|
|
"""Return merged proc metadata, preferring declarations in the active file."""
|
|
normalized_current = self._normalized_path(current_path)
|
|
with self._index_lock:
|
|
cached = self._proc_metadata_cache.get(normalized_current)
|
|
if cached is not None and cached[0] == self._index_generation:
|
|
return cached[1], cached[2]
|
|
|
|
signatures: dict[str, list[str]] = {}
|
|
docs: dict[str, str] = {}
|
|
signature_paths = sorted(
|
|
self.proc_signatures,
|
|
key=lambda path: self._normalized_path(path).casefold(),
|
|
)
|
|
doc_paths = sorted(
|
|
self.proc_docs,
|
|
key=lambda path: self._normalized_path(path).casefold(),
|
|
)
|
|
|
|
for path in signature_paths:
|
|
if self._normalized_path(path) != normalized_current:
|
|
signatures.update(self.proc_signatures[path])
|
|
for path in signature_paths:
|
|
if self._normalized_path(path) == normalized_current:
|
|
signatures.update(self.proc_signatures[path])
|
|
|
|
for path in doc_paths:
|
|
if self._normalized_path(path) != normalized_current:
|
|
docs.update(self.proc_docs[path])
|
|
for path in doc_paths:
|
|
if self._normalized_path(path) == normalized_current:
|
|
docs.update(self.proc_docs[path])
|
|
|
|
cached_value = (self._index_generation, signatures, docs)
|
|
self._proc_metadata_cache[normalized_current] = cached_value
|
|
return signatures, docs
|
|
|
|
def proc_documentation(
|
|
self, name: str, current_path: pathlib.Path | str
|
|
) -> str | None:
|
|
_, docs = self.proc_metadata_snapshot(current_path)
|
|
return docs.get(name)
|
|
|
|
def custom_inlay_signatures_snapshot(
|
|
self, current_path: pathlib.Path | str
|
|
) -> dict[str, InlayHintSignature]:
|
|
"""Return custom inlay signatures cached until the workspace index changes."""
|
|
normalized_current = self._normalized_path(current_path)
|
|
with self._index_lock:
|
|
cached = self._custom_inlay_cache.get(normalized_current)
|
|
if cached is not None and cached[0] == self._index_generation:
|
|
return cached[1]
|
|
|
|
signatures = build_custom_inlay_signatures(
|
|
self.proc_signatures,
|
|
self.proc_docs,
|
|
self.navigation_indexes,
|
|
os.fspath(current_path),
|
|
)
|
|
self._custom_inlay_cache[normalized_current] = (
|
|
self._index_generation,
|
|
signatures,
|
|
)
|
|
return signatures
|
|
|
|
def variable_index_for_document(
|
|
self, document: TextDocument, tree=None
|
|
) -> tuple[set[str], dict[str, set[str]], list[ProcRange]]:
|
|
"""Return the per-version variable index used by completion requests."""
|
|
filepath = str(pathlib.Path(uris.to_fs_path(document.uri)))
|
|
with self._index_lock:
|
|
cached = self.variable_indexes.get(filepath)
|
|
if cached is not None and cached[0] == document.version:
|
|
return cached[1]
|
|
|
|
if tree is None:
|
|
tree = self.get_tree(document)
|
|
variable_index = build_variable_index(document.source, tree)
|
|
|
|
with self._index_lock:
|
|
cached = self.variable_indexes.get(filepath)
|
|
if (
|
|
cached is None
|
|
or cached[0] is None
|
|
or document.version is None
|
|
or cached[0] <= document.version
|
|
):
|
|
self.variable_indexes[filepath] = (document.version, variable_index)
|
|
return variable_index
|
|
return cached[1]
|
|
|
|
def index_is_current(self, document: TextDocument) -> bool:
|
|
filepath = str(pathlib.Path(uris.to_fs_path(document.uri)))
|
|
with self._index_lock:
|
|
return (
|
|
filepath in self._committed_index_versions
|
|
and self._committed_index_versions[filepath] == document.version
|
|
)
|
|
|
|
def index_update_pending(self, filepath: pathlib.Path | str) -> bool:
|
|
filepath = os.fspath(filepath)
|
|
with self._index_lock:
|
|
return (
|
|
filepath not in self._committed_index_versions
|
|
or self._index_versions.get(filepath)
|
|
!= self._committed_index_versions[filepath]
|
|
)
|
|
|
|
def _cancel_document_analysis(self, uri: str) -> None:
|
|
with self._analysis_lock:
|
|
timer = self._analysis_timers.pop(uri, None)
|
|
self._analysis_tokens.pop(uri, None)
|
|
if timer is not None:
|
|
timer.cancel()
|
|
|
|
def cancel_analysis_under_uri(self, uri: str) -> None:
|
|
"""Cancel delayed analysis for a closed/deleted file or folder."""
|
|
try:
|
|
target = pathlib.Path(uris.to_fs_path(uri))
|
|
except (TypeError, ValueError):
|
|
self._cancel_document_analysis(uri)
|
|
return
|
|
|
|
with self._analysis_lock:
|
|
matching_uris = []
|
|
for pending_uri in self._analysis_timers:
|
|
try:
|
|
pending_path = pathlib.Path(uris.to_fs_path(pending_uri))
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if self._is_same_or_child(pending_path, target):
|
|
matching_uris.append(pending_uri)
|
|
|
|
timers = [self._analysis_timers.pop(key) for key in matching_uris]
|
|
for key in matching_uris:
|
|
self._analysis_tokens.pop(key, None)
|
|
|
|
for timer in timers:
|
|
timer.cancel()
|
|
|
|
def _invalidate_document_work(self, document: TextDocument) -> None:
|
|
"""Prevent older diagnostic/index work from committing after a new edit."""
|
|
filepath = str(pathlib.Path(uris.to_fs_path(document.uri)))
|
|
with self._index_lock:
|
|
self._next_diagnostic_token += 1
|
|
self._diagnostic_tokens[document.uri] = self._next_diagnostic_token
|
|
self._next_index_token += 1
|
|
self._index_tokens[filepath] = self._next_index_token
|
|
self._index_versions[filepath] = document.version
|
|
|
|
def schedule_document_analysis(
|
|
self, document: TextDocument, delay_seconds: float = 0.15
|
|
) -> None:
|
|
"""Coalesce rapid edits and analyze only the latest immutable snapshot."""
|
|
snapshot = TextDocument(
|
|
uri=document.uri,
|
|
source=document.source,
|
|
version=document.version,
|
|
language_id=document.language_id,
|
|
)
|
|
self._invalidate_document_work(snapshot)
|
|
|
|
with self._analysis_lock:
|
|
previous = self._analysis_timers.pop(snapshot.uri, None)
|
|
if previous is not None:
|
|
previous.cancel()
|
|
|
|
self._next_analysis_token += 1
|
|
token = self._next_analysis_token
|
|
self._analysis_tokens[snapshot.uri] = token
|
|
|
|
def analyze() -> None:
|
|
with self._analysis_lock:
|
|
if self._analysis_tokens.get(snapshot.uri) != token:
|
|
return
|
|
|
|
try:
|
|
diagnostic_state = self.diagnostic_snapshot(snapshot.uri)
|
|
if (
|
|
diagnostic_state is None
|
|
or diagnostic_state[0] != snapshot.version
|
|
):
|
|
self.compute_diagnostics(snapshot)
|
|
with self._analysis_lock:
|
|
if self._analysis_tokens.get(snapshot.uri) != token:
|
|
return
|
|
if not self.index_is_current(snapshot):
|
|
self.update_poco_completion_for_file(snapshot)
|
|
except Exception:
|
|
LOGGER.exception("Delayed analysis failed for %s", snapshot.uri)
|
|
finally:
|
|
with self._analysis_lock:
|
|
if self._analysis_tokens.get(snapshot.uri) == token:
|
|
self._analysis_tokens.pop(snapshot.uri, None)
|
|
self._analysis_timers.pop(snapshot.uri, None)
|
|
|
|
timer = threading.Timer(delay_seconds, analyze)
|
|
timer.daemon = True
|
|
self._analysis_timers[snapshot.uri] = timer
|
|
timer.start()
|
|
|
|
def analyze_document_now(self, document: TextDocument) -> None:
|
|
"""Cancel delayed work and synchronously analyze the current document."""
|
|
self._cancel_document_analysis(document.uri)
|
|
self._invalidate_document_work(document)
|
|
diagnostic_state = self.diagnostic_snapshot(document.uri)
|
|
if diagnostic_state is None or diagnostic_state[0] != document.version:
|
|
self.compute_diagnostics(document)
|
|
if not self.index_is_current(document):
|
|
self.update_poco_completion_for_file(document)
|
|
|
|
def index_snapshot(self) -> tuple[dict, dict, dict]:
|
|
"""Return stable copies for request handlers running beside the indexer."""
|
|
with self._index_lock:
|
|
return (
|
|
{path: list(items) for path, items in self.poco_completion.items()},
|
|
{
|
|
path: dict(signatures)
|
|
for path, signatures in self.proc_signatures.items()
|
|
},
|
|
{path: dict(docs) for path, docs in self.proc_docs.items()},
|
|
)
|
|
|
|
def diagnostic_snapshot(self, uri: str):
|
|
with self._index_lock:
|
|
return self.diagnostics.get(uri)
|
|
|
|
def navigation_snapshot(self) -> dict[str, FileSymbolIndex]:
|
|
"""Return an immutable snapshot of the workspace symbol indexes."""
|
|
with self._index_lock:
|
|
return dict(self.navigation_indexes)
|
|
|
|
def _begin_index_update(self, filepath: str, version: int | None) -> int | None:
|
|
with self._index_lock:
|
|
indexed_version = self._index_versions.get(filepath)
|
|
if version is None and indexed_version is not None:
|
|
return None
|
|
if (
|
|
version is not None
|
|
and indexed_version is not None
|
|
and version < indexed_version
|
|
):
|
|
return None
|
|
|
|
self._next_index_token += 1
|
|
token = self._next_index_token
|
|
self._index_tokens[filepath] = token
|
|
self._index_versions[filepath] = version
|
|
return token
|
|
|
|
def _discard_index_update(self, filepath: str, token: int) -> None:
|
|
with self._index_lock:
|
|
if self._index_tokens.get(filepath) != token:
|
|
return
|
|
self.poco_completion.pop(filepath, None)
|
|
self.proc_signatures.pop(filepath, None)
|
|
self.proc_docs.pop(filepath, None)
|
|
self.class_indexes.pop(filepath, None)
|
|
self.navigation_indexes.pop(filepath, None)
|
|
self.variable_indexes.pop(filepath, None)
|
|
self._committed_index_versions.pop(filepath, None)
|
|
self._invalidate_workspace_caches_locked()
|
|
|
|
def indexed_paths_under_uri(self, uri: str) -> list[pathlib.Path]:
|
|
target = pathlib.Path(uris.to_fs_path(uri))
|
|
with self._index_lock:
|
|
indexed_paths = set(self.poco_completion)
|
|
indexed_paths.update(self.proc_signatures)
|
|
indexed_paths.update(self.proc_docs)
|
|
indexed_paths.update(self.navigation_indexes)
|
|
indexed_paths.update(self.variable_indexes)
|
|
indexed_paths.update(self._index_tokens)
|
|
return [
|
|
pathlib.Path(path)
|
|
for path in indexed_paths
|
|
if self._is_same_or_child(path, target)
|
|
]
|
|
|
|
def remove_file_state(self, uri: str) -> None:
|
|
"""Remove cached and indexed state for a file or a complete folder."""
|
|
self.cancel_analysis_under_uri(uri)
|
|
target = pathlib.Path(uris.to_fs_path(uri))
|
|
|
|
with self._index_lock:
|
|
index_changed = False
|
|
for store in (
|
|
self.poco_completion,
|
|
self.proc_signatures,
|
|
self.proc_docs,
|
|
self.class_indexes,
|
|
self.navigation_indexes,
|
|
self.variable_indexes,
|
|
self._index_tokens,
|
|
self._index_versions,
|
|
self._committed_index_versions,
|
|
):
|
|
for path in list(store):
|
|
if self._is_same_or_child(path, target):
|
|
del store[path]
|
|
index_changed = True
|
|
|
|
if index_changed:
|
|
self._invalidate_workspace_caches_locked()
|
|
|
|
diagnostic_uris = set(self.diagnostics)
|
|
diagnostic_uris.update(self._diagnostic_tokens)
|
|
for diagnostic_uri in diagnostic_uris:
|
|
try:
|
|
diagnostic_path = pathlib.Path(uris.to_fs_path(diagnostic_uri))
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if self._is_same_or_child(diagnostic_path, target):
|
|
self.diagnostics.pop(diagnostic_uri, None)
|
|
self._diagnostic_tokens.pop(diagnostic_uri, None)
|
|
|
|
with self._parser_lock:
|
|
cached_keys = set(self._ast_cache)
|
|
cached_keys.update(self._line_cache)
|
|
for key in cached_keys:
|
|
try:
|
|
cached_path = pathlib.Path(uris.to_fs_path(key[0]))
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if self._is_same_or_child(cached_path, target):
|
|
self._ast_cache.pop(key, None)
|
|
self._line_cache.pop(key, None)
|
|
|
|
def update_poco_completion_for_file(
|
|
self,
|
|
document: TextDocument,
|
|
*,
|
|
cache_tree: bool = True,
|
|
require_file_exists: bool = False,
|
|
):
|
|
"""Update poco_completion for a specific file when it changes"""
|
|
filepath = str(pathlib.Path(uris.to_fs_path(document.uri)))
|
|
token = self._begin_index_update(filepath, document.version)
|
|
if token is None:
|
|
return False
|
|
|
|
collector = CompletionCollector()
|
|
try:
|
|
tree = (
|
|
self.get_tree(document)
|
|
if cache_tree
|
|
else self.parse_source(document.source)
|
|
)
|
|
tree.accept(collector, recurse=True)
|
|
collector.custom_functions.extend(class_completion_items(tree))
|
|
classes = indexed_classes(tree, document.uri, document.source)
|
|
docs = build_proc_docs(tree, document.source)
|
|
navigation_index = build_file_symbol_index(
|
|
filepath, document.uri, tree
|
|
)
|
|
variable_index = build_variable_index(document.source, tree)
|
|
except Exception as e:
|
|
LOGGER.debug("Error parsing %s: %s", filepath, e)
|
|
self._discard_index_update(filepath, token)
|
|
return False
|
|
|
|
if require_file_exists and not pathlib.Path(filepath).is_file():
|
|
self._discard_index_update(filepath, token)
|
|
return False
|
|
|
|
with self._index_lock:
|
|
if self._index_tokens.get(filepath) != token:
|
|
return False
|
|
self.poco_completion[filepath] = list(collector.custom_functions)
|
|
self.proc_signatures[filepath] = dict(collector.proc_signatures)
|
|
self.proc_docs[filepath] = docs
|
|
self.class_indexes[filepath] = classes
|
|
self.navigation_indexes[filepath] = navigation_index
|
|
self.variable_indexes[filepath] = (document.version, variable_index)
|
|
self._committed_index_versions[filepath] = document.version
|
|
self._invalidate_workspace_caches_locked()
|
|
return True
|
|
|
|
def format(
|
|
self,
|
|
document: TextDocument,
|
|
options: lsp.FormattingOptions,
|
|
range: Optional[Tuple[int, int]] = None,
|
|
):
|
|
# parser = Parser(command_plugins=["nx_plugins.poco_plugin.py"])
|
|
# parser._commands.update(commands)
|
|
|
|
indent = "\t" if not options.insert_spaces else " " * options.tab_size
|
|
formatter = Formatter(
|
|
FormatterOpts(
|
|
indent=indent,
|
|
spaces_in_braces=False,
|
|
balanced_spaces_in_braces=False,
|
|
max_blank_lines=500,
|
|
indent_namespace_eval=True,
|
|
indent_mixed_tab_size=0,
|
|
emacs=False,
|
|
debug_whitespace=False,
|
|
),
|
|
)
|
|
|
|
with self._parser_lock:
|
|
if range is not None:
|
|
start, end = range
|
|
return formatter.format_partial(document.source[start:end], self.parser)
|
|
|
|
return formatter.format_top(document.source, self.parser)
|
|
|
|
def linter(
|
|
self,
|
|
document: TextDocument,
|
|
) -> List[Violation]:
|
|
tree, cached_violations = self.get_tree_and_violations(document)
|
|
violations = list(cached_violations)
|
|
for checker in checks.get_checkers():
|
|
violations += checker.check(document.source, tree)
|
|
return violations
|
|
|
|
def lint(self, document: TextDocument):
|
|
diagnostics = []
|
|
|
|
try:
|
|
violations = self.linter(document)
|
|
except TclSyntaxError as e:
|
|
return [
|
|
lsp.Diagnostic(
|
|
message=str(e),
|
|
severity=lsp.DiagnosticSeverity.Error,
|
|
range=lsp.Range(
|
|
start=lsp.Position(e.start[0] - 1, e.start[1] - 1),
|
|
end=lsp.Position(e.end[0] - 1, e.end[1] - 1),
|
|
),
|
|
code="syntax error",
|
|
source=DIAGNOSTIC_SOURCE,
|
|
)
|
|
]
|
|
|
|
for violation in violations:
|
|
message = violation.message
|
|
severity = lsp.DiagnosticSeverity.Warning
|
|
start = lsp.Position(
|
|
line=violation.start[0] - 1, character=violation.start[1] - 1
|
|
)
|
|
end = lsp.Position(
|
|
line=violation.end[0] - 1, character=violation.end[1] - 1
|
|
)
|
|
|
|
diagnostics.append(
|
|
lsp.Diagnostic(
|
|
message=message,
|
|
severity=severity,
|
|
range=lsp.Range(
|
|
start=start,
|
|
end=end,
|
|
),
|
|
code=violation.id,
|
|
source=DIAGNOSTIC_SOURCE,
|
|
)
|
|
)
|
|
|
|
return diagnostics
|
|
|
|
def _compute_diagnostics(self, document: TextDocument) -> List[lsp.Diagnostic]:
|
|
return self.lint(document)
|
|
|
|
def compute_diagnostics(self, document: TextDocument):
|
|
with self._index_lock:
|
|
self._next_diagnostic_token += 1
|
|
token = self._next_diagnostic_token
|
|
self._diagnostic_tokens[document.uri] = token
|
|
|
|
diagnostics = self._compute_diagnostics(document)
|
|
|
|
with self._index_lock:
|
|
if self._diagnostic_tokens.get(document.uri) != token:
|
|
return
|
|
|
|
current = self.diagnostics.get(document.uri)
|
|
if current is not None:
|
|
current_version, _ = current
|
|
if (
|
|
current_version is not None
|
|
and document.version is not None
|
|
and current_version > document.version
|
|
):
|
|
return
|
|
|
|
# Keep the result id in sync even when only the document version changed.
|
|
if current != (document.version, diagnostics):
|
|
self.diagnostics[document.uri] = (document.version, diagnostics)
|