Compare commits

...
5 Commits
Author SHA1 Message Date
Christoph 07ccd2d26a Update version to 2026.8.200 2026-08-19 11:51:21 +00:00
Christoph Brandau af195a577b refactor(lsp): cache analysis results and debounce diagnostics
build_and_puplish.yml / build_and_publish (release) Successful in 29s
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
Christoph Brandau ecb50be2b8 feat(inlay-hints): add configurable parameter name hints
Adds configurable inlay hints for TCL procedures and merges signatures from built-ins and workspace files. The feature supports parameterNames and suppressWhenArgumentMatchesName and respects an optional range filter and current-file priority.

- Introduces built-in and custom inlay hint builders
- Honors inlayHints parameterNames and suppression options
- Adds tests validating hints, ranges, and priority rules
2026-08-19 12:59:51 +02:00
Christoph 61d4785775 Update version to 2026.8.100 2026-08-19 07:47:31 +00:00
Christoph Brandau 541704f45a feat(cdl): add CDL event handler parsing and snippet support
build_and_puplish.yml / build_and_publish (release) Successful in 36s
This adds a small DSL-aware helper to extract CDL event
declarations and generate a ready-to-use snippet. It also
extends hover support to show the snippet and parameter
hints for CDL events, improving developer productivity.

- Generate a MOM event handler scaffold and local vars
- Hover shows the generated snippet and mom_ parameter hints
2026-08-19 09:45:57 +02:00
15 changed files with 1240 additions and 200 deletions
+5
View File
@@ -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]
+6
View File
@@ -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
+116
View File
@@ -0,0 +1,116 @@
export interface CdlEventHandler {
eventName: string
parameterNames: string[]
}
function structuralCode(line: string): string {
let result = ""
let inString = false
let escaped = false
for (const character of line) {
if (escaped) {
escaped = false
result += inString ? " " : character
continue
}
if (character === "\\") {
escaped = true
result += inString ? " " : character
continue
}
if (character === '"') {
inString = !inString
result += " "
continue
}
if (character === "#" && !inString) {
break
}
result += inString ? " " : character
}
return result
}
function braceDelta(line: string): number {
let delta = 0
for (const character of structuralCode(line)) {
if (character === "{") {
delta += 1
} else if (character === "}") {
delta -= 1
}
}
return delta
}
export function cdlEventHandlerAtLine(
source: string,
declarationLine: number
): CdlEventHandler | undefined {
const lines = source.split(/\r?\n/)
const declaration = lines[declarationLine]
if (declaration === undefined) {
return undefined
}
const eventMatch = /^\s*EVENT\s+([^\s{]+)/.exec(structuralCode(declaration))
if (!eventMatch) {
return undefined
}
const parameterNames: string[] = []
let eventOpened = false
let depth = 0
for (let lineNumber = declarationLine; lineNumber < lines.length; lineNumber++) {
const line = lines[lineNumber]
const code = structuralCode(line)
if (eventOpened && depth === 1) {
const parameterMatch = /^\s*PARAM\s+([^\s{]+)/.exec(code)
if (parameterMatch) {
parameterNames.push(parameterMatch[1])
}
}
const delta = braceDelta(line)
if (!eventOpened && delta > 0) {
eventOpened = true
}
if (eventOpened) {
depth += delta
if (depth <= 0) {
break
}
}
}
return {
eventName: eventMatch[1],
parameterNames
}
}
function momEventName(eventName: string): string {
return `MOM_${eventName.replace(/^MOM_/i, "")}`
}
function momVariableName(parameterName: string): string {
return `mom_${parameterName.replace(/^mom_/i, "")}`
}
export function createCdlEventHandlerSnippet(handler: CdlEventHandler): string {
const globals = [
...new Set(handler.parameterNames.map((parameter) => momVariableName(parameter)))
]
const lines = [`proc ${momEventName(handler.eventName)} { } {`]
if (globals.length > 0) {
lines.push(...globals.map((variable) => ` global ${variable}`), "")
}
lines.push(" #Put your UDE Handler Tcl here", "", "}")
return lines.join("\n")
}
+41 -27
View File
@@ -1,4 +1,10 @@
import * as vscode from "vscode"
import {
cdlEventHandlerAtLine,
createCdlEventHandlerSnippet
} from "./cdlEventHandler"
const MACHINE_HEADER_REGEX = /^MACHINE\s+\S+/
export function formatCdlFile(content: string): string {
let indentLevel = 0
@@ -37,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
}
@@ -53,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,
@@ -110,30 +121,33 @@ export function completionHandlerCdl(document: vscode.TextDocument, position: vs
}
export function hoverCdlHandler(document: vscode.TextDocument, position: vscode.Position) {
const wordRange = document.getWordRangeAtPosition(position)
const word = document.getText(wordRange)
const text = document.getText()
const lines = text.split("\n")
let hoverText: string | undefined
for (const line of lines) {
const words = line.split(/\s+/)
const wordIndex = words.indexOf(word)
if (wordIndex > 0) {
if (words[wordIndex - 1] === "EVENT") {
hoverText = `MOM_${word}`
} else if (words[wordIndex - 1] === "PARAM") {
hoverText = `mom_${word}`
const line = document.lineAt(position.line).text
const eventMatch = /^\s*EVENT\s+([^\s{]+)/.exec(line)
if (eventMatch) {
const eventStart = line.indexOf(eventMatch[1], eventMatch.index)
const declarationEnd = eventStart + eventMatch[1].length
if (position.character <= declarationEnd) {
const handler = cdlEventHandlerAtLine(document.getText(), position.line)
if (handler) {
const markdown = new vscode.MarkdownString()
markdown.appendCodeblock(createCdlEventHandlerSnippet(handler), "tcl")
return new vscode.Hover(
markdown,
new vscode.Range(position.line, eventMatch.index, position.line, declarationEnd)
)
}
break
}
}
if (hoverText) {
return new vscode.Hover(hoverText)
const parameterMatch = /^\s*PARAM\s+([^\s{]+)/.exec(line)
if (parameterMatch) {
const parameterStart = line.indexOf(parameterMatch[1], parameterMatch.index)
const parameterEnd = parameterStart + parameterMatch[1].length
if (position.character >= parameterStart && position.character <= parameterEnd) {
return new vscode.Hover(`mom_${parameterMatch[1].replace(/^mom_/i, "")}`)
}
}
return undefined
}
+23 -3
View File
@@ -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
View File
@@ -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
View File
@@ -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.200",
"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
View File
@@ -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
View File
@@ -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(
+9 -4
View File
@@ -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
View File
@@ -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),
)
)
+17 -41
View File
@@ -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)
+17 -9
View File
@@ -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