import logging import os import pathlib import threading import xml.etree.ElementTree as ET from dataclasses import dataclass 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, incremental_parse, parser from tools.completion_items import CompletionCollector from tools.tcloo_symbols import class_completion_items from tools.tcloo_completion import indexed_classes from tools.def_symbols import ADDRESS, BLOCK_TEMPLATE, DefDocument, parse_def_document, read_def_source from tools.file_sourcing import get_all_psc_files, psc_defined_event_files, psc_script_files from tools.formatter import NxFormatter as Formatter from tools.index_cache import FileStat, IndexCache, file_stat from tools.inlay_hint import InlayHintSignature, build_custom_inlay_signatures from tools.navigation import ( FileSymbolIndex, SymbolIdentity, build_file_symbol_index, definition_identities, ) 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__) @dataclass(frozen=True) class _FileIndex: """Everything indexed for one file; also the persistent cache entry.""" completion_items: list[lsp.CompletionItem] proc_signatures: dict[str, list[str]] proc_docs: dict[str, str] classes: dict navigation_index: FileSymbolIndex variable_index: tuple[set[str], dict[str, set[str]], list[ProcRange]] 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() # .def path -> parsed declarations, in PSC DefinedEvents order. self.def_documents: dict[str, DefDocument] = {} 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._thread_parsers = threading.local() # uri -> (normalized source, tree, violations) of the last successful # parse; survives version changes so edits can be reparsed partially. self._last_parse: dict[str, tuple[str, object, list]] = {} self.index_cache = IndexCache() 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._definition_identities_cache: tuple[ int, frozenset[SymbolIdentity] ] = (-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, pos=None): self.parser.violations = [] tree = self.parser.parse(source, pos=pos) return tree, list(self.parser.violations) def _parse_document(self, document: TextDocument): """Parse a document version, reusing unchanged parts of the last one. Callers hold the parser lock. """ source = incremental_parse.normalize_newlines(document.source) previous = self._last_parse.get(document.uri) result = None if previous is not None: try: result = incremental_parse.reparse( *previous, source, self._parse_source ) except TclSyntaxError: # E.g. a quote opened in the edit closes further down. result = None if result is None: result = self._parse_source(source) self._last_parse[document.uri] = (source, *result) return result def parse_source(self, source: str): """Parse without retaining an AST, on a parser owned by this thread. Background indexing must not hold the shared parser lock for whole files while request handlers wait for their document's tree. """ local_parser = getattr(self._thread_parsers, "parser", None) if local_parser is None: # Plugin commands live in tclint's shared registry, see __init__. local_parser = self._thread_parsers.parser = parser.CustomParser() local_parser.violations = [] return local_parser.parse(source) 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_document(document) 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_document(document) 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._definition_identities_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_def_symbols(self, roots, report=LOGGER.warning): """Read the block templates and addresses of all .def files listed as PSC DefinedEvents.""" documents: dict[str, DefDocument] = {} for root in roots: for psc in get_all_psc_files(root): try: def_files = psc_defined_event_files(psc) except (OSError, ET.ParseError) as error: report(f"Could not read PSC {psc}: {error}") continue for def_file in def_files: if str(def_file) in documents: continue try: documents[str(def_file)] = parse_def_document(read_def_source(def_file)) except OSError as error: report(f"Could not read DEF file {def_file}: {error}") with self._index_lock: self.def_documents = documents def def_documents_snapshot(self, current_path=None, current_source: str | None = None) -> dict[str, DefDocument]: """Return the PSC .def documents; ``current_source`` replaces the file being edited.""" with self._index_lock: documents = dict(self.def_documents) if current_path is not None and current_source is not None: key = next((path for path in documents if self.paths_equal(path, current_path)), os.fspath(current_path)) documents[key] = parse_def_document(current_source) return documents def _def_symbol_items(self, kind: str, item_kind, description: str) -> list[lsp.CompletionItem]: return [ lsp.CompletionItem( label=name, kind=item_kind, detail=f"{description} ({pathlib.Path(path).name})", ) for path, document in self.def_documents_snapshot().items() for name in document.names(kind) ] def block_template_items(self) -> list[lsp.CompletionItem]: return self._def_symbol_items(BLOCK_TEMPLATE, lsp.CompletionItemKind.Struct, "Block template") def address_items(self) -> list[lsp.CompletionItem]: return self._def_symbol_items(ADDRESS, lsp.CompletionItemKind.Field, "Address") def refresh_psc_scripts(self, roots, report=LOGGER.warning): """Index PSC dependencies through the same pipeline as workspace procs.""" self.refresh_def_symbols(roots, report=report) 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: source_stat = None if document is None: source_stat = file_stat(path_string) 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, source_stat=source_stat): 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 navigation_state( self, ) -> tuple[dict[str, FileSymbolIndex], frozenset[SymbolIdentity]]: """Return indexes plus their definitions, cached by index generation.""" with self._index_lock: generation, definitions = self._definition_identities_cache if generation != self._index_generation: definitions = frozenset( definition_identities(self.navigation_indexes) ) self._definition_identities_cache = ( self._index_generation, definitions, ) return dict(self.navigation_indexes), definitions 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) for uri in list(self._last_parse): try: parsed_path = pathlib.Path(uris.to_fs_path(uri)) except (TypeError, ValueError): continue if self._is_same_or_child(parsed_path, target): del self._last_parse[uri] def _build_file_index( self, document: TextDocument, filepath: str, cache_tree: bool ) -> "_FileIndex": tree = ( self.get_tree(document) if cache_tree else self.parse_source(document.source) ) collector = CompletionCollector() tree.accept(collector, recurse=True) collector.custom_functions.extend(class_completion_items(tree)) return _FileIndex( completion_items=list(collector.custom_functions), proc_signatures=dict(collector.proc_signatures), proc_docs=build_proc_docs(tree, document.source), classes=indexed_classes(tree, document.uri, document.source), navigation_index=build_file_symbol_index(filepath, document.uri, tree), variable_index=build_variable_index(document.source, tree), ) def update_poco_completion_for_file( self, document: TextDocument, *, cache_tree: bool = True, require_file_exists: bool = False, from_disk: bool = False, source_stat: FileStat | None = None, ): """Update poco_completion for a specific file when it changes. `from_disk` marks documents that read their source from disk lazily; `source_stat` is the file's stat taken before a caller read it. Such results are served from and stored in the persistent index cache. """ filepath = str(pathlib.Path(uris.to_fs_path(document.uri))) token = self._begin_index_update(filepath, document.version) if token is None: return False stat = source_stat or (file_stat(filepath) if from_disk else None) index = self.index_cache.get(filepath, stat) if stat is not None else None if index is None: try: index = self._build_file_index(document, filepath, cache_tree) except Exception as e: LOGGER.debug("Error parsing %s: %s", filepath, e) self._discard_index_update(filepath, token) return False # Only cache results whose file did not change while being read. if stat is not None and file_stat(filepath) == stat: self.index_cache.put(filepath, stat, index) 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(index.completion_items) self.proc_signatures[filepath] = dict(index.proc_signatures) self.proc_docs[filepath] = index.proc_docs self.class_indexes[filepath] = index.classes self.navigation_indexes[filepath] = index.navigation_index self.variable_indexes[filepath] = (document.version, index.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)