# Copyright (c) Microsoft Corporation. All rights reserved. # Licensed under the MIT License. """Implementation of tool support over LSP.""" from __future__ import annotations import json import operator import os import pathlib import re import sys import threading import time from collections import ChainMap from functools import reduce from typing import Any, Optional # ********************************************************** # Update sys.path before importing any bundled libraries. # ********************************************************** def update_sys_path(path_to_add: str, strategy: str) -> None: """Add given path to `sys.path`.""" if path_to_add not in sys.path and os.path.isdir(path_to_add): if strategy == "useBundled": sys.path.insert(0, path_to_add) elif strategy == "fromEnvironment": sys.path.append(path_to_add) # Ensure that we can import LSP libraries, and other bundled libraries. update_sys_path( os.fspath(pathlib.Path(__file__).parent.parent / "libs"), os.getenv("LS_IMPORT_STRATEGY", "useBundled"), ) # ********************************************************** # Imports needed for the language server goes below this. # ********************************************************** # pylint: disable=wrong-import-position,import-error import lsprotocol.types as lsp from pygls import uris from pygls.workspace.text_document import TextDocument import lsp_jsonrpc as jsonrpc from common.load_data import standard_items from lsp_tclserver import TclLanguageServer from tools.completion_items import ( CompletionContext, array_element_completions, completion_context, ranked_completion_items, ) from tools.def_navigation import ( all_def_target_locations, def_declarations, def_hover_markdown, effective_def_locations, def_rename_edits, def_symbol_at, ) from tools.def_flow import derived_def_symbol from tools.def_symbols import parse_def_document from tools.folding_ranges import build_folding_ranges from tools.index_cache import IndexCache from tools.inlay_hint import ( InlayHintGenerator, build_builtin_inlay_signatures, ) from tools.navigation import ( DEF_SYMBOL_KINDS, SymbolIdentity, call_hierarchy_identity, call_hierarchy_items, document_highlights, incoming_call_hierarchy, matching_occurrences, outgoing_call_hierarchy, symbol_at_position, workspace_symbols, ) from tools.semantic_tokens import ( TOKEN_TYPE_INDEX, TOKEN_TYPES, TokenModifier, _Highlighter, ) from tools.signature_help import build_signature_help from tools.tcloo_arguments import method_signature_help from tclint.lexer import TclSyntaxError from tools.tcloo_completion import may_contain_classes, parse_completion_source, tcloo_completions from tools.tcloo_symbols import class_completion_items from tools.tcloo_navigation import tcloo_definition from tools.tcl_command_completion import ( TCL_COMMAND_ITEMS, TCL_COMMAND_NAMES, DynamicCompletionKind, line_prefix_at_position, path_completion_items, tcl_argument_completion, ) WORKSPACE_SETTINGS = {} GLOBAL_SETTINGS = {} # Extension storage folder of the workspace; without it nothing is persisted. INDEX_CACHE_PATH: dict[str, str | None] = {} MAX_WORKERS = 5 LSP_SERVER = TclLanguageServer(name="NX Postprocessor Support", version="0.0.1", max_workers=MAX_WORKERS) BUILTIN_PROC_NAMES = {item.label for item in standard_items.tcl_keyword_list + standard_items.nx_procs} | set(TCL_COMMAND_NAMES) BUILTIN_VARIABLE_NAMES = {item.label for item in standard_items.nx_variables} _TCL_COMMAND_ITEMS_BY_LABEL = {item.label: item for item in TCL_COMMAND_ITEMS} _TCL_KEYWORD_ITEMS = [_TCL_COMMAND_ITEMS_BY_LABEL.get(item.label, item) for item in standard_items.tcl_keyword_list] _STATIC_TCL_LABELS = {item.label for item in standard_items.tcl_keyword_list} STATIC_COMPLETION_ITEMS = tuple(_TCL_KEYWORD_ITEMS + [item for item in TCL_COMMAND_ITEMS if item.label not in _STATIC_TCL_LABELS] + standard_items.nx_procs + standard_items.nx_variables) STATIC_VARIABLE_ITEMS = {item.label: item for item in standard_items.nx_variables} BUILTIN_INLAY_SIGNATURES = build_builtin_inlay_signatures(standard_items.json_data.get("MOM_procs", [])) BUILTIN_HOVER_ITEMS = {} for _hover_item in standard_items.json_data.get("MOM_procs", []) + standard_items.json_data.get("mom_variables", []): BUILTIN_HOVER_ITEMS.setdefault(_hover_item.get("label"), _hover_item) # ********************************************************** # Tool specific code goes below this. # ********************************************************** # Delete "Linting features" section if your tool is NOT a linter. # ********************************************************** # Linting features start here # ********************************************************** # See `pylint` implementation for a full featured linter extension: # Pylint: https://github.com/microsoft/vscode-pylint/blob/main/bundled/tool @LSP_SERVER.feature(lsp.TEXT_DOCUMENT_DID_OPEN) def did_open(params: lsp.DidOpenTextDocumentParams) -> None: """LSP handler for textDocument/didOpen request.""" document = LSP_SERVER.workspace.get_text_document(params.text_document.uri) LSP_SERVER.clear_cache_for_uri(document.uri) LSP_SERVER.analyze_document_now(document) @LSP_SERVER.feature(lsp.TEXT_DOCUMENT_DID_SAVE) def did_save(params: lsp.DidSaveTextDocumentParams) -> None: """LSP handler for textDocument/didSave request.""" document = LSP_SERVER.workspace.get_text_document(params.text_document.uri) LSP_SERVER.analyze_document_now(document) @LSP_SERVER.feature(lsp.TEXT_DOCUMENT_DID_CLOSE) def did_close(params: lsp.DidCloseTextDocumentParams) -> None: """LSP handler for textDocument/didClose request.""" uri = params.text_document.uri LSP_SERVER.remove_file_state(uri) _index_tcl_file_from_disk(uri) @LSP_SERVER.feature(lsp.TEXT_DOCUMENT_DID_CHANGE) def did_change(params: lsp.DidChangeTextDocumentParams) -> None: """LSP handler for textDocument/didChange request""" document = LSP_SERVER.workspace.get_text_document(params.text_document.uri) LSP_SERVER.clear_cache_for_uri(document.uri) LSP_SERVER.schedule_document_analysis(document) FILE_OPERATION_OPTIONS = lsp.FileOperationRegistrationOptions( filters=[ lsp.FileOperationFilter( scheme="file", pattern=lsp.FileOperationPattern(glob="**/*"), ) ] ) def _index_tcl_file_from_disk(uri: str) -> None: if not uri.startswith("file:"): return path = pathlib.Path(uris.to_fs_path(uri)) if path.suffix.lower() != ".tcl" or not path.is_file(): return try: document = TextDocument(uri=uri, language_id="tcl") LSP_SERVER.update_poco_completion_for_file( document, cache_tree=False, require_file_exists=True, from_disk=True, ) except (OSError, UnicodeError) as error: log_warning(f"Could not re-index {path}: {error}") @LSP_SERVER.feature(lsp.WORKSPACE_DID_DELETE_FILES, FILE_OPERATION_OPTIONS) def did_delete_files(params: lsp.DeleteFilesParams) -> None: for deleted_file in params.files: LSP_SERVER.remove_file_state(deleted_file.uri) @LSP_SERVER.feature(lsp.WORKSPACE_DID_RENAME_FILES, FILE_OPERATION_OPTIONS) def did_rename_files(params: lsp.RenameFilesParams) -> None: for renamed_file in params.files: old_path = pathlib.Path(uris.to_fs_path(renamed_file.old_uri)) new_path = pathlib.Path(uris.to_fs_path(renamed_file.new_uri)) indexed_paths = LSP_SERVER.indexed_paths_under_uri(renamed_file.old_uri) new_index_paths: set[pathlib.Path] = set() for indexed_path in indexed_paths: if LSP_SERVER.paths_equal(indexed_path, old_path): new_index_paths.add(new_path) else: relative_path = os.path.relpath(indexed_path, old_path) new_index_paths.add(new_path / relative_path) # Also handles renaming a previously unindexed file to a TCL file. if new_path.suffix.lower() == ".tcl": new_index_paths.add(new_path) LSP_SERVER.remove_file_state(renamed_file.old_uri) for new_index_path in new_index_paths: _index_tcl_file_from_disk(new_index_path.as_uri()) @LSP_SERVER.feature(lsp.WORKSPACE_DID_CHANGE_WATCHED_FILES) def did_change_watched_files(params: lsp.DidChangeWatchedFilesParams) -> None: """Keep indexes for closed Tcl files synchronized with disk changes.""" for change in params.changes: suffix = pathlib.Path(uris.to_fs_path(change.uri)).suffix.lower() if suffix == ".psc": _refresh_psc_index() continue if suffix == ".def": LSP_SERVER.refresh_def_symbols(_workspace_roots(), report=log_warning) continue if change.type == lsp.FileChangeType.Deleted: LSP_SERVER.remove_file_state(change.uri) else: _index_tcl_file_from_disk(change.uri) def _workspace_roots() -> list[pathlib.Path]: folders = LSP_SERVER.workspace.folders roots = [pathlib.Path(uris.to_fs_path(uri)) for uri in folders] if not roots and LSP_SERVER.workspace.root_path: roots = [pathlib.Path(LSP_SERVER.workspace.root_path)] return roots def _refresh_psc_index(): LSP_SERVER.refresh_psc_scripts(_workspace_roots(), report=log_warning) @LSP_SERVER.feature( lsp.TEXT_DOCUMENT_DIAGNOSTIC, lsp.DiagnosticOptions( identifier="pull-diagnostics", inter_file_dependencies=False, workspace_diagnostics=False, ), ) def document_diagnostic(params: lsp.DocumentDiagnosticParams): """Return diagnostics for the requested document""" uri = params.text_document.uri doc = LSP_SERVER.workspace.get_text_document(uri) diagnostic_state = LSP_SERVER.diagnostic_snapshot(uri) was_cached = diagnostic_state is not None and diagnostic_state[0] == doc.version if not was_cached: LSP_SERVER.compute_diagnostics(doc) diagnostic_state = LSP_SERVER.diagnostic_snapshot(uri) if diagnostic_state is None: return lsp.FullDocumentDiagnosticReport(items=[]) version, diagnostics = diagnostic_state result_id = f"{uri}@{version}" if was_cached and result_id == params.previous_result_id: return lsp.UnchangedDocumentDiagnosticReport(result_id) return lsp.FullDocumentDiagnosticReport(items=diagnostics, result_id=result_id) @LSP_SERVER.feature( lsp.TEXT_DOCUMENT_COMPLETION, lsp.CompletionOptions(trigger_characters=["$", " ", "-", "(", ","], resolve_provider=True), ) def on_completion(params: lsp.CompletionParams) -> lsp.CompletionList: result = _on_completion(params) doc = LSP_SERVER.workspace.get_text_document(params.text_document.uri) symbol_list = _symbol_list_completion(LSP_SERVER.get_lines(doc), params.position) if symbol_list is not None and symbol_list.is_incomplete: # Re-request while typing so the full list appears at BLOCK_LIST/ADDR_LIST. result.items = [*symbol_list.items, *result.items] result.is_incomplete = True return result @LSP_SERVER.feature(lsp.COMPLETION_ITEM_RESOLVE) def on_completion_resolve(item: lsp.CompletionItem) -> lsp.CompletionItem: """Show the .def declaration of a selected block template/address, as on hover.""" target = item.data.get("def") if isinstance(item.data, dict) else None if item.documentation is None and isinstance(target, list) and len(target) == 2: markdown = def_hover_markdown(LSP_SERVER.def_documents_snapshot(), tuple(target)) if markdown: item.documentation = lsp.MarkupContent(kind=lsp.MarkupKind.Markdown, value=markdown) return item # Keywords that expand to all loaded .def names: keyword -> (items, description). SYMBOL_LIST_KEYWORDS = { "BLOCK_LIST": (lambda: LSP_SERVER.block_template_items(), "block templates"), "ADDR_LIST": (lambda: LSP_SERVER.address_items(), "addresses"), } DEF_NAME_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*") _WORD_BEFORE_CURSOR_RE = re.compile(r"(? lsp.CompletionList | None: """Complete BLOCK_LIST/ADDR_LIST to all loaded block templates/addresses. While the typed word is a prefix of a keyword, an incomplete list with the keyword itself is returned. Once the keyword is typed, the complete list of quoted names replaces it. """ line_prefix = line_prefix_at_position(source_lines, position) match = _WORD_BEFORE_CURSOR_RE.search(line_prefix or "") if match is None: return None word = match.group() for keyword, (symbol_items, _) in SYMBOL_LIST_KEYWORDS.items(): if not word.startswith(keyword): continue items = [] ranked = ranked_completion_items(((0, item) for item in symbol_items()), CompletionContext.GENERAL) for item in ranked: item = _quoted_item(item, source_lines, position, word) # Keep every name visible for the typed keyword; text after it narrows. item.filter_text = f"{keyword}{item.label}" items.append(item) return lsp.CompletionList(is_incomplete=False, items=items) keywords = [ lsp.CompletionItem( label=keyword, kind=lsp.CompletionItemKind.Snippet, detail=f"Show all loaded {description}", insert_text=keyword, sort_text=f"000:{keyword.casefold()}", command=lsp.Command(title=f"Show {description}", command="editor.action.triggerSuggest"), ) for keyword, (_, description) in SYMBOL_LIST_KEYWORDS.items() if keyword.startswith(word) ] if keywords: return lsp.CompletionList(is_incomplete=True, items=keywords) return None def _on_completion(params: lsp.CompletionParams) -> lsp.CompletionList: doc = LSP_SERVER.workspace.get_text_document(params.text_document.uri) position = params.position source_lines = LSP_SERVER.get_lines(doc) def current_tree(): try: return LSP_SERVER.get_tree(doc) except TclSyntaxError: return None oo_items = tcloo_completions(source_lines, position, LSP_SERVER.class_snapshot(doc.path), current_tree) if oo_items is not None: return lsp.CompletionList(is_incomplete=False, items=oo_items) array_items = array_element_completions( source_lines, position, LSP_SERVER.navigation_snapshot().values(), str(pathlib.Path(uris.to_fs_path(doc.uri))), ) if array_items is not None: return lsp.CompletionList(is_incomplete=False, items=array_items) context = completion_context(source_lines, position) symbol_list = _symbol_list_completion(source_lines, position) if symbol_list is not None and symbol_list.is_incomplete is False: return symbol_list # Variable completion wins inside command arguments. Otherwise prefer the # narrow command grammar when the cursor is at a known subcommand/option. argument_completion = None if context != CompletionContext.VARIABLE: argument_completion = tcl_argument_completion(source_lines, position) if argument_completion is not None and argument_completion.dynamic_kind is None: items = ranked_completion_items( ((0, item) for item in argument_completion.items), CompletionContext.GENERAL, ) return lsp.CompletionList(is_incomplete=False, items=items) if argument_completion is not None and argument_completion.dynamic_kind == DynamicCompletionKind.PATH: dynamic_items: tuple[lsp.CompletionItem, ...] = () if doc.uri.startswith("file:"): document_path = pathlib.Path(uris.to_fs_path(doc.uri)) dynamic_items = path_completion_items( document_path.parent, argument_completion, position, ) path_candidates = [(0, item) for item in argument_completion.items] path_candidates.extend((10, item) for item in dynamic_items) items = ranked_completion_items( path_candidates, CompletionContext.GENERAL, ) return lsp.CompletionList(is_incomplete=False, items=items) # Space and dash are registered only to open command-aware suggestions. # Do not display the broad fallback list when such a trigger has no match. if ( argument_completion is None and params.context is not None and params.context.trigger_kind == lsp.CompletionTriggerKind.TriggerCharacter and params.context.trigger_character in {" ", "-", "(", ","} ): return lsp.CompletionList(is_incomplete=False, items=[]) try: tree = LSP_SERVER.get_tree(doc) except TclSyntaxError: tree = parse_completion_source(doc.source) if tree is None: return lsp.CompletionList(is_incomplete=False, items=[]) globals_set, procs_locals, proc_ranges = LSP_SERVER.variable_index_for_document(doc, tree) local_names: set[str] = set() for proc_range in proc_ranges: end_line = proc_range.end_line or proc_range.start_line if proc_range.start_line <= position.line <= end_line: local_names.update(procs_locals.get(proc_range.name, set())) break candidates: list[tuple[int, lsp.CompletionItem]] = [] candidates.extend((0, item) for item in class_completion_items(tree)) for name in sorted(local_names - globals_set): candidates.append(( 0, lsp.CompletionItem( label=name, kind=lsp.CompletionItemKind.Variable, detail="Local variable", ), )) for name in sorted(globals_set): item = STATIC_VARIABLE_ITEMS.get(name) if item is None: item = lsp.CompletionItem( label=name, kind=lsp.CompletionItemKind.Variable, detail="Workspace variable", ) candidates.append(( 100, item, )) filepath = str(pathlib.Path(uris.to_fs_path(doc.uri))) items_by_file = LSP_SERVER.completion_items_by_file_snapshot() for item_path in sorted(items_by_file, key=lambda path: path.casefold()): priority = 100 if LSP_SERVER.paths_equal(item_path, filepath) else 200 candidates.extend((priority, item) for item in items_by_file[item_path]) if argument_completion is not None: static_candidates = [(0, item) for item in argument_completion.items] if argument_completion.dynamic_kind == DynamicCompletionKind.VARIABLE: variable_candidates = list(candidates) variable_candidates.extend((300, item) for item in standard_items.nx_variables) items = ranked_completion_items( variable_candidates, CompletionContext.VARIABLE, ) # Command options are valid alongside variables, even though they # are keywords and must not pass through the variable-only filter. items = [*ranked_completion_items(static_candidates, CompletionContext.GENERAL), *items] return lsp.CompletionList(is_incomplete=False, items=items) def_symbol_items = { DynamicCompletionKind.BLOCK_TEMPLATE: LSP_SERVER.block_template_items, DynamicCompletionKind.ADDRESS: LSP_SERVER.address_items, DynamicCompletionKind.VALUE: list, }.get(argument_completion.dynamic_kind) if def_symbol_items is not None: # MOM_do_template, MOM_force, ... take a .def name, a fixed value (e.g. # Always|Once|Off) or a variable holding one. def_symbols = [*static_candidates, *((0, item) for item in def_symbol_items())] variable_candidates = list(candidates) variable_candidates.extend((300, item) for item in standard_items.nx_variables) variables = [] for item in ranked_completion_items(variable_candidates, CompletionContext.VARIABLE): # Without a typed "$" the variable must be substituted. item.insert_text = f"${item.label}" item.filter_text = item.label item.text_edit = None variables.append(item) quoted = [ _quoted_item(item, source_lines, position, argument_completion.active_prefix) for item in ranked_completion_items(def_symbols, CompletionContext.GENERAL) ] return lsp.CompletionList(is_incomplete=False, items=[*quoted, *variables]) if argument_completion.dynamic_kind == DynamicCompletionKind.PROCEDURE: procedure_kinds = { lsp.CompletionItemKind.Constructor, lsp.CompletionItemKind.Function, lsp.CompletionItemKind.Method, } procedure_candidates = [(priority, item) for priority, item in candidates if item.kind in procedure_kinds] procedure_candidates.extend((300, item) for item in standard_items.nx_procs) items = ranked_completion_items( [*static_candidates, *procedure_candidates], CompletionContext.GENERAL, ) return lsp.CompletionList(is_incomplete=False, items=items) if argument_completion.dynamic_kind == DynamicCompletionKind.NAMESPACE: namespace_candidates = list(static_candidates) prefix_is_absolute = argument_completion.active_prefix.startswith("::") for index in LSP_SERVER.navigation_snapshot().values(): priority = 100 if LSP_SERVER.paths_equal(index.path, filepath) else 200 for occurrence in index.occurrences: if occurrence.identity.kind != "namespace": continue name = occurrence.identity.name label = name if prefix_is_absolute else name.removeprefix("::") namespace_candidates.append(( priority, lsp.CompletionItem( label=label, kind=lsp.CompletionItemKind.Module, detail="Tcl namespace", ), )) items = ranked_completion_items( namespace_candidates, CompletionContext.GENERAL, ) return lsp.CompletionList(is_incomplete=False, items=items) candidates.extend((300, item) for item in STATIC_COMPLETION_ITEMS) items = ranked_completion_items(candidates, context) return lsp.CompletionList(is_incomplete=False, items=items) def _quoted_item( item: lsp.CompletionItem, source_lines, position: lsp.Position, active_prefix: str, ) -> lsp.CompletionItem: """Insert ``item`` as a quoted word, replacing quotes the user already typed.""" line = source_lines[position.line] if position.line < len(source_lines) else "" utf16 = line.encode("utf-16-le") start = position.character - len(active_prefix.encode("utf-16-le")) // 2 end = position.character opened = start > 0 and utf16[(start - 1) * 2 : start * 2].decode("utf-16-le") == '"' if opened: start -= 1 # Also replace a closing quote the editor inserted automatically. if utf16[end * 2 : (end + 1) * 2].decode("utf-16-le") == '"': end += 1 item.text_edit = lsp.TextEdit( range=lsp.Range( start=lsp.Position(line=position.line, character=start), end=lsp.Position(line=position.line, character=end), ), new_text=f'"{item.label}"', ) item.insert_text = None item.filter_text = f'"{item.label}' if opened else item.label return item @LSP_SERVER.feature( lsp.TEXT_DOCUMENT_SIGNATURE_HELP, lsp.SignatureHelpOptions( trigger_characters=[" "], retrigger_characters=[" "], ), ) def signature_help(params: lsp.SignatureHelpParams) -> lsp.SignatureHelp | None: document = LSP_SERVER.workspace.get_text_document(params.text_document.uri) try: tree = LSP_SERVER.get_tree(document) except TclSyntaxError: tree = parse_completion_source(document.source) if tree is None: return None method_help = method_signature_help(document.source, params.position, LSP_SERVER.class_snapshot(document.path), tree) if method_help is not None: return method_help custom_signatures, custom_docs = LSP_SERVER.proc_metadata_snapshot(document.path) return build_signature_help( document.source, tree, params.position, custom_signatures, custom_docs, standard_items.json_data.get("MOM_procs", []), ) # @LSP_SERVER.feature(lsp.TEXT_DOCUMENT_DOCUMENT_SYMBOL) # def document_symbols(params: lsp.DocumentSymbolParams): # doc = LSP_SERVER.workspace.get_text_document(params.text_document.uri) # ast = LSP_SERVER.parser.parse(doc.source) # symbols = LSP_SERVER.extract_tcl_symbols(ast) # return symbols @LSP_SERVER.feature(lsp.TEXT_DOCUMENT_DOCUMENT_SYMBOL) def document_symbols(params: lsp.DocumentSymbolParams): from tools.document_symbols import build_document_symbols doc = LSP_SERVER.workspace.get_text_document(params.text_document.uri) return build_document_symbols(doc.source) @LSP_SERVER.feature(lsp.TEXT_DOCUMENT_INLAY_HINT) def inlay_hints(params: lsp.InlayHintParams): document = LSP_SERVER.workspace.get_text_document(params.text_document.uri) settings = _get_settings_by_document(document) if not settings.get("inlayHint", False): return [] inlay_settings = settings.get("inlayHints", {}) parameter_names = inlay_settings.get("parameterNames", "all") if parameter_names == "none": return [] # Reuse cached AST tree = LSP_SERVER.get_tree(document) # Built-in NX procedures are the fallback. Workspace procedures replace them, # and a declaration in the current file wins over duplicate workspace names. custom_signatures = LSP_SERVER.custom_inlay_signatures_snapshot(document.path) signatures = ChainMap(custom_signatures, BUILTIN_INLAY_SIGNATURES) generator = InlayHintGenerator( document.source, signatures, external_classes=LSP_SERVER.class_snapshot(document.path), source_lines=LSP_SERVER.get_lines(document), requested_range=params.range, parameter_names=parameter_names, suppress_when_argument_matches_name=inlay_settings.get("suppressWhenArgumentMatchesName", True), ) return generator.generate(tree) @LSP_SERVER.feature( lsp.TEXT_DOCUMENT_SEMANTIC_TOKENS_FULL, lsp.SemanticTokensLegend( token_types=TOKEN_TYPES, token_modifiers=[m.name for m in TokenModifier], ), ) def semantic_tokens(params: lsp.SemanticTokensParams): document = LSP_SERVER.workspace.get_text_document(params.text_document.uri) data = [] plugins = [] hl = _Highlighter(plugins, LSP_SERVER.custom_function_names_snapshot()) # Reuse cached AST tree = LSP_SERVER.get_tree(document) classes = LSP_SERVER.class_snapshot(document.path) if may_contain_classes(document.source, classes): hl.highlight_classes(tree, classes) hl.highlight_methods(tree, document.source, document.uri, classes) tree.accept(hl, recurse=True) tokens = hl.tokens() for token in tokens: data.extend([ token.line, token.offset, token.length, TOKEN_TYPE_INDEX[token.tok_type], reduce(operator.or_, token.tok_modifiers, 0), ]) return lsp.SemanticTokens(data=data) @LSP_SERVER.feature(lsp.TEXT_DOCUMENT_FOLDING_RANGE) def folding_ranges(params: lsp.FoldingRangeParams): document = LSP_SERVER.workspace.get_text_document(params.text_document.uri) tree = LSP_SERVER.get_tree(document) return build_folding_ranges(tree) @LSP_SERVER.feature(lsp.TEXT_DOCUMENT_HOVER) def hover(params: lsp.HoverParams) -> lsp.Hover: pos = params.position document_uri = params.text_document.uri document = LSP_SERVER.workspace.get_text_document(document_uri) col = params.position.character try: line = LSP_SERVER.get_lines(document)[pos.line] except IndexError: return None # Do not show hover for proc name in its declaration from tools.proc_docs import is_proc_declaration_line if is_proc_declaration_line(line, pos.character): return None # Identify the token under the cursor for m in re.finditer(r"\b\w+\b", line): if m.start() <= col <= m.end(): token = m.group(0) break else: return None def_hover = _tcl_def_hover(document_uri, pos, token) if def_hover is not None: return def_hover # 1) If token is a known MOM proc/variable, return built-in hover match = BUILTIN_HOVER_ITEMS.get(token) if match and match.get("kind") == "function": label = match.get("label", "") parameters = match.get("parameters", []) param_lines = "\n".join(f"- `{p['name']}`: {p['desc']}" for p in parameters) or "_None_" example_data = match.get("example", []) example_md = "\n".join(f"{line}" for line in example_data) returns_data = match.get("returns", ["None"]) returns_md = "\n".join(f"- {line}" for line in returns_data) doc_md = f"""\ ### 📘 {label} **Purpose** {match.get("description", "No description available.")} **Format** `{match.get("format", label)}` **Parameters** {param_lines} **Return value** {returns_md} **Example** ```tcl {example_md}""" return lsp.Hover(lsp.MarkupContent(kind=lsp.MarkupKind.Markdown, value=doc_md)) # 2) Otherwise, check if the token is a custom proc and show its preceding doc block # Build a merged map of proc -> docs gathered during initialization and updates proc_doc = LSP_SERVER.proc_documentation(token, document.path) if proc_doc is not None: return lsp.Hover(lsp.MarkupContent(kind=lsp.MarkupKind.Markdown, value=proc_doc)) return None def _def_target(identity: SymbolIdentity) -> tuple[str, str] | None: return (identity.kind, identity.name) if identity.kind in DEF_SYMBOL_KINDS else None def _word_at(uri: str, position: lsp.Position) -> str | None: document = LSP_SERVER.workspace.get_text_document(uri) try: line = LSP_SERVER.get_lines(document)[position.line] except IndexError: return None for match in re.finditer(r"\b\w+\b", line): if match.start() <= position.character <= match.end(): return match.group(0) return None def _tcl_def_symbol(uri: str, position: lsp.Position, token: str | None): """The .def target at ``position``: a direct NX command argument or a derived name. Returns (documents, target, range) or None. """ documents = LSP_SERVER.def_documents_snapshot() # Cheap guard: only names declared in a .def file are analyzed at all. kinds = {kind for kind in DEF_SYMBOL_KINDS if token and any(token in document.names(kind) for document in documents.values())} if not kinds: return None context = _navigation_context(uri, position) target = context and _def_target(context[3]) if target: return documents, target, context[2].range try: tree = LSP_SERVER.get_tree(LSP_SERVER.workspace.get_text_document(uri)) except TclSyntaxError: return None derived = derived_def_symbol(tree, position, LSP_SERVER.def_wrapper_table()) if derived is None or derived[1] != token: return None found, name, range_ = derived kind = next((kind for kind in sorted(found & kinds) if def_declarations(documents, (kind, name))), None) return (documents, (kind, name), range_) if kind else None def _tcl_def_hover(uri: str, position: lsp.Position, token: str) -> lsp.Hover | None: symbol = _tcl_def_symbol(uri, position, token) if symbol is None: return None documents, target, range_ = symbol markdown = def_hover_markdown(documents, target) if not markdown: return None return lsp.Hover(lsp.MarkupContent(kind=lsp.MarkupKind.Markdown, value=markdown), range=range_) @LSP_SERVER.feature(lsp.TEXT_DOCUMENT_DEFINITION) def goto_definition(params: lsp.DefinitionParams): """Resolve TclOO declarations, then indexed proc and variable definitions.""" try: workspace = LSP_SERVER.workspace except RuntimeError: workspace = None if workspace is not None: document = workspace.get_text_document(params.text_document.uri) try: tree = LSP_SERVER.get_tree(document) except TclSyntaxError: tree = None # tcloo_definition repairs open delimiters itself. target = tcloo_definition(document.source, document.uri, params.position, LSP_SERVER.class_snapshot(document.path), tree) if target is not None: return [target] context = _navigation_context(params.text_document.uri, params.position) if context is None: symbol = _tcl_def_symbol(params.text_document.uri, params.position, _word_at(params.text_document.uri, params.position)) return symbol and effective_def_locations(symbol[0], symbol[1]) or None indexes, definitions, _, identity = context target = _def_target(identity) if target is not None: return effective_def_locations(LSP_SERVER.def_documents_snapshot(), target) or None locations = [lsp.Location(uri=index.uri, range=occurrence.range) for index, occurrence in matching_occurrences(identity, indexes, definitions) if occurrence.is_definition] return _sorted_locations(locations) or None def _navigation_context(uri: str, position: lsp.Position): indexes, definitions = LSP_SERVER.navigation_state() filepath = str(pathlib.Path(uris.to_fs_path(uri))) index = indexes.get(filepath) if index is None or LSP_SERVER.index_update_pending(filepath): document = LSP_SERVER.workspace.get_text_document(uri) LSP_SERVER.update_poco_completion_for_file(document) indexes, definitions = LSP_SERVER.navigation_state() index = indexes.get(filepath) if index is None: return None result = symbol_at_position(index, position, definitions) if result is None: return None occurrence, identity = result return indexes, definitions, occurrence, identity def _sorted_locations(locations: list[lsp.Location]) -> list[lsp.Location]: return sorted( locations, key=lambda location: ( location.uri, location.range.start.line, location.range.start.character, ), ) @LSP_SERVER.feature(lsp.TEXT_DOCUMENT_REFERENCES) def references(params: lsp.ReferenceParams) -> list[lsp.Location]: context = _navigation_context(params.text_document.uri, params.position) if context is None: symbol = _tcl_def_symbol(params.text_document.uri, params.position, _word_at(params.text_document.uri, params.position)) if symbol is None: return [] indexes, _ = LSP_SERVER.navigation_state() return all_def_target_locations(symbol[0], indexes, symbol[1], params.context.include_declaration) indexes, definitions, _, identity = context target = _def_target(identity) if target is not None: return all_def_target_locations( LSP_SERVER.def_documents_snapshot(), indexes, target, params.context.include_declaration ) locations = [ lsp.Location(uri=index.uri, range=occurrence.range) for index, occurrence in matching_occurrences(identity, indexes, definitions) if params.context.include_declaration or not occurrence.is_definition ] return _sorted_locations(locations) @LSP_SERVER.feature(lsp.TEXT_DOCUMENT_DOCUMENT_HIGHLIGHT) def document_highlight(params: lsp.DocumentHighlightParams): context = _navigation_context(params.text_document.uri, params.position) if context is None: return [] indexes, definitions, _, identity = context filepath = str(pathlib.Path(uris.to_fs_path(params.text_document.uri))) index = indexes.get(filepath) if index is None: return [] return document_highlights(index, identity, definitions) def _is_renamable( identity: SymbolIdentity, indexes, definitions: set[SymbolIdentity], ) -> bool: if identity not in definitions or identity.kind not in {"proc", "variable"}: return False basename = identity.name.rsplit("::", 1)[-1] if identity.kind == "proc": if basename in BUILTIN_PROC_NAMES: return False definition_count = sum(occurrence.is_definition and occurrence.identity == identity for index in indexes.values() for occurrence in index.occurrences) return definition_count == 1 return basename not in BUILTIN_VARIABLE_NAMES @LSP_SERVER.feature(lsp.TEXT_DOCUMENT_PREPARE_RENAME) def prepare_rename(params: lsp.PrepareRenameParams): context = _navigation_context(params.text_document.uri, params.position) if context is None: return None indexes, definitions, occurrence, identity = context target = _def_target(identity) if target is not None: if not def_declarations(LSP_SERVER.def_documents_snapshot(), target): return None elif not _is_renamable(identity, indexes, definitions): return None return lsp.PrepareRenamePlaceholder(range=occurrence.range, placeholder=occurrence.placeholder) @LSP_SERVER.feature( lsp.TEXT_DOCUMENT_RENAME, lsp.RenameOptions(prepare_provider=True), ) def rename(params: lsp.RenameParams) -> lsp.WorkspaceEdit | None: if not DEF_NAME_RE.fullmatch(params.new_name): return None context = _navigation_context(params.text_document.uri, params.position) if context is None: return None indexes, definitions, _, identity = context target = _def_target(identity) if target is not None: return def_rename_edits(LSP_SERVER.def_documents_snapshot(), indexes, target, params.new_name) if not _is_renamable(identity, indexes, definitions): return None changes: dict[str, list[lsp.TextEdit]] = {} seen = set() for index, occurrence in matching_occurrences(identity, indexes, definitions): key = ( index.uri, occurrence.range.start.line, occurrence.range.start.character, occurrence.range.end.line, occurrence.range.end.character, ) if key in seen: continue seen.add(key) changes.setdefault(index.uri, []).append(lsp.TextEdit(range=occurrence.range, new_text=params.new_name)) for edits in changes.values(): edits.sort( key=lambda edit: ( edit.range.start.line, edit.range.start.character, ), reverse=True, ) return lsp.WorkspaceEdit(changes=changes) # .def files are not synchronized with the server; the client sends their # current text with each request. DEF_REQUEST_DEFINITION = "nxPostSupport/def/definition" DEF_REQUEST_HOVER = "nxPostSupport/def/hover" DEF_REQUEST_REFERENCES = "nxPostSupport/def/references" DEF_REQUEST_PREPARE_RENAME = "nxPostSupport/def/prepareRename" DEF_REQUEST_RENAME = "nxPostSupport/def/rename" def _def_request_context(params): uri = params.textDocument.uri position = lsp.Position(line=params.position.line, character=params.position.character) path = uris.to_fs_path(uri) documents = LSP_SERVER.def_documents_snapshot(path, params.text) symbol = def_symbol_at(parse_def_document(params.text), position) if symbol is None: return None target, range_, _ = symbol current = next(key for key in documents if LSP_SERVER.paths_equal(key, path)) return documents, {current: uri}, target, range_ @LSP_SERVER.feature(DEF_REQUEST_DEFINITION) def def_definition(params) -> list[lsp.Location] | None: context = _def_request_context(params) if context is None: return None documents, def_uris, target, _ = context return effective_def_locations(documents, target, def_uris) or None @LSP_SERVER.feature(DEF_REQUEST_HOVER) def def_hover(params) -> lsp.Hover | None: context = _def_request_context(params) if context is None: return None documents, _, target, range_ = context markdown = def_hover_markdown(documents, target) if markdown is None: return None return lsp.Hover(lsp.MarkupContent(kind=lsp.MarkupKind.Markdown, value=markdown), range=range_) @LSP_SERVER.feature(DEF_REQUEST_REFERENCES) def def_references(params) -> list[lsp.Location]: context = _def_request_context(params) if context is None: return [] documents, def_uris, target, _ = context include_declaration = bool(getattr(params, "includeDeclaration", True)) indexes, _ = LSP_SERVER.navigation_state() return all_def_target_locations(documents, indexes, target, include_declaration, def_uris) @LSP_SERVER.feature(DEF_REQUEST_PREPARE_RENAME) def def_prepare_rename(params): context = _def_request_context(params) if context is None: return None documents, _, target, range_ = context if not def_declarations(documents, target): return None return lsp.PrepareRenamePlaceholder(range=range_, placeholder=target[1]) @LSP_SERVER.feature(DEF_REQUEST_RENAME) def def_rename(params) -> lsp.WorkspaceEdit | None: if not DEF_NAME_RE.fullmatch(params.newName): return None context = _def_request_context(params) if context is None: return None documents, def_uris, target, _ = context indexes, _ = LSP_SERVER.navigation_state() return def_rename_edits(documents, indexes, target, params.newName, def_uris) @LSP_SERVER.feature(lsp.WORKSPACE_SYMBOL) def workspace_symbol(params: lsp.WorkspaceSymbolParams): return workspace_symbols(LSP_SERVER.navigation_snapshot(), params.query) @LSP_SERVER.feature(lsp.TEXT_DOCUMENT_PREPARE_CALL_HIERARCHY) def prepare_call_hierarchy(params: lsp.CallHierarchyPrepareParams): context = _navigation_context(params.text_document.uri, params.position) if context is None: return None indexes, _, _, identity = context items = call_hierarchy_items(identity, indexes) return items or None @LSP_SERVER.feature(lsp.CALL_HIERARCHY_INCOMING_CALLS) def incoming_calls(params: lsp.CallHierarchyIncomingCallsParams): identity = call_hierarchy_identity(params.item) if identity is None: return [] indexes, definitions = LSP_SERVER.navigation_state() return incoming_call_hierarchy(identity, indexes, definitions) @LSP_SERVER.feature(lsp.CALL_HIERARCHY_OUTGOING_CALLS) def outgoing_calls(params: lsp.CallHierarchyOutgoingCallsParams): identity = call_hierarchy_identity(params.item) if identity is None: return [] indexes, definitions = LSP_SERVER.navigation_state() return outgoing_call_hierarchy(identity, indexes, definitions) # ********************************************************** # Linting features end here # ********************************************************** # ********************************************************** # Formatting features start here # ********************************************************** # Sample implementations: # Black: https://github.com/microsoft/vscode-black-formatter/blob/main/bundled/tool # ********************************************************** # Formatting features ends here # ********************************************************** @LSP_SERVER.feature(lsp.TEXT_DOCUMENT_FORMATTING) def formatting(params: lsp.DocumentFormattingParams) -> list[lsp.TextEdit] | None: """LSP handler for textDocument/formatting request.""" doc = LSP_SERVER.workspace.get_text_document(params.text_document.uri) source = doc.source start = lsp.Position(line=0, character=0) last_line = source.rsplit("\n", 1)[-1] end = lsp.Position(line=source.count("\n"), character=len(last_line)) if GLOBAL_SETTINGS.get("formatter", True): source = LSP_SERVER.format(doc, params.options) return [ lsp.TextEdit( range=lsp.Range(start=start, end=end), new_text=source, ) ] # ********************************************************** # Required Language Server Initialization and Exit handlers. # ********************************************************** @LSP_SERVER.feature(lsp.WORKSPACE_DID_CHANGE_CONFIGURATION) def did_change_configuration(_: lsp.DidChangeConfigurationParams): """LSP Handler for Config Changes""" @LSP_SERVER.feature(lsp.INITIALIZE) def initialize(params: lsp.InitializeParams) -> lsp.InitializeResult: """LSP handler for initialize request.""" log_to_output(f"CWD Server: {os.getcwd()}") paths = "\r\n ".join(sys.path) log_to_output(f"sys.path used to run Server:\r\n {paths}") GLOBAL_SETTINGS.update(**params.initialization_options.get("globalSettings", {})) INDEX_CACHE_PATH["path"] = params.initialization_options.get("indexCachePath") settings = params.initialization_options["settings"] _update_workspace_settings(settings) log_to_output(f"Settings used to run Server:\r\n{json.dumps(settings, indent=4, ensure_ascii=False)}\r\n") log_to_output(f"Global settings:\r\n{json.dumps(GLOBAL_SETTINGS, indent=4, ensure_ascii=False)}\r\n") semantic_tokens_legend = lsp.SemanticTokensLegend( token_types=TOKEN_TYPES, token_modifiers=[m.name for m in TokenModifier], ) return lsp.InitializeResult( capabilities=lsp.ServerCapabilities( document_formatting_provider=GLOBAL_SETTINGS.get("formatter", True), folding_range_provider=True, semantic_tokens_provider=lsp.SemanticTokensOptions(legend=semantic_tokens_legend, full=True, range=False), definition_provider=True, document_highlight_provider=True, references_provider=True, rename_provider=lsp.RenameOptions(prepare_provider=True), workspace_symbol_provider=True, call_hierarchy_provider=True, ) ) @LSP_SERVER.feature(lsp.INITIALIZED) def initialized(_params: lsp.InitializedParams): """Kick off background indexing to avoid blocking initialization.""" def index_workspace(): try: try: root = LSP_SERVER.workspace.root_path except RuntimeError: root = None if not root: log_to_output("Background indexing skipped: no workspace folder is open.") return log_to_output("Background indexing started...") started = time.perf_counter() LSP_SERVER.index_cache = IndexCache.load(INDEX_CACHE_PATH.get("path")) root_path = pathlib.Path(root) skipped_directories = { ".git", ".nox", ".venv", "dist", "node_modules", "out", } tcl_files = (path for path in root_path.rglob("*.tcl") if not any(part.casefold() in skipped_directories for part in path.relative_to(root_path).parts[:-1])) for filepath in sorted(tcl_files, key=lambda path: str(path).casefold()): try: document = TextDocument(uri=filepath.as_uri(), language_id="tcl") LSP_SERVER.update_poco_completion_for_file( document, cache_tree=False, require_file_exists=True, from_disk=True, ) except Exception as error: log_to_output(f"Fehler beim Parsen von {filepath}: {error}") _refresh_psc_index() LSP_SERVER.index_cache.save() log_to_output(f"Background indexing completed in {time.perf_counter() - started:.1f}s.") except Exception as e: log_to_output(f"Background indexing failed: {e}") threading.Thread(target=index_workspace, name="nxps-indexer", daemon=True).start() @LSP_SERVER.feature(lsp.EXIT) def on_exit(_params: Optional[Any] = None) -> None: """Handle clean up on exit.""" jsonrpc.shutdown_json_rpc() @LSP_SERVER.feature(lsp.SHUTDOWN) def on_shutdown(_params: Optional[Any] = None) -> None: """Handle clean up on shutdown.""" jsonrpc.shutdown_json_rpc() def _get_global_defaults(): return { "path": GLOBAL_SETTINGS.get("path", []), "interpreter": GLOBAL_SETTINGS.get("interpreter", [sys.executable]), "args": GLOBAL_SETTINGS.get("args", []), "importStrategy": GLOBAL_SETTINGS.get("importStrategy", "useBundled"), "showNotifications": GLOBAL_SETTINGS.get("showNotifications", "off"), "formatter": GLOBAL_SETTINGS.get("formatter", True), "inlayHint": GLOBAL_SETTINGS.get("inlayHint", True), "inlayHints": GLOBAL_SETTINGS.get( "inlayHints", { "parameterNames": "all", "suppressWhenArgumentMatchesName": True, }, ), } def _update_workspace_settings(settings): if not settings: key = os.getcwd() WORKSPACE_SETTINGS[key] = { "cwd": key, "workspaceFS": key, "workspace": uris.from_fs_path(key), **_get_global_defaults(), } return for setting in settings: key = uris.to_fs_path(setting["workspace"]) WORKSPACE_SETTINGS[key] = { "cwd": key, **setting, "workspaceFS": key, } def _get_settings_by_path(file_path: pathlib.Path): workspaces = {s["workspaceFS"] for s in WORKSPACE_SETTINGS.values()} while file_path != file_path.parent: str_file_path = str(file_path) if str_file_path in workspaces: return WORKSPACE_SETTINGS[str_file_path] file_path = file_path.parent setting_values = list(WORKSPACE_SETTINGS.values()) return setting_values[0] def _get_document_key(document: TextDocument): if WORKSPACE_SETTINGS: document_workspace = pathlib.Path(document.path) workspaces = {s["workspaceFS"] for s in WORKSPACE_SETTINGS.values()} # Find workspace settings for the given file. while document_workspace != document_workspace.parent: if str(document_workspace) in workspaces: return str(document_workspace) document_workspace = document_workspace.parent return None def _get_settings_by_document(document: TextDocument | None): if document is None or document.path is None: return list(WORKSPACE_SETTINGS.values())[0] key = _get_document_key(document) if key is None: # This is either a non-workspace file or there is no workspace. key = os.fspath(pathlib.Path(document.path).parent) return { "cwd": key, "workspaceFS": key, "workspace": uris.from_fs_path(key), **_get_global_defaults(), } return WORKSPACE_SETTINGS[str(key)] # ***************************************************** # Logging and notification. # ***************************************************** def log_to_output(message: str, msg_type: lsp.MessageType = lsp.MessageType.Log) -> None: LSP_SERVER.window_log_message(lsp.LogMessageParams(message=message, type=msg_type)) def log_error(message: str) -> None: log_to_output(message, lsp.MessageType.Error) if os.getenv("LS_SHOW_NOTIFICATION", "off") in ["onError", "onWarning", "always"]: LSP_SERVER.window_show_message(lsp.ShowMessageParams(message=message, type=lsp.MessageType.Error)) def log_warning(message: str) -> None: log_to_output(message, lsp.MessageType.Warning) if os.getenv("LS_SHOW_NOTIFICATION", "off") in ["onWarning", "always"]: LSP_SERVER.window_show_message(lsp.ShowMessageParams(message=message, type=lsp.MessageType.Warning)) def log_always(message: str) -> None: log_to_output(message, lsp.MessageType.Info) if os.getenv("LS_SHOW_NOTIFICATION", "off") in ["always"]: LSP_SERVER.window_show_message(lsp.ShowMessageParams(message=message, type=lsp.MessageType.Info)) # ***************************************************** # Start the server. # ***************************************************** if __name__ == "__main__": LSP_SERVER.start_io()