Christoph/nx_post_support
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects 1 Releases 25 Wiki Activity
Files
af195a577ba0861df358f3df42c2ad80afc7bcfe
nx_post_support/server/src/lsp_server.py
T
Christoph Brandau af195a577b
build_and_puplish.yml / build_and_publish (release) Successful in 29s
Details
refactor(lsp): cache analysis results and debounce diagnostics 
This change adds cached and incremental analysis for the LSP
server to improve responsiveness. The client now debounces
diagnostic updates to avoid excessive recomputation. The
server introduces per-document line caches and various
caches for completions, inlay hints, and metadata to
support faster, incremental updates.

- Debounce diagnostics on text changes to reduce noise.
- Add caches for completions, inlay hints, and metadata.
- Introduce incremental analysis with per-document line caches.
2026-08-19 13:41:53 +02:00

889 lines
30 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
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 lsp_jsonrpc as jsonrpc
import lsprotocol.types as lsp
from common.load_data import standard_items
from lsp_tclserver import TclLanguageServer
from pygls import uris, workspace
from pygls.workspace.text_document import TextDocument
from tools.folding_ranges import build_folding_ranges
from tools.inlay_hint import (
InlayHintGenerator,
build_builtin_inlay_signatures,
)
from tools.navigation import (
SymbolIdentity,
definition_identities,
matching_occurrences,
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
WORKSPACE_SETTINGS = {}
GLOBAL_SETTINGS = {}
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
}
BUILTIN_VARIABLE_NAMES = {item.label for item in standard_items.nx_variables}
STATIC_COMPLETION_ITEMS = tuple(
standard_items.tcl_keyword_list
+ standard_items.nx_procs
+ standard_items.nx_variables
)
STATIC_COMPLETION_KEYS = frozenset(
(item.label, getattr(item, "kind", None)) for item in STATIC_COMPLETION_ITEMS
)
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,
)
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:
if change.type == lsp.FileChangeType.Deleted:
LSP_SERVER.remove_file_state(change.uri)
else:
_index_tcl_file_from_disk(change.uri)
@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)
def on_completion(params: lsp.CompletionParams) -> lsp.CompletionList:
from tools.completion_items import BUILTIN_VAR_LABELS
doc = LSP_SERVER.workspace.get_text_document(params.text_document.uri)
workspace_items = LSP_SERVER.completion_items_snapshot()
tree = LSP_SERVER.get_tree(doc)
globals_set, procs_locals, proc_ranges = LSP_SERVER.variable_index_for_document(
doc, tree
)
# Always include globals (excluding built-ins)
dynamic_items = []
for name in sorted(globals_set):
if name not in BUILTIN_VAR_LABELS:
dynamic_items.append(
lsp.CompletionItem(label=name, kind=lsp.CompletionItemKind.Variable)
)
# Include proc-local variables when cursor is inside that proc
pos = params.position
if pos is not None:
for pr in proc_ranges:
if pr.start_line <= pos.line <= (pr.end_line or pr.start_line):
for name in sorted(procs_locals.get(pr.name, set())):
# Exclude built-ins and globals to avoid duplication
if name not in BUILTIN_VAR_LABELS and name not in globals_set:
dynamic_items.append(
lsp.CompletionItem(
label=name, kind=lsp.CompletionItemKind.Variable
)
)
break
# Merge with de-duplication. Each file keeps its complete index, so a proc
# declared in multiple files must only appear once in the completion list.
merged: list[lsp.CompletionItem] = list(STATIC_COMPLETION_ITEMS)
seen_items: set[tuple[str, lsp.CompletionItemKind | None]] = set(
STATIC_COMPLETION_KEYS
)
for it in (*workspace_items, *dynamic_items):
key = (it.label, getattr(it, "kind", None))
if key in seen_items:
continue
seen_items.add(key)
merged.append(it)
return lsp.CompletionList(is_incomplete=False, items=merged)
@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)
tree = LSP_SERVER.get_tree(document)
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,
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)
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 Tcl proc and variable definitions through the symbol index."""
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 = LSP_SERVER.navigation_snapshot()
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 = LSP_SERVER.navigation_snapshot()
index = indexes.get(filepath)
if index is None:
return None
definitions = definition_identities(indexes)
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)
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.PrepareRenameResult_Type1(
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)
# **********************************************************
# 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", {}))
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,
references_provider=True,
rename_provider=lsp.RenameOptions(prepare_provider=True),
workspace_symbol_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...")
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,
)
except Exception as error:
log_to_output(f"Fehler beim Parsen von {filepath}: {error}")
log_to_output("Background indexing completed.")
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: workspace.Document):
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: workspace.Document | 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.show_message_log(message, msg_type)
def log_error(message: str) -> None:
LSP_SERVER.show_message_log(message, lsp.MessageType.Error)
if os.getenv("LS_SHOW_NOTIFICATION", "off") in ["onError", "onWarning", "always"]:
LSP_SERVER.show_message(message, lsp.MessageType.Error)
def log_warning(message: str) -> None:
LSP_SERVER.show_message_log(message, lsp.MessageType.Warning)
if os.getenv("LS_SHOW_NOTIFICATION", "off") in ["onWarning", "always"]:
LSP_SERVER.show_message(message, lsp.MessageType.Warning)
def log_always(message: str) -> None:
LSP_SERVER.show_message_log(message, lsp.MessageType.Info)
if os.getenv("LS_SHOW_NOTIFICATION", "off") in ["always"]:
LSP_SERVER.show_message(message, lsp.MessageType.Info)
# *****************************************************
# Start the server.
# *****************************************************
if __name__ == "__main__":
LSP_SERVER.start_io()
Reference in New Issue View Git Blame Copy Permalink