Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
af195a577b | ||
|
|
ecb50be2b8 | ||
|
|
61d4785775 |
@@ -1,8 +1,13 @@
|
||||
## Unreleased
|
||||
|
||||
- Prevent truncated TCL inlay hints and add configurable parameter hint modes
|
||||
- Add inlay hints for built-in NX procedures, variadic arguments, and visible ranges
|
||||
- Add inlay hint documentation and navigation to custom procedure definitions
|
||||
- Add signature help for custom TCL procedures and built-in NX/MOM procedures
|
||||
- Clean stale TCL indexes on close, delete, and rename operations
|
||||
- Make background parsing and index updates thread-safe
|
||||
- Improve TCL response times with debounced edits and cached semantic, inlay, hover, completion, and variable indexes
|
||||
- Debounce CDL/DEF diagnostics and remove per-line diagnostic logging
|
||||
|
||||
## [0.0.1]
|
||||
|
||||
|
||||
@@ -31,6 +31,12 @@ The extension can be configured through VS Code settings:
|
||||
- `nx-post-support.interpreter` - Specify custom Python interpreter path for the language server
|
||||
- `nx-post-support.formatter` - Enable/disable the TCL formatter (default: false)
|
||||
- `nx-post-support.inlayHint` - Enable/disable inlay Hints (default: true)
|
||||
- `nx-post-support.inlayHints.parameterNames` - Show parameter names for `all`, only `literals`, or `none` (default: `all`)
|
||||
- `nx-post-support.inlayHints.suppressWhenArgumentMatchesName` - Hide redundant hints such as `value:` before `$value` (default: true)
|
||||
|
||||
TCL files default to unlimited inlay hint length so that VS Code does not
|
||||
truncate later parameter names on a line. An explicit user setting for
|
||||
`editor.inlayHints.maximumLength` still takes precedence.
|
||||
|
||||
## Usage
|
||||
|
||||
|
||||
@@ -4,6 +4,8 @@ import {
|
||||
createCdlEventHandlerSnippet
|
||||
} from "./cdlEventHandler"
|
||||
|
||||
const MACHINE_HEADER_REGEX = /^MACHINE\s+\S+/
|
||||
|
||||
export function formatCdlFile(content: string): string {
|
||||
let indentLevel = 0
|
||||
const formattedLines = []
|
||||
@@ -41,15 +43,19 @@ export function formatDefFile(content: string): string {
|
||||
}
|
||||
|
||||
export function isFirstLineMachine(content: string): boolean {
|
||||
const lines = content.split("\n").map((line) => line.trim())
|
||||
for (const line of lines) {
|
||||
console.log(line)
|
||||
let lineStart = 0
|
||||
while (lineStart <= content.length) {
|
||||
const newline = content.indexOf("\n", lineStart)
|
||||
const lineEnd = newline === -1 ? content.length : newline
|
||||
const line = content.slice(lineStart, lineEnd).trim()
|
||||
if (line === "" || line.startsWith("#")) {
|
||||
console.log("skipping line")
|
||||
if (newline === -1) {
|
||||
return false
|
||||
}
|
||||
lineStart = newline + 1
|
||||
continue
|
||||
}
|
||||
const machineRegex = /^MACHINE\s+\S+/
|
||||
return machineRegex.test(line)
|
||||
return MACHINE_HEADER_REGEX.test(line)
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -57,10 +63,11 @@ export function isFirstLineMachine(content: string): boolean {
|
||||
export function diagnosticHandler(document: vscode.TextDocument) {
|
||||
const diagnostics: vscode.Diagnostic[] = []
|
||||
if (document.languageId === "cdl" || document.languageId === "def") {
|
||||
if (!isFirstLineMachine(document.getText())) {
|
||||
const text = document.getText()
|
||||
if (!isFirstLineMachine(text)) {
|
||||
const range = new vscode.Range(
|
||||
document.positionAt(0),
|
||||
document.positionAt(document.getText().length)
|
||||
document.positionAt(text.length)
|
||||
)
|
||||
const diagnostic = new vscode.Diagnostic(
|
||||
range,
|
||||
|
||||
@@ -18,6 +18,12 @@ export interface ISettings {
|
||||
interpreter: string[]
|
||||
importStrategy: string
|
||||
showNotifications: string
|
||||
formatter: boolean
|
||||
inlayHint: boolean
|
||||
inlayHints: {
|
||||
parameterNames: "all" | "literals" | "none"
|
||||
suppressWhenArgumentMatchesName: boolean
|
||||
}
|
||||
}
|
||||
|
||||
export function getExtensionSettings(
|
||||
@@ -80,7 +86,13 @@ export async function getWorkspaceSettings(
|
||||
importStrategy: config.get<string>(`importStrategy`) ?? "useBundled",
|
||||
showNotifications: config.get<string>(`showNotifications`) ?? "off",
|
||||
formatter: config.get<boolean>(`formatter`) ?? true,
|
||||
inlayHint: config.get<boolean>(`inlayHint`) ?? true
|
||||
inlayHint: config.get<boolean>(`inlayHint`) ?? true,
|
||||
inlayHints: {
|
||||
parameterNames:
|
||||
config.get<"all" | "literals" | "none">(`inlayHints.parameterNames`) ?? "all",
|
||||
suppressWhenArgumentMatchesName:
|
||||
config.get<boolean>(`inlayHints.suppressWhenArgumentMatchesName`) ?? true
|
||||
}
|
||||
}
|
||||
return workspaceSetting
|
||||
}
|
||||
@@ -113,7 +125,13 @@ export async function getGlobalSettings(
|
||||
importStrategy: getGlobalValue<string>(config, "importStrategy", "useBundled"),
|
||||
showNotifications: getGlobalValue<string>(config, "showNotifications", "off"),
|
||||
formatter: config.get<boolean>(`formatter`) ?? true,
|
||||
inlayHint: config.get<boolean>(`inlayHint`) ?? true
|
||||
inlayHint: config.get<boolean>(`inlayHint`) ?? true,
|
||||
inlayHints: {
|
||||
parameterNames:
|
||||
config.get<"all" | "literals" | "none">(`inlayHints.parameterNames`) ?? "all",
|
||||
suppressWhenArgumentMatchesName:
|
||||
config.get<boolean>(`inlayHints.suppressWhenArgumentMatchesName`) ?? true
|
||||
}
|
||||
}
|
||||
return setting
|
||||
}
|
||||
@@ -129,7 +147,9 @@ export function checkIfConfigurationChanged(
|
||||
`${namespace}.importStrategy`,
|
||||
`${namespace}.showNotifications`,
|
||||
`${namespace}.formatter`,
|
||||
`${namespace}.inlayHint`
|
||||
`${namespace}.inlayHint`,
|
||||
`${namespace}.inlayHints.parameterNames`,
|
||||
`${namespace}.inlayHints.suppressWhenArgumentMatchesName`
|
||||
]
|
||||
const changed = settings.map((s) => e.affectsConfiguration(s))
|
||||
return changed.includes(true)
|
||||
|
||||
+50
-11
@@ -246,28 +246,67 @@ export async function activate(context: vscode.ExtensionContext) {
|
||||
const diagnosticCollectionDef = vscode.languages.createDiagnosticCollection("def")
|
||||
context.subscriptions.push(diagnosticCollectionCdl, diagnosticCollectionDef)
|
||||
|
||||
const diagnosticTimers = new Map<string, ReturnType<typeof setTimeout>>()
|
||||
const updateDiagnostics = (document: vscode.TextDocument) => {
|
||||
if (document.languageId === "cdl") {
|
||||
diagnosticCollectionCdl.set(document.uri, diagnosticHandler(document))
|
||||
} else if (document.languageId === "def") {
|
||||
diagnosticCollectionDef.set(document.uri, diagnosticHandler(document))
|
||||
}
|
||||
}
|
||||
const scheduleDiagnostics = (document: vscode.TextDocument) => {
|
||||
const key = document.uri.toString()
|
||||
const previous = diagnosticTimers.get(key)
|
||||
if (previous !== undefined) {
|
||||
clearTimeout(previous)
|
||||
}
|
||||
diagnosticTimers.set(
|
||||
key,
|
||||
setTimeout(() => {
|
||||
diagnosticTimers.delete(key)
|
||||
updateDiagnostics(document)
|
||||
}, 120)
|
||||
)
|
||||
}
|
||||
|
||||
// Check if the first line of the CDL file contains "MACHINE"
|
||||
context.subscriptions.push(
|
||||
vscode.workspace.onDidOpenTextDocument((document) => {
|
||||
if (document.languageId === "cdl" || document.languageId === "def") {
|
||||
if (document.languageId === "cdl") {
|
||||
diagnosticCollectionCdl.set(document.uri, diagnosticHandler(document))
|
||||
} else if (document.languageId === "def") {
|
||||
diagnosticCollectionDef.set(document.uri, diagnosticHandler(document))
|
||||
}
|
||||
updateDiagnostics(document)
|
||||
}
|
||||
}),
|
||||
vscode.workspace.onDidChangeTextDocument((event) => {
|
||||
const document = event.document
|
||||
if (document.languageId === "cdl" || document.languageId === "def") {
|
||||
if (document.languageId === "cdl") {
|
||||
diagnosticCollectionCdl.set(document.uri, diagnosticHandler(document))
|
||||
} else if (document.languageId === "def") {
|
||||
diagnosticCollectionDef.set(document.uri, diagnosticHandler(document))
|
||||
}
|
||||
scheduleDiagnostics(document)
|
||||
}
|
||||
})
|
||||
}),
|
||||
vscode.workspace.onDidCloseTextDocument((document) => {
|
||||
const key = document.uri.toString()
|
||||
const timer = diagnosticTimers.get(key)
|
||||
if (timer !== undefined) {
|
||||
clearTimeout(timer)
|
||||
diagnosticTimers.delete(key)
|
||||
}
|
||||
diagnosticCollectionCdl.delete(document.uri)
|
||||
diagnosticCollectionDef.delete(document.uri)
|
||||
}),
|
||||
{
|
||||
dispose() {
|
||||
for (const timer of diagnosticTimers.values()) {
|
||||
clearTimeout(timer)
|
||||
}
|
||||
diagnosticTimers.clear()
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
for (const document of vscode.workspace.textDocuments) {
|
||||
if (document.languageId === "cdl" || document.languageId === "def") {
|
||||
updateDiagnostics(document)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export function deactivate(): Thenable<void> | undefined {
|
||||
|
||||
+26
-1
@@ -2,7 +2,7 @@
|
||||
"name": "nx-post-support",
|
||||
"displayName": "NX Postprocessor Support",
|
||||
"description": "VS Code extension for NX CAM postprocessor development with syntax highlighting, formatting, linting, and auto-completion for CDL, TCL, and DEF files",
|
||||
"version": "2026.6.201",
|
||||
"version": "2026.8.100",
|
||||
"publisher": "Christoph",
|
||||
"icon": "images/nx-1.png",
|
||||
"extensionDependencies": [
|
||||
@@ -96,6 +96,26 @@
|
||||
"default": true,
|
||||
"description": "Use the Inlay Hints in from `NX Postprocessor Support`"
|
||||
},
|
||||
"nx-post-support.inlayHints.parameterNames": {
|
||||
"type": "string",
|
||||
"default": "all",
|
||||
"enum": [
|
||||
"all",
|
||||
"literals",
|
||||
"none"
|
||||
],
|
||||
"enumDescriptions": [
|
||||
"Show parameter name hints for all arguments.",
|
||||
"Show parameter name hints only for literal arguments.",
|
||||
"Do not show parameter name hints."
|
||||
],
|
||||
"description": "Controls which TCL procedure arguments receive parameter name hints."
|
||||
},
|
||||
"nx-post-support.inlayHints.suppressWhenArgumentMatchesName": {
|
||||
"type": "boolean",
|
||||
"default": true,
|
||||
"description": "Hide a parameter hint when a variable argument already has the same name, for example `output` in `my_proc $output`."
|
||||
},
|
||||
"nx-post-support.importStrategy": {
|
||||
"default": "useBundled",
|
||||
"description": "Defines where `NX Postprocessor Support` is imported from.",
|
||||
@@ -120,6 +140,11 @@
|
||||
"type": "array"
|
||||
}
|
||||
}
|
||||
},
|
||||
"configurationDefaults": {
|
||||
"[tcl]": {
|
||||
"editor.inlayHints.maximumLength": 0
|
||||
}
|
||||
}
|
||||
},
|
||||
"scripts": {
|
||||
|
||||
+96
-79
@@ -5,15 +5,15 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import operator
|
||||
import os
|
||||
import pathlib
|
||||
import re
|
||||
import sys
|
||||
import threading
|
||||
|
||||
from typing import Any, Optional
|
||||
import operator
|
||||
from collections import ChainMap
|
||||
from functools import reduce
|
||||
from typing import Any, Optional
|
||||
|
||||
|
||||
# **********************************************************
|
||||
@@ -40,11 +40,15 @@ update_sys_path(
|
||||
# pylint: disable=wrong-import-position,import-error
|
||||
import lsp_jsonrpc as jsonrpc
|
||||
import lsprotocol.types as lsp
|
||||
from pygls import uris, workspace
|
||||
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.semantic_tokens import _Highlighter, TOKEN_TYPES, TokenModifier
|
||||
from tools.inlay_hint import InlayHintGenerator
|
||||
from tools.inlay_hint import (
|
||||
InlayHintGenerator,
|
||||
build_builtin_inlay_signatures,
|
||||
)
|
||||
from tools.navigation import (
|
||||
SymbolIdentity,
|
||||
definition_identities,
|
||||
@@ -52,10 +56,13 @@ from tools.navigation import (
|
||||
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 lsp_tclserver import TclLanguageServer
|
||||
from pygls.workspace.text_document import TextDocument
|
||||
|
||||
|
||||
WORKSPACE_SETTINGS = {}
|
||||
GLOBAL_SETTINGS = {}
|
||||
@@ -71,6 +78,23 @@ BUILTIN_PROC_NAMES = {
|
||||
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.
|
||||
@@ -91,15 +115,14 @@ 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.compute_diagnostics(document)
|
||||
# Also update custom completion and proc docs for this file
|
||||
LSP_SERVER.update_poco_completion_for_file(document)
|
||||
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."""
|
||||
_ = LSP_SERVER.workspace.get_text_document(params.text_document.uri)
|
||||
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)
|
||||
@@ -115,8 +138,7 @@ 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.compute_diagnostics(document)
|
||||
LSP_SERVER.update_poco_completion_for_file(document)
|
||||
LSP_SERVER.schedule_document_analysis(document)
|
||||
|
||||
|
||||
FILE_OPERATION_OPTIONS = lsp.FileOperationRegistrationOptions(
|
||||
@@ -199,10 +221,12 @@ def did_change_watched_files(params: lsp.DidChangeWatchedFilesParams) -> None:
|
||||
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
|
||||
if diagnostic_state is None:
|
||||
doc = LSP_SERVER.workspace.get_text_document(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)
|
||||
|
||||
@@ -220,24 +244,15 @@ def document_diagnostic(params: lsp.DocumentDiagnosticParams):
|
||||
|
||||
@LSP_SERVER.feature(lsp.TEXT_DOCUMENT_COMPLETION)
|
||||
def on_completion(params: lsp.CompletionParams) -> lsp.CompletionList:
|
||||
from tools.variable_index import build_variable_index
|
||||
from tools.completion_items import BUILTIN_VAR_LABELS
|
||||
|
||||
doc = LSP_SERVER.workspace.get_text_document(params.text_document.uri)
|
||||
|
||||
# Base items
|
||||
poco_completion, _, _ = LSP_SERVER.index_snapshot()
|
||||
poco = [item for items in poco_completion.values() for item in items]
|
||||
base_items = (
|
||||
standard_items.tcl_keyword_list
|
||||
+ standard_items.nx_procs
|
||||
+ standard_items.nx_variables
|
||||
+ poco
|
||||
)
|
||||
|
||||
# Build variable index from current document
|
||||
workspace_items = LSP_SERVER.completion_items_snapshot()
|
||||
tree = LSP_SERVER.get_tree(doc)
|
||||
globals_set, procs_locals, proc_ranges = build_variable_index(doc.source, tree)
|
||||
globals_set, procs_locals, proc_ranges = LSP_SERVER.variable_index_for_document(
|
||||
doc, tree
|
||||
)
|
||||
|
||||
# Always include globals (excluding built-ins)
|
||||
dynamic_items = []
|
||||
@@ -264,9 +279,11 @@ def on_completion(params: lsp.CompletionParams) -> lsp.CompletionList:
|
||||
|
||||
# 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] = []
|
||||
seen_items: set[tuple[str, lsp.CompletionItemKind | None]] = set()
|
||||
for it in base_items + dynamic_items:
|
||||
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
|
||||
@@ -287,22 +304,9 @@ 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)
|
||||
|
||||
filepath = str(pathlib.Path(uris.to_fs_path(document.uri)))
|
||||
custom_signatures: dict[str, list[str]] = {}
|
||||
custom_docs: dict[str, str] = {}
|
||||
_, proc_signatures, proc_docs = LSP_SERVER.index_snapshot()
|
||||
|
||||
# Prefer declarations from the current document if duplicate proc names
|
||||
# exist in the workspace.
|
||||
for indexed_path, signatures in proc_signatures.items():
|
||||
if indexed_path != filepath:
|
||||
custom_signatures.update(signatures)
|
||||
custom_signatures.update(proc_signatures.get(filepath, {}))
|
||||
|
||||
for indexed_path, docs in proc_docs.items():
|
||||
if indexed_path != filepath:
|
||||
custom_docs.update(docs)
|
||||
custom_docs.update(proc_docs.get(filepath, {}))
|
||||
custom_signatures, custom_docs = LSP_SERVER.proc_metadata_snapshot(
|
||||
document.path
|
||||
)
|
||||
|
||||
return build_signature_help(
|
||||
document.source,
|
||||
@@ -333,22 +337,37 @@ def document_symbols(params: lsp.DocumentSymbolParams):
|
||||
|
||||
@LSP_SERVER.feature(lsp.TEXT_DOCUMENT_INLAY_HINT)
|
||||
def inlay_hints(params: lsp.InlayHintParams):
|
||||
if not GLOBAL_SETTINGS.get("inlayHint", False):
|
||||
return []
|
||||
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)
|
||||
|
||||
# Merge proc signatures across files and traverse once
|
||||
merged_signatures = {}
|
||||
_, proc_signatures, _ = LSP_SERVER.index_snapshot()
|
||||
for sigs in proc_signatures.values():
|
||||
merged_signatures.update(sigs)
|
||||
# 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(merged_signatures)
|
||||
tree.accept(generator, recurse=True)
|
||||
return generator.hints
|
||||
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(
|
||||
@@ -363,8 +382,7 @@ def semantic_tokens(params: lsp.SemanticTokensParams):
|
||||
|
||||
data = []
|
||||
plugins = []
|
||||
poco_completion, _, _ = LSP_SERVER.index_snapshot()
|
||||
hl = _Highlighter(plugins, poco_completion)
|
||||
hl = _Highlighter(plugins, LSP_SERVER.custom_function_names_snapshot())
|
||||
|
||||
# Reuse cached AST
|
||||
tree = LSP_SERVER.get_tree(document)
|
||||
@@ -377,7 +395,7 @@ def semantic_tokens(params: lsp.SemanticTokensParams):
|
||||
token.line,
|
||||
token.offset,
|
||||
token.length,
|
||||
TOKEN_TYPES.index(token.tok_type),
|
||||
TOKEN_TYPE_INDEX[token.tok_type],
|
||||
reduce(operator.or_, token.tok_modifiers, 0),
|
||||
]
|
||||
)
|
||||
@@ -399,14 +417,14 @@ def hover(params: lsp.HoverParams) -> lsp.Hover:
|
||||
col = params.position.character
|
||||
|
||||
try:
|
||||
line = document.lines[pos.line]
|
||||
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_position
|
||||
from tools.proc_docs import is_proc_declaration_line
|
||||
|
||||
if is_proc_declaration_position(document.source, pos.line, pos.character):
|
||||
if is_proc_declaration_line(line, pos.character):
|
||||
return None
|
||||
|
||||
# Identify the token under the cursor
|
||||
@@ -418,11 +436,7 @@ def hover(params: lsp.HoverParams) -> lsp.Hover:
|
||||
return None
|
||||
|
||||
# 1) If token is a known MOM proc/variable, return built-in hover
|
||||
command = token
|
||||
data = standard_items.json_data
|
||||
all_items = data.get("MOM_procs", []) + data.get("mom_variables", [])
|
||||
|
||||
match = next((item for item in all_items if item["label"] == command), None)
|
||||
match = BUILTIN_HOVER_ITEMS.get(token)
|
||||
if match and match.get("kind") == "function":
|
||||
label = match.get("label", "")
|
||||
parameters = match.get("parameters", [])
|
||||
@@ -455,14 +469,10 @@ def hover(params: lsp.HoverParams) -> lsp.Hover:
|
||||
|
||||
# 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_docs: dict[str, str] = {}
|
||||
_, _, indexed_proc_docs = LSP_SERVER.index_snapshot()
|
||||
for file_docs in indexed_proc_docs.values():
|
||||
proc_docs.update(file_docs)
|
||||
|
||||
if token in proc_docs:
|
||||
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_docs[token])
|
||||
lsp.MarkupContent(kind=lsp.MarkupKind.Markdown, value=proc_doc)
|
||||
)
|
||||
|
||||
return None
|
||||
@@ -490,7 +500,7 @@ 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:
|
||||
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()
|
||||
@@ -769,6 +779,13 @@ def _get_global_defaults():
|
||||
"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,
|
||||
},
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
|
||||
+320
-5
@@ -11,15 +11,16 @@ from pygls.workspace.text_document import TextDocument
|
||||
from tclint.format import FormatterOpts
|
||||
from tclint.lexer import TclSyntaxError
|
||||
from tclint.violations import Violation
|
||||
|
||||
from tools import checks, parser
|
||||
from tools.completion_items import CompletionCollector
|
||||
from tools.formatter import NxFormatter as Formatter
|
||||
from tools.inlay_hint import InlayHintSignature, build_custom_inlay_signatures
|
||||
from tools.navigation import FileSymbolIndex, build_file_symbol_index
|
||||
from tools.proc_docs import build_proc_docs
|
||||
|
||||
from tools.variable_index import ProcRange, build_variable_index
|
||||
|
||||
DIAGNOSTIC_SOURCE = "nx-post-support"
|
||||
LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class TclLanguageServer(server.LanguageServer):
|
||||
@@ -33,15 +34,40 @@ class TclLanguageServer(server.LanguageServer):
|
||||
self.proc_signatures: dict = {}
|
||||
self.proc_docs: dict = {}
|
||||
self.navigation_indexes: dict[str, FileSymbolIndex] = {}
|
||||
self.variable_indexes: dict[
|
||||
str,
|
||||
tuple[
|
||||
int | None,
|
||||
tuple[set[str], dict[str, set[str]], list[ProcRange]],
|
||||
],
|
||||
] = {}
|
||||
# Cache: (uri, version) -> (tree, violations)
|
||||
self._ast_cache = {}
|
||||
self._line_cache: dict[tuple[str, int | None], tuple[str, ...]] = {}
|
||||
self._parser_lock = threading.RLock()
|
||||
self._index_lock = threading.RLock()
|
||||
self._index_tokens: dict[str, int] = {}
|
||||
self._index_versions: dict[str, int | None] = {}
|
||||
self._committed_index_versions: dict[str, int | None] = {}
|
||||
self._next_index_token = 0
|
||||
self._diagnostic_tokens: dict[str, int] = {}
|
||||
self._next_diagnostic_token = 0
|
||||
self._index_generation = 0
|
||||
self._workspace_completion_cache: tuple[int, tuple] = (-1, ())
|
||||
self._custom_function_names_cache: tuple[int, frozenset[str]] = (
|
||||
-1,
|
||||
frozenset(),
|
||||
)
|
||||
self._proc_metadata_cache: dict[
|
||||
str, tuple[int, dict[str, list[str]], dict[str, str]]
|
||||
] = {}
|
||||
self._custom_inlay_cache: dict[
|
||||
str, tuple[int, dict[str, InlayHintSignature]]
|
||||
] = {}
|
||||
self._analysis_lock = threading.RLock()
|
||||
self._analysis_timers: dict[str, threading.Timer] = {}
|
||||
self._analysis_tokens: dict[str, int] = {}
|
||||
self._next_analysis_token = 0
|
||||
|
||||
def _parse_source(self, source: str):
|
||||
self.parser.violations = []
|
||||
@@ -74,11 +100,24 @@ class TclLanguageServer(server.LanguageServer):
|
||||
self._ast_cache[key] = (tree, violations)
|
||||
return tree, violations
|
||||
|
||||
def get_lines(self, document: TextDocument) -> tuple[str, ...]:
|
||||
"""Return split source lines once per document version."""
|
||||
key = (document.uri, document.version)
|
||||
with self._parser_lock:
|
||||
lines = self._line_cache.get(key)
|
||||
if lines is None:
|
||||
lines = tuple(document.source.splitlines())
|
||||
self._line_cache[key] = lines
|
||||
return lines
|
||||
|
||||
def clear_cache_for_uri(self, uri: str):
|
||||
with self._parser_lock:
|
||||
to_delete = [key for key in self._ast_cache if key[0] == uri]
|
||||
for key in to_delete:
|
||||
del self._ast_cache[key]
|
||||
for key in list(self._line_cache):
|
||||
if key[0] == uri:
|
||||
del self._line_cache[key]
|
||||
|
||||
@staticmethod
|
||||
def _normalized_path(path: pathlib.Path | str) -> str:
|
||||
@@ -101,6 +140,263 @@ class TclLanguageServer(server.LanguageServer):
|
||||
) -> bool:
|
||||
return cls._normalized_path(first) == cls._normalized_path(second)
|
||||
|
||||
def _invalidate_workspace_caches_locked(self) -> None:
|
||||
"""Invalidate request-level aggregates after an index mutation."""
|
||||
self._index_generation += 1
|
||||
self._workspace_completion_cache = (-1, ())
|
||||
self._custom_function_names_cache = (-1, frozenset())
|
||||
self._proc_metadata_cache.clear()
|
||||
self._custom_inlay_cache.clear()
|
||||
|
||||
def completion_items_snapshot(self) -> tuple:
|
||||
"""Return de-duplicated workspace completion items, cached by generation."""
|
||||
with self._index_lock:
|
||||
generation, items = self._workspace_completion_cache
|
||||
if generation == self._index_generation:
|
||||
return items
|
||||
|
||||
merged = []
|
||||
seen = set()
|
||||
for path_items in self.poco_completion.values():
|
||||
for item in path_items:
|
||||
key = (item.label, getattr(item, "kind", None))
|
||||
if key in seen:
|
||||
continue
|
||||
seen.add(key)
|
||||
merged.append(item)
|
||||
|
||||
items = tuple(merged)
|
||||
self._workspace_completion_cache = (self._index_generation, items)
|
||||
return items
|
||||
|
||||
def custom_function_names_snapshot(self) -> frozenset[str]:
|
||||
"""Return custom completion labels for semantic highlighting."""
|
||||
with self._index_lock:
|
||||
generation, names = self._custom_function_names_cache
|
||||
if generation == self._index_generation:
|
||||
return names
|
||||
|
||||
names = frozenset(
|
||||
item.label
|
||||
for path_items in self.poco_completion.values()
|
||||
for item in path_items
|
||||
)
|
||||
self._custom_function_names_cache = (self._index_generation, names)
|
||||
return names
|
||||
|
||||
def proc_metadata_snapshot(
|
||||
self, current_path: pathlib.Path | str
|
||||
) -> tuple[dict[str, list[str]], dict[str, str]]:
|
||||
"""Return merged proc metadata, preferring declarations in the active file."""
|
||||
normalized_current = self._normalized_path(current_path)
|
||||
with self._index_lock:
|
||||
cached = self._proc_metadata_cache.get(normalized_current)
|
||||
if cached is not None and cached[0] == self._index_generation:
|
||||
return cached[1], cached[2]
|
||||
|
||||
signatures: dict[str, list[str]] = {}
|
||||
docs: dict[str, str] = {}
|
||||
signature_paths = sorted(
|
||||
self.proc_signatures,
|
||||
key=lambda path: self._normalized_path(path).casefold(),
|
||||
)
|
||||
doc_paths = sorted(
|
||||
self.proc_docs,
|
||||
key=lambda path: self._normalized_path(path).casefold(),
|
||||
)
|
||||
|
||||
for path in signature_paths:
|
||||
if self._normalized_path(path) != normalized_current:
|
||||
signatures.update(self.proc_signatures[path])
|
||||
for path in signature_paths:
|
||||
if self._normalized_path(path) == normalized_current:
|
||||
signatures.update(self.proc_signatures[path])
|
||||
|
||||
for path in doc_paths:
|
||||
if self._normalized_path(path) != normalized_current:
|
||||
docs.update(self.proc_docs[path])
|
||||
for path in doc_paths:
|
||||
if self._normalized_path(path) == normalized_current:
|
||||
docs.update(self.proc_docs[path])
|
||||
|
||||
cached_value = (self._index_generation, signatures, docs)
|
||||
self._proc_metadata_cache[normalized_current] = cached_value
|
||||
return signatures, docs
|
||||
|
||||
def proc_documentation(
|
||||
self, name: str, current_path: pathlib.Path | str
|
||||
) -> str | None:
|
||||
_, docs = self.proc_metadata_snapshot(current_path)
|
||||
return docs.get(name)
|
||||
|
||||
def custom_inlay_signatures_snapshot(
|
||||
self, current_path: pathlib.Path | str
|
||||
) -> dict[str, InlayHintSignature]:
|
||||
"""Return custom inlay signatures cached until the workspace index changes."""
|
||||
normalized_current = self._normalized_path(current_path)
|
||||
with self._index_lock:
|
||||
cached = self._custom_inlay_cache.get(normalized_current)
|
||||
if cached is not None and cached[0] == self._index_generation:
|
||||
return cached[1]
|
||||
|
||||
signatures = build_custom_inlay_signatures(
|
||||
self.proc_signatures,
|
||||
self.proc_docs,
|
||||
self.navigation_indexes,
|
||||
os.fspath(current_path),
|
||||
)
|
||||
self._custom_inlay_cache[normalized_current] = (
|
||||
self._index_generation,
|
||||
signatures,
|
||||
)
|
||||
return signatures
|
||||
|
||||
def variable_index_for_document(
|
||||
self, document: TextDocument, tree=None
|
||||
) -> tuple[set[str], dict[str, set[str]], list[ProcRange]]:
|
||||
"""Return the per-version variable index used by completion requests."""
|
||||
filepath = str(pathlib.Path(uris.to_fs_path(document.uri)))
|
||||
with self._index_lock:
|
||||
cached = self.variable_indexes.get(filepath)
|
||||
if cached is not None and cached[0] == document.version:
|
||||
return cached[1]
|
||||
|
||||
if tree is None:
|
||||
tree = self.get_tree(document)
|
||||
variable_index = build_variable_index(document.source, tree)
|
||||
|
||||
with self._index_lock:
|
||||
cached = self.variable_indexes.get(filepath)
|
||||
if (
|
||||
cached is None
|
||||
or cached[0] is None
|
||||
or document.version is None
|
||||
or cached[0] <= document.version
|
||||
):
|
||||
self.variable_indexes[filepath] = (document.version, variable_index)
|
||||
return variable_index
|
||||
return cached[1]
|
||||
|
||||
def index_is_current(self, document: TextDocument) -> bool:
|
||||
filepath = str(pathlib.Path(uris.to_fs_path(document.uri)))
|
||||
with self._index_lock:
|
||||
return (
|
||||
filepath in self._committed_index_versions
|
||||
and self._committed_index_versions[filepath] == document.version
|
||||
)
|
||||
|
||||
def index_update_pending(self, filepath: pathlib.Path | str) -> bool:
|
||||
filepath = os.fspath(filepath)
|
||||
with self._index_lock:
|
||||
return (
|
||||
filepath not in self._committed_index_versions
|
||||
or self._index_versions.get(filepath)
|
||||
!= self._committed_index_versions[filepath]
|
||||
)
|
||||
|
||||
def _cancel_document_analysis(self, uri: str) -> None:
|
||||
with self._analysis_lock:
|
||||
timer = self._analysis_timers.pop(uri, None)
|
||||
self._analysis_tokens.pop(uri, None)
|
||||
if timer is not None:
|
||||
timer.cancel()
|
||||
|
||||
def cancel_analysis_under_uri(self, uri: str) -> None:
|
||||
"""Cancel delayed analysis for a closed/deleted file or folder."""
|
||||
try:
|
||||
target = pathlib.Path(uris.to_fs_path(uri))
|
||||
except (TypeError, ValueError):
|
||||
self._cancel_document_analysis(uri)
|
||||
return
|
||||
|
||||
with self._analysis_lock:
|
||||
matching_uris = []
|
||||
for pending_uri in self._analysis_timers:
|
||||
try:
|
||||
pending_path = pathlib.Path(uris.to_fs_path(pending_uri))
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
if self._is_same_or_child(pending_path, target):
|
||||
matching_uris.append(pending_uri)
|
||||
|
||||
timers = [self._analysis_timers.pop(key) for key in matching_uris]
|
||||
for key in matching_uris:
|
||||
self._analysis_tokens.pop(key, None)
|
||||
|
||||
for timer in timers:
|
||||
timer.cancel()
|
||||
|
||||
def _invalidate_document_work(self, document: TextDocument) -> None:
|
||||
"""Prevent older diagnostic/index work from committing after a new edit."""
|
||||
filepath = str(pathlib.Path(uris.to_fs_path(document.uri)))
|
||||
with self._index_lock:
|
||||
self._next_diagnostic_token += 1
|
||||
self._diagnostic_tokens[document.uri] = self._next_diagnostic_token
|
||||
self._next_index_token += 1
|
||||
self._index_tokens[filepath] = self._next_index_token
|
||||
self._index_versions[filepath] = document.version
|
||||
|
||||
def schedule_document_analysis(
|
||||
self, document: TextDocument, delay_seconds: float = 0.15
|
||||
) -> None:
|
||||
"""Coalesce rapid edits and analyze only the latest immutable snapshot."""
|
||||
snapshot = TextDocument(
|
||||
uri=document.uri,
|
||||
source=document.source,
|
||||
version=document.version,
|
||||
language_id=document.language_id,
|
||||
)
|
||||
self._invalidate_document_work(snapshot)
|
||||
|
||||
with self._analysis_lock:
|
||||
previous = self._analysis_timers.pop(snapshot.uri, None)
|
||||
if previous is not None:
|
||||
previous.cancel()
|
||||
|
||||
self._next_analysis_token += 1
|
||||
token = self._next_analysis_token
|
||||
self._analysis_tokens[snapshot.uri] = token
|
||||
|
||||
def analyze() -> None:
|
||||
with self._analysis_lock:
|
||||
if self._analysis_tokens.get(snapshot.uri) != token:
|
||||
return
|
||||
|
||||
try:
|
||||
diagnostic_state = self.diagnostic_snapshot(snapshot.uri)
|
||||
if (
|
||||
diagnostic_state is None
|
||||
or diagnostic_state[0] != snapshot.version
|
||||
):
|
||||
self.compute_diagnostics(snapshot)
|
||||
with self._analysis_lock:
|
||||
if self._analysis_tokens.get(snapshot.uri) != token:
|
||||
return
|
||||
if not self.index_is_current(snapshot):
|
||||
self.update_poco_completion_for_file(snapshot)
|
||||
except Exception:
|
||||
LOGGER.exception("Delayed analysis failed for %s", snapshot.uri)
|
||||
finally:
|
||||
with self._analysis_lock:
|
||||
if self._analysis_tokens.get(snapshot.uri) == token:
|
||||
self._analysis_tokens.pop(snapshot.uri, None)
|
||||
self._analysis_timers.pop(snapshot.uri, None)
|
||||
|
||||
timer = threading.Timer(delay_seconds, analyze)
|
||||
timer.daemon = True
|
||||
self._analysis_timers[snapshot.uri] = timer
|
||||
timer.start()
|
||||
|
||||
def analyze_document_now(self, document: TextDocument) -> None:
|
||||
"""Cancel delayed work and synchronously analyze the current document."""
|
||||
self._cancel_document_analysis(document.uri)
|
||||
self._invalidate_document_work(document)
|
||||
diagnostic_state = self.diagnostic_snapshot(document.uri)
|
||||
if diagnostic_state is None or diagnostic_state[0] != document.version:
|
||||
self.compute_diagnostics(document)
|
||||
if not self.index_is_current(document):
|
||||
self.update_poco_completion_for_file(document)
|
||||
|
||||
def index_snapshot(self) -> tuple[dict, dict, dict]:
|
||||
"""Return stable copies for request handlers running beside the indexer."""
|
||||
with self._index_lock:
|
||||
@@ -148,6 +444,9 @@ class TclLanguageServer(server.LanguageServer):
|
||||
self.proc_signatures.pop(filepath, None)
|
||||
self.proc_docs.pop(filepath, None)
|
||||
self.navigation_indexes.pop(filepath, None)
|
||||
self.variable_indexes.pop(filepath, None)
|
||||
self._committed_index_versions.pop(filepath, None)
|
||||
self._invalidate_workspace_caches_locked()
|
||||
|
||||
def indexed_paths_under_uri(self, uri: str) -> list[pathlib.Path]:
|
||||
target = pathlib.Path(uris.to_fs_path(uri))
|
||||
@@ -156,6 +455,7 @@ class TclLanguageServer(server.LanguageServer):
|
||||
indexed_paths.update(self.proc_signatures)
|
||||
indexed_paths.update(self.proc_docs)
|
||||
indexed_paths.update(self.navigation_indexes)
|
||||
indexed_paths.update(self.variable_indexes)
|
||||
indexed_paths.update(self._index_tokens)
|
||||
return [
|
||||
pathlib.Path(path)
|
||||
@@ -165,20 +465,28 @@ class TclLanguageServer(server.LanguageServer):
|
||||
|
||||
def remove_file_state(self, uri: str) -> None:
|
||||
"""Remove cached and indexed state for a file or a complete folder."""
|
||||
self.cancel_analysis_under_uri(uri)
|
||||
target = pathlib.Path(uris.to_fs_path(uri))
|
||||
|
||||
with self._index_lock:
|
||||
index_changed = False
|
||||
for store in (
|
||||
self.poco_completion,
|
||||
self.proc_signatures,
|
||||
self.proc_docs,
|
||||
self.navigation_indexes,
|
||||
self.variable_indexes,
|
||||
self._index_tokens,
|
||||
self._index_versions,
|
||||
self._committed_index_versions,
|
||||
):
|
||||
for path in list(store):
|
||||
if self._is_same_or_child(path, target):
|
||||
del store[path]
|
||||
index_changed = True
|
||||
|
||||
if index_changed:
|
||||
self._invalidate_workspace_caches_locked()
|
||||
|
||||
diagnostic_uris = set(self.diagnostics)
|
||||
diagnostic_uris.update(self._diagnostic_tokens)
|
||||
@@ -192,13 +500,16 @@ class TclLanguageServer(server.LanguageServer):
|
||||
self._diagnostic_tokens.pop(diagnostic_uri, None)
|
||||
|
||||
with self._parser_lock:
|
||||
for key in list(self._ast_cache):
|
||||
cached_keys = set(self._ast_cache)
|
||||
cached_keys.update(self._line_cache)
|
||||
for key in cached_keys:
|
||||
try:
|
||||
cached_path = pathlib.Path(uris.to_fs_path(key[0]))
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
if self._is_same_or_child(cached_path, target):
|
||||
del self._ast_cache[key]
|
||||
self._ast_cache.pop(key, None)
|
||||
self._line_cache.pop(key, None)
|
||||
|
||||
def update_poco_completion_for_file(
|
||||
self,
|
||||
@@ -225,8 +536,9 @@ class TclLanguageServer(server.LanguageServer):
|
||||
navigation_index = build_file_symbol_index(
|
||||
filepath, document.uri, tree
|
||||
)
|
||||
variable_index = build_variable_index(document.source, tree)
|
||||
except Exception as e:
|
||||
logging.debug(f"Error parsing {filepath}: {e}")
|
||||
LOGGER.debug("Error parsing %s: %s", filepath, e)
|
||||
self._discard_index_update(filepath, token)
|
||||
return False
|
||||
|
||||
@@ -241,6 +553,9 @@ class TclLanguageServer(server.LanguageServer):
|
||||
self.proc_signatures[filepath] = dict(collector.proc_signatures)
|
||||
self.proc_docs[filepath] = docs
|
||||
self.navigation_indexes[filepath] = navigation_index
|
||||
self.variable_indexes[filepath] = (document.version, variable_index)
|
||||
self._committed_index_versions[filepath] = document.version
|
||||
self._invalidate_workspace_caches_locked()
|
||||
return True
|
||||
|
||||
def format(
|
||||
|
||||
@@ -1,8 +1,9 @@
|
||||
from tclint.syntax_tree import Visitor, Command, BareWord, List
|
||||
import lsprotocol.types as lsp
|
||||
from common.load_data import standard_items
|
||||
from tclint.syntax_tree import BareWord, Command, List, Visitor
|
||||
|
||||
BUILTIN_VAR_LABELS = {ci.label for ci in standard_items.nx_variables}
|
||||
BUILTIN_PROC_LABELS = {ci.label for ci in standard_items.nx_procs}
|
||||
|
||||
|
||||
class CompletionItems:
|
||||
@@ -22,6 +23,7 @@ class CompletionCollector(Visitor):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self._custom_functions: list[lsp.CompletionItem] = []
|
||||
self._custom_function_keys: set[tuple[str, lsp.CompletionItemKind | None]] = set()
|
||||
self._proc_signatures = {}
|
||||
|
||||
@property
|
||||
@@ -34,8 +36,11 @@ class CompletionCollector(Visitor):
|
||||
|
||||
def _append_unique(self, item: lsp.CompletionItem):
|
||||
# Avoid duplicate labels within the same file scan
|
||||
if not any(ci.label == item.label for ci in self._custom_functions):
|
||||
self._custom_functions.append(item)
|
||||
key = (item.label, item.kind)
|
||||
if key in self._custom_function_keys:
|
||||
return
|
||||
self._custom_function_keys.add(key)
|
||||
self._custom_functions.append(item)
|
||||
|
||||
def visit_command(self, command: Command):
|
||||
routine = command.routine
|
||||
@@ -46,7 +51,7 @@ class CompletionCollector(Visitor):
|
||||
if not getattr(first_arg, "value", None):
|
||||
return
|
||||
|
||||
if any(item.label == first_arg.value for item in standard_items.nx_procs):
|
||||
if first_arg.value in BUILTIN_PROC_LABELS:
|
||||
return
|
||||
|
||||
# Record proc name as a completion item
|
||||
|
||||
+252
-16
@@ -1,29 +1,265 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
from collections.abc import Mapping, Sequence
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
import lsprotocol.types as lsp
|
||||
from tclint.syntax_tree import Visitor, Command
|
||||
from tclint.syntax_tree import Command, VarSub, Visitor
|
||||
from tools.navigation import FileSymbolIndex
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class InlayHintParameter:
|
||||
name: str
|
||||
documentation: str | None = None
|
||||
variadic: bool = False
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class InlayHintSignature:
|
||||
parameters: tuple[InlayHintParameter, ...]
|
||||
display_label: str
|
||||
documentation: str | None = None
|
||||
location: lsp.Location | None = None
|
||||
|
||||
|
||||
def _normalized_path(path: str) -> str:
|
||||
return os.path.normcase(os.path.abspath(path))
|
||||
|
||||
|
||||
def _definition_locations(
|
||||
index: FileSymbolIndex | None,
|
||||
) -> dict[str, lsp.Location]:
|
||||
if index is None:
|
||||
return {}
|
||||
|
||||
locations = {}
|
||||
for occurrence in index.occurrences:
|
||||
if (
|
||||
occurrence.is_definition
|
||||
and occurrence.identity.kind == "proc"
|
||||
):
|
||||
locations[occurrence.placeholder] = lsp.Location(
|
||||
uri=index.uri, range=occurrence.range
|
||||
)
|
||||
return locations
|
||||
|
||||
|
||||
def build_custom_inlay_signatures(
|
||||
signatures_by_path: dict[str, dict[str, list[str]]],
|
||||
docs_by_path: dict[str, dict[str, str]],
|
||||
indexes_by_path: dict[str, FileSymbolIndex],
|
||||
current_path: str,
|
||||
) -> dict[str, InlayHintSignature]:
|
||||
"""Merge workspace signatures deterministically, preferring the current file."""
|
||||
current_normalized = _normalized_path(current_path)
|
||||
paths = sorted(
|
||||
signatures_by_path, key=lambda path: _normalized_path(path).casefold()
|
||||
)
|
||||
paths.sort(key=lambda path: _normalized_path(path) == current_normalized)
|
||||
|
||||
result: dict[str, InlayHintSignature] = {}
|
||||
for path in paths:
|
||||
docs = docs_by_path.get(path, {})
|
||||
definition_locations = _definition_locations(indexes_by_path.get(path))
|
||||
for proc_name, parameter_names in signatures_by_path[path].items():
|
||||
parameters = tuple(
|
||||
InlayHintParameter(
|
||||
name=parameter_name,
|
||||
variadic=(
|
||||
parameter_name == "args"
|
||||
and parameter_index == len(parameter_names) - 1
|
||||
),
|
||||
)
|
||||
for parameter_index, parameter_name in enumerate(parameter_names)
|
||||
)
|
||||
result[proc_name] = InlayHintSignature(
|
||||
parameters=parameters,
|
||||
display_label=" ".join([proc_name, *parameter_names]),
|
||||
documentation=docs.get(proc_name),
|
||||
location=definition_locations.get(
|
||||
proc_name.removeprefix("::").rsplit("::", 1)[-1]
|
||||
),
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
def _is_builtin_variadic(item: dict[str, Any], parameter_name: str, index: int) -> bool:
|
||||
parameters = item.get("parameters", [])
|
||||
if index != len(parameters) - 1:
|
||||
return False
|
||||
if "..." in parameter_name or "…" in parameter_name:
|
||||
return True
|
||||
|
||||
format_label = item.get("format", "")
|
||||
return (
|
||||
f"<{parameter_name}>+" in format_label or f"[{parameter_name}]+" in format_label
|
||||
)
|
||||
|
||||
|
||||
def _builtin_parameter_label(parameter_name: str) -> str:
|
||||
if "..." not in parameter_name and "…" not in parameter_name:
|
||||
return parameter_name.strip("<>[]")
|
||||
|
||||
first_name = parameter_name.split()[0].strip("<>[]")
|
||||
return re.sub(r"(?:_?1)$", "", first_name) or first_name
|
||||
|
||||
|
||||
def build_builtin_inlay_signatures(
|
||||
items: list[dict[str, Any]],
|
||||
) -> dict[str, InlayHintSignature]:
|
||||
result = {}
|
||||
for item in items:
|
||||
proc_name = item.get("label")
|
||||
if not proc_name:
|
||||
continue
|
||||
parameters = tuple(
|
||||
InlayHintParameter(
|
||||
name=_builtin_parameter_label(parameter.get("name", "")),
|
||||
documentation=parameter.get("desc") or None,
|
||||
variadic=_is_builtin_variadic(
|
||||
item, parameter.get("name", ""), parameter_index
|
||||
),
|
||||
)
|
||||
for parameter_index, parameter in enumerate(item.get("parameters", []))
|
||||
if parameter.get("name")
|
||||
)
|
||||
result[proc_name] = InlayHintSignature(
|
||||
parameters=parameters,
|
||||
display_label=item.get("format") or proc_name,
|
||||
documentation=item.get("description") or None,
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
def _position_in_range(
|
||||
position: lsp.Position, requested_range: lsp.Range | None
|
||||
) -> bool:
|
||||
if requested_range is None:
|
||||
return True
|
||||
value = (position.line, position.character)
|
||||
start = (requested_range.start.line, requested_range.start.character)
|
||||
end = (requested_range.end.line, requested_range.end.character)
|
||||
return start <= value < end
|
||||
|
||||
|
||||
class InlayHintGenerator(Visitor):
|
||||
def __init__(self, proc_signatures):
|
||||
def __init__(
|
||||
self,
|
||||
source: str,
|
||||
proc_signatures: Mapping[str, InlayHintSignature],
|
||||
*,
|
||||
source_lines: Sequence[str] | None = None,
|
||||
requested_range: lsp.Range | None = None,
|
||||
parameter_names: str = "all",
|
||||
suppress_when_argument_matches_name: bool = True,
|
||||
):
|
||||
self.source_lines = (
|
||||
source_lines if source_lines is not None else source.splitlines()
|
||||
)
|
||||
self.proc_signatures = proc_signatures
|
||||
self.hints = []
|
||||
self.requested_range = requested_range
|
||||
self.parameter_names = parameter_names
|
||||
self.suppress_when_argument_matches_name = suppress_when_argument_matches_name
|
||||
self.hints: list[lsp.InlayHint] = []
|
||||
|
||||
def _node_intersects_requested_range(self, node) -> bool:
|
||||
if self.requested_range is None:
|
||||
return True
|
||||
start = getattr(node, "pos", None)
|
||||
end = getattr(node, "end_pos", None)
|
||||
if start is None or end is None:
|
||||
return True
|
||||
|
||||
node_start_line = start[0] - 1
|
||||
node_end_line = end[0] - 1
|
||||
return not (
|
||||
node_end_line < self.requested_range.start.line
|
||||
or node_start_line > self.requested_range.end.line
|
||||
)
|
||||
|
||||
def generate(self, tree) -> list[lsp.InlayHint]:
|
||||
"""Walk only syntax-tree branches overlapping the requested editor range."""
|
||||
self.hints.clear()
|
||||
|
||||
def walk(node) -> None:
|
||||
if not self._node_intersects_requested_range(node):
|
||||
return
|
||||
if isinstance(node, Command):
|
||||
self.visit_command(node)
|
||||
for child in getattr(node, "children", []):
|
||||
walk(child)
|
||||
|
||||
walk(tree)
|
||||
return self.hints
|
||||
|
||||
def _position(self, line: int, column: int) -> lsp.Position:
|
||||
line_index = line - 1
|
||||
character_index = column - 1
|
||||
if 0 <= line_index < len(self.source_lines):
|
||||
prefix = self.source_lines[line_index][:character_index]
|
||||
character_index = len(prefix.encode("utf-16-le")) // 2
|
||||
return lsp.Position(line=line_index, character=character_index)
|
||||
|
||||
@staticmethod
|
||||
def _parameter_for_argument(
|
||||
signature: InlayHintSignature, argument_index: int
|
||||
) -> InlayHintParameter | None:
|
||||
if argument_index < len(signature.parameters):
|
||||
return signature.parameters[argument_index]
|
||||
if signature.parameters and signature.parameters[-1].variadic:
|
||||
return signature.parameters[-1]
|
||||
return None
|
||||
|
||||
def _should_show(self, argument, parameter: InlayHintParameter) -> bool:
|
||||
if self.parameter_names == "none":
|
||||
return False
|
||||
if self.parameter_names == "literals" and isinstance(argument, VarSub):
|
||||
return False
|
||||
if not self.suppress_when_argument_matches_name or not isinstance(
|
||||
argument, VarSub
|
||||
):
|
||||
return True
|
||||
return getattr(argument, "value", None) != parameter.name
|
||||
|
||||
@staticmethod
|
||||
def _tooltip(signature: InlayHintSignature) -> lsp.MarkupContent:
|
||||
value = f"`{signature.display_label}`"
|
||||
if signature.documentation:
|
||||
value += f"\n\n{signature.documentation}"
|
||||
return lsp.MarkupContent(kind=lsp.MarkupKind.Markdown, value=value)
|
||||
|
||||
def visit_command(self, command: Command):
|
||||
name = getattr(command.routine, "contents", None)
|
||||
if name not in self.proc_signatures:
|
||||
signature = self.proc_signatures.get(name)
|
||||
if signature is None or self.parameter_names == "none":
|
||||
return
|
||||
|
||||
param_names = self.proc_signatures[name]
|
||||
for idx, arg in enumerate(command.args):
|
||||
if idx >= len(param_names):
|
||||
for argument_index, argument in enumerate(command.args):
|
||||
parameter = self._parameter_for_argument(signature, argument_index)
|
||||
if parameter is None:
|
||||
break
|
||||
param_name = param_names[idx]
|
||||
if not argument.pos or not self._should_show(argument, parameter):
|
||||
continue
|
||||
|
||||
if arg.pos:
|
||||
line, col = arg.pos
|
||||
self.hints.append(
|
||||
lsp.InlayHint(
|
||||
position=lsp.Position(line=line - 1, character=col - 1),
|
||||
label=f"{param_name}:",
|
||||
kind=lsp.InlayHintKind.Parameter,
|
||||
)
|
||||
line, column = argument.pos
|
||||
position = self._position(line, column)
|
||||
if not _position_in_range(position, self.requested_range):
|
||||
continue
|
||||
|
||||
label = lsp.InlayHintLabelPart(
|
||||
value=f"{parameter.name}:",
|
||||
tooltip=parameter.documentation,
|
||||
location=signature.location,
|
||||
)
|
||||
self.hints.append(
|
||||
lsp.InlayHint(
|
||||
position=position,
|
||||
label=[label],
|
||||
kind=lsp.InlayHintKind.Parameter,
|
||||
tooltip=self._tooltip(signature),
|
||||
)
|
||||
)
|
||||
|
||||
@@ -1,8 +1,9 @@
|
||||
import re
|
||||
from typing import Dict, List
|
||||
|
||||
from tclint.syntax_tree import Visitor, Command
|
||||
from tools.parser import CustomParser
|
||||
from tclint.syntax_tree import Command, Visitor
|
||||
|
||||
PROC_DECLARATION_RE = re.compile(r"^\s*proc\s+([^\s\{]+)")
|
||||
|
||||
|
||||
def _strip_comment_prefix(line: str) -> str:
|
||||
@@ -135,43 +136,18 @@ def build_proc_docs(tree, source_text: str) -> Dict[str, str]:
|
||||
return extractor.docs
|
||||
|
||||
|
||||
def is_proc_declaration_position(source_text: str, line_zero_based: int, char_zero_based: int) -> bool:
|
||||
def is_proc_declaration_line(line: str, char_zero_based: int) -> bool:
|
||||
"""Return True if the position is on the proc name on this source line."""
|
||||
match = PROC_DECLARATION_RE.match(line)
|
||||
return bool(match and match.start(1) <= char_zero_based <= match.end(1))
|
||||
|
||||
|
||||
def is_proc_declaration_position(
|
||||
source_text: str, line_zero_based: int, char_zero_based: int
|
||||
) -> bool:
|
||||
"""Return True if the position is on a proc name within its declaration."""
|
||||
parser = CustomParser()
|
||||
tree = parser.parse(source_text)
|
||||
|
||||
# Walk commands to find 'proc' declarations and check if position intersects the name arg
|
||||
class _DeclFinder(Visitor):
|
||||
def __init__(self):
|
||||
self.is_decl = False
|
||||
|
||||
def visit_command(self, command: Command):
|
||||
if self.is_decl:
|
||||
return
|
||||
routine = getattr(command.routine, "contents", None)
|
||||
if routine != "proc" or not command.args:
|
||||
return
|
||||
name_node = command.args[0]
|
||||
if not hasattr(name_node, "pos"):
|
||||
return
|
||||
# Calculate range for the name token
|
||||
try:
|
||||
start_line, start_col = name_node.pos
|
||||
end_line, end_col = getattr(name_node, "end_pos", name_node.pos)
|
||||
except Exception:
|
||||
return
|
||||
if start_line - 1 == line_zero_based:
|
||||
length = 0
|
||||
if hasattr(name_node, "value") and name_node.value is not None:
|
||||
length = len(name_node.value)
|
||||
elif hasattr(name_node, "contents") and name_node.contents is not None:
|
||||
length = len(name_node.contents)
|
||||
if length:
|
||||
start_c = start_col - 1
|
||||
end_c = start_c + length
|
||||
if start_c <= char_zero_based <= end_c:
|
||||
self.is_decl = True
|
||||
|
||||
finder = _DeclFinder()
|
||||
tree.accept(finder, recurse=True)
|
||||
return finder.is_decl
|
||||
try:
|
||||
line = source_text.splitlines()[line_zero_based]
|
||||
except IndexError:
|
||||
return False
|
||||
return is_proc_declaration_line(line, char_zero_based)
|
||||
|
||||
@@ -1,17 +1,17 @@
|
||||
import enum
|
||||
from typing import List
|
||||
from tclint.syntax_tree import Visitor, QuotedWord, Command, BareWord
|
||||
from tclint.commands.plugins import PluginManager
|
||||
|
||||
import attrs
|
||||
from common.load_data import standard_items
|
||||
import lsprotocol.types as lsp
|
||||
|
||||
from tclint.commands.plugins import PluginManager
|
||||
from tclint.syntax_tree import BareWord, Command, QuotedWord, Visitor
|
||||
|
||||
# Constructing a PluginManager scans entry points, and get_commands() rebuilds
|
||||
# the builtin command set on every call. Semantic tokens are requested often, so
|
||||
# cache the manager and the resolved commands per plugin set.
|
||||
_PLUGIN_MANAGER = None
|
||||
_COMMANDS_CACHE = {}
|
||||
_STANDARD_PROC_NAMES = frozenset(item.label for item in standard_items.nx_procs)
|
||||
|
||||
|
||||
def _load_commands(plugins):
|
||||
@@ -60,13 +60,23 @@ TOKEN_TYPES = [
|
||||
"string",
|
||||
"parameter",
|
||||
]
|
||||
TOKEN_TYPE_INDEX = {}
|
||||
for _token_index, _token_name in enumerate(TOKEN_TYPES):
|
||||
TOKEN_TYPE_INDEX.setdefault(_token_name, _token_index)
|
||||
|
||||
|
||||
class _Highlighter(Visitor):
|
||||
def __init__(self, plugins, custom_functions: dict[str : list[lsp.CompletionItem]]):
|
||||
def __init__(self, plugins, custom_functions):
|
||||
self._commands = _load_commands(plugins)
|
||||
self._tokens = []
|
||||
self.custom_functions = custom_functions
|
||||
if isinstance(custom_functions, dict):
|
||||
self._custom_function_names = frozenset(
|
||||
item.label
|
||||
for items in custom_functions.values()
|
||||
for item in items
|
||||
)
|
||||
else:
|
||||
self._custom_function_names = frozenset(custom_functions)
|
||||
|
||||
def _append_token(self, position, length: int, tok_type: str, modifiers: List[TokenModifier] | None = None):
|
||||
if position is None or length <= 0:
|
||||
@@ -129,9 +139,7 @@ class _Highlighter(Visitor):
|
||||
# Highlight functions (custom or standard) when used as the routine
|
||||
name = getattr(routine, "contents", None)
|
||||
if name:
|
||||
in_custom = any(item.label == name for items in self.custom_functions.values() for item in items)
|
||||
in_standard = any(item.label == name for item in standard_items.nx_procs)
|
||||
if in_custom or in_standard:
|
||||
if name in self._custom_function_names or name in _STANDARD_PROC_NAMES:
|
||||
line, col = routine.contents_pos
|
||||
self._append_token((line - 1, col - 1), len(name), "function", [])
|
||||
|
||||
|
||||
@@ -3,17 +3,15 @@ from concurrent.futures import ThreadPoolExecutor
|
||||
from pathlib import Path
|
||||
from threading import Event
|
||||
|
||||
|
||||
THIS_DIR = Path(__file__).parent
|
||||
SRC_DIR = THIS_DIR.parent.parent / "src"
|
||||
if str(SRC_DIR) not in sys.path:
|
||||
sys.path.insert(0, str(SRC_DIR))
|
||||
|
||||
import lsprotocol.types as lsp # type: ignore
|
||||
from pygls.workspace.text_document import TextDocument
|
||||
|
||||
import lsp_server
|
||||
import lsprotocol.types as lsp # type: ignore
|
||||
from lsp_tclserver import TclLanguageServer
|
||||
from pygls.workspace.text_document import TextDocument
|
||||
|
||||
|
||||
def _server() -> TclLanguageServer:
|
||||
@@ -63,6 +61,7 @@ def test_close_replaces_unsaved_index_with_saved_file(tmp_path: Path, monkeypatc
|
||||
server = _server()
|
||||
server.update_poco_completion_for_file(document)
|
||||
server.get_tree(document)
|
||||
server.get_lines(document)
|
||||
monkeypatch.setattr(lsp_server, "LSP_SERVER", server)
|
||||
|
||||
lsp_server.did_close(
|
||||
@@ -75,6 +74,7 @@ def test_close_replaces_unsaved_index_with_saved_file(tmp_path: Path, monkeypatc
|
||||
assert "unsaved_proc" not in signatures[document.path]
|
||||
assert "saved_proc" in signatures[document.path]
|
||||
assert all(key[0] != document.uri for key in server._ast_cache)
|
||||
assert all(key[0] != document.uri for key in server._line_cache)
|
||||
|
||||
|
||||
def test_delete_and_rename_notifications_update_index(tmp_path: Path, monkeypatch):
|
||||
@@ -218,3 +218,52 @@ def test_delete_invalidates_in_flight_diagnostics(tmp_path: Path, monkeypatch):
|
||||
future.result(timeout=5)
|
||||
|
||||
assert server.diagnostic_snapshot(document.uri) is None
|
||||
|
||||
|
||||
def test_rapid_changes_analyze_only_latest_snapshot(tmp_path: Path, monkeypatch):
|
||||
server = _server()
|
||||
path = tmp_path / "debounced.tcl"
|
||||
first = _document(path, "set value 1", version=1)
|
||||
latest = _document(path, "set value 2", version=2)
|
||||
calls = []
|
||||
completed = Event()
|
||||
|
||||
monkeypatch.setattr(
|
||||
server,
|
||||
"compute_diagnostics",
|
||||
lambda document: calls.append(("diagnostics", document.version)),
|
||||
)
|
||||
|
||||
def record_index(document):
|
||||
calls.append(("index", document.version))
|
||||
completed.set()
|
||||
return True
|
||||
|
||||
monkeypatch.setattr(server, "update_poco_completion_for_file", record_index)
|
||||
|
||||
server.schedule_document_analysis(first, delay_seconds=0.05)
|
||||
server.schedule_document_analysis(latest, delay_seconds=0.05)
|
||||
|
||||
assert completed.wait(timeout=2)
|
||||
assert calls == [("diagnostics", 2), ("index", 2)]
|
||||
|
||||
|
||||
def test_variable_and_workspace_request_caches_are_reused(tmp_path: Path):
|
||||
server = _server()
|
||||
document = _document(
|
||||
tmp_path / "cached.tcl",
|
||||
"proc cached_proc {argument} { set local_value $argument }",
|
||||
)
|
||||
|
||||
assert server.update_poco_completion_for_file(document)
|
||||
first_variables = server.variable_index_for_document(document)
|
||||
second_variables = server.variable_index_for_document(document)
|
||||
first_completions = server.completion_items_snapshot()
|
||||
second_completions = server.completion_items_snapshot()
|
||||
first_names = server.custom_function_names_snapshot()
|
||||
second_names = server.custom_function_names_snapshot()
|
||||
|
||||
assert first_variables is second_variables
|
||||
assert first_completions is second_completions
|
||||
assert first_names is second_names
|
||||
assert "cached_proc" in first_names
|
||||
|
||||
@@ -0,0 +1,209 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
THIS_DIR = Path(__file__).parent
|
||||
SRC_DIR = THIS_DIR.parent.parent / "src"
|
||||
if str(SRC_DIR) not in sys.path:
|
||||
sys.path.insert(0, str(SRC_DIR))
|
||||
|
||||
import lsprotocol.types as lsp # type: ignore
|
||||
from tools.inlay_hint import (
|
||||
InlayHintGenerator,
|
||||
InlayHintParameter,
|
||||
InlayHintSignature,
|
||||
build_builtin_inlay_signatures,
|
||||
build_custom_inlay_signatures,
|
||||
)
|
||||
from tools.navigation import build_file_symbol_index
|
||||
from tools.parser import CustomParser
|
||||
|
||||
|
||||
def _signature(*names: str, variadic: bool = False) -> InlayHintSignature:
|
||||
parameters = tuple(
|
||||
InlayHintParameter(
|
||||
name=name,
|
||||
variadic=variadic and index == len(names) - 1,
|
||||
)
|
||||
for index, name in enumerate(names)
|
||||
)
|
||||
return InlayHintSignature(
|
||||
parameters=parameters,
|
||||
display_label=" ".join(["test_proc", *names]),
|
||||
)
|
||||
|
||||
|
||||
def _generate(
|
||||
source: str,
|
||||
signatures: dict[str, InlayHintSignature],
|
||||
**options,
|
||||
) -> list[lsp.InlayHint]:
|
||||
tree = CustomParser().parse(source)
|
||||
generator = InlayHintGenerator(source, signatures, **options)
|
||||
return generator.generate(tree)
|
||||
|
||||
|
||||
def _labels(hints: list[lsp.InlayHint]) -> list[str]:
|
||||
labels = []
|
||||
for hint in hints:
|
||||
if isinstance(hint.label, str):
|
||||
labels.append(hint.label)
|
||||
else:
|
||||
labels.append("".join(part.value for part in hint.label))
|
||||
return labels
|
||||
|
||||
|
||||
def test_many_parameters_are_returned_without_server_side_truncation():
|
||||
names = tuple(f"parameter_{index}" for index in range(10))
|
||||
source = "test_proc " + " ".join(str(index) for index in range(10))
|
||||
|
||||
hints = _generate(source, {"test_proc": _signature(*names)})
|
||||
|
||||
assert _labels(hints) == [f"{name}:" for name in names]
|
||||
assert all("…" not in label and "..." not in label for label in _labels(hints))
|
||||
|
||||
|
||||
def test_only_hints_inside_requested_range_are_returned():
|
||||
source = "test_proc first\nset spacer 1\ntest_proc second"
|
||||
requested_range = lsp.Range(
|
||||
start=lsp.Position(line=2, character=0),
|
||||
end=lsp.Position(line=3, character=0),
|
||||
)
|
||||
|
||||
hints = _generate(
|
||||
source,
|
||||
{"test_proc": _signature("value")},
|
||||
requested_range=requested_range,
|
||||
)
|
||||
|
||||
assert len(hints) == 1
|
||||
assert hints[0].position.line == 2
|
||||
|
||||
|
||||
def test_range_walk_keeps_nested_commands_inside_proc_body():
|
||||
source = "proc wrapper {} {\n test_proc nested\n}"
|
||||
requested_range = lsp.Range(
|
||||
start=lsp.Position(line=1, character=0),
|
||||
end=lsp.Position(line=2, character=0),
|
||||
)
|
||||
|
||||
hints = _generate(
|
||||
source,
|
||||
{"test_proc": _signature("value")},
|
||||
requested_range=requested_range,
|
||||
)
|
||||
|
||||
assert _labels(hints) == ["value:"]
|
||||
assert hints[0].position.line == 1
|
||||
|
||||
|
||||
def test_matching_variable_name_can_be_suppressed():
|
||||
source = "test_proc $value $other"
|
||||
signature = _signature("value", "result")
|
||||
|
||||
suppressed = _generate(source, {"test_proc": signature})
|
||||
visible = _generate(
|
||||
source,
|
||||
{"test_proc": signature},
|
||||
suppress_when_argument_matches_name=False,
|
||||
)
|
||||
|
||||
assert _labels(suppressed) == ["result:"]
|
||||
assert _labels(visible) == ["value:", "result:"]
|
||||
|
||||
|
||||
def test_literal_mode_hides_variable_argument_hints():
|
||||
source = 'test_proc $value "literal"'
|
||||
|
||||
hints = _generate(
|
||||
source,
|
||||
{"test_proc": _signature("first", "second")},
|
||||
parameter_names="literals",
|
||||
)
|
||||
|
||||
assert _labels(hints) == ["second:"]
|
||||
|
||||
|
||||
def test_variadic_parameter_labels_every_remaining_argument():
|
||||
hints = _generate(
|
||||
"test_proc first second third fourth",
|
||||
{"test_proc": _signature("required", "args", variadic=True)},
|
||||
)
|
||||
|
||||
assert _labels(hints) == ["required:", "args:", "args:", "args:"]
|
||||
|
||||
|
||||
def test_builtin_signature_has_variadic_hints_and_parameter_documentation():
|
||||
signatures = build_builtin_inlay_signatures(
|
||||
[
|
||||
{
|
||||
"label": "MOM_force",
|
||||
"description": "Controls address output.",
|
||||
"format": "MOM_force <mode> <address_1 ... address_n>",
|
||||
"parameters": [
|
||||
{"name": "mode", "desc": "Output mode."},
|
||||
{
|
||||
"name": "address_1 ... address_n",
|
||||
"desc": "Output addresses.",
|
||||
},
|
||||
],
|
||||
}
|
||||
]
|
||||
)
|
||||
|
||||
hints = _generate("MOM_force Always X Y", signatures)
|
||||
|
||||
assert _labels(hints) == ["mode:", "address:", "address:"]
|
||||
assert hints[1].label[0].tooltip == "Output addresses." # type: ignore[index]
|
||||
assert hints[0].tooltip.value.endswith("Controls address output.") # type: ignore[union-attr]
|
||||
|
||||
|
||||
def test_current_file_signature_and_definition_location_take_priority(tmp_path: Path):
|
||||
other_path = tmp_path / "other.tcl"
|
||||
current_path = tmp_path / "current.tcl"
|
||||
other_source = "proc shared {from_other} { return $from_other }"
|
||||
current_source = "proc shared {from_current args} { return $from_current }"
|
||||
parser = CustomParser()
|
||||
other_tree = parser.parse(other_source)
|
||||
current_tree = parser.parse(current_source)
|
||||
indexes = {
|
||||
str(other_path): build_file_symbol_index(
|
||||
str(other_path), other_path.as_uri(), other_tree
|
||||
),
|
||||
str(current_path): build_file_symbol_index(
|
||||
str(current_path), current_path.as_uri(), current_tree
|
||||
),
|
||||
}
|
||||
|
||||
signatures = build_custom_inlay_signatures(
|
||||
{
|
||||
str(current_path): {"shared": ["from_current", "args"]},
|
||||
str(other_path): {"shared": ["from_other"]},
|
||||
},
|
||||
{
|
||||
str(current_path): {"shared": "Current documentation."},
|
||||
str(other_path): {"shared": "Other documentation."},
|
||||
},
|
||||
indexes,
|
||||
str(current_path),
|
||||
)
|
||||
|
||||
signature = signatures["shared"]
|
||||
assert [parameter.name for parameter in signature.parameters] == [
|
||||
"from_current",
|
||||
"args",
|
||||
]
|
||||
assert signature.parameters[-1].variadic
|
||||
assert signature.documentation == "Current documentation."
|
||||
assert signature.location is not None
|
||||
assert signature.location.uri == current_path.as_uri()
|
||||
|
||||
|
||||
def test_positions_use_lsp_utf16_offsets():
|
||||
source = 'test_proc "😀" second'
|
||||
|
||||
hints = _generate(
|
||||
source,
|
||||
{"test_proc": _signature("first", "second")},
|
||||
)
|
||||
|
||||
assert hints[1].position.character == source.index("second") + 1
|
||||
Reference in New Issue
Block a user