Christoph/nx_post_support
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects 1 Releases 39 Wiki Activity
Files
a1eccf90f4497014fac6a87fddfd080753d4612a
nx_post_support/server/src/lsp_server.py
T
Christoph da9091fa38 feat: complete BLOCK_LIST/ADDR_LIST keywords and space comments 
Add keyword-driven completions for BLOCK_LIST and ADDR_LIST in the LSP server.
- While typing a prefix of these keywords the server returns the keyword(s)
  as incomplete snippet suggestions that trigger a re-request.
- Once the keyword is typed the server replaces it with a full list of the
  corresponding loaded block templates or addresses (quoted), and sets each
  item's filter_text to include the keyword so further typing narrows results.
- Integrate this flow into on_completion and factor the symbol-list logic
  into a helper that returns either incomplete keyword suggestions or the
  completed symbol list.

Also implement NxFormatter.format_comment to ensure a single space after a
leading '#' for comments that don't already start with whitespace, while
leaving sequences of '#' (separators), shebangs, and already-spaced comments
unchanged. This affects both standalone and inline comments.

Add/rename tests to cover the new completions and comment-spacing behavior.
2026-09-24 13:57:19 +02:00

1192 lines
45 KiB
Python
Raw Blame History

# 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.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 (
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=["$", " ", "-", "(", ","]),
)
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
# 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"),
}
_WORD_BEFORE_CURSOR_RE = re.compile(r"(?<![$\w])[A-Za-z_]+$")
def _symbol_list_completion(source_lines, position: lsp.Position) -> 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
# 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
@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:
return None
indexes, definitions, _, identity = context
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:
return []
indexes, definitions, _, identity = context
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
if 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 re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", params.new_name):
return None
context = _navigation_context(params.text_document.uri, params.position)
if context is None:
return None
indexes, definitions, _, identity = context
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)
@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()
Reference in New Issue View Git Blame Copy Permalink