438 lines
14 KiB
Python
438 lines
14 KiB
Python
import argparse
|
|
import dataclasses
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Dict, List, Optional, Tuple
|
|
import uuid
|
|
|
|
from lsprotocol import types as lsp
|
|
|
|
from pygls.server import LanguageServer
|
|
from pygls.workspace import TextDocument
|
|
from pygls.uris import to_fs_path
|
|
|
|
from tclint.cli import tclint
|
|
from tclint.config import get_config, DEFAULT_CONFIGS, RunConfig, Config, ConfigError
|
|
from tclint.format import Formatter, FormatterOpts
|
|
from tclint.lexer import TclSyntaxError
|
|
from tclint.parser import Parser
|
|
from tclint.cli import utils
|
|
|
|
try:
|
|
from tclint._version import __version__ # type: ignore
|
|
except ModuleNotFoundError:
|
|
__version__ = "(unknown version)"
|
|
|
|
|
|
DIAGNOSTIC_SOURCE = "tclint"
|
|
|
|
|
|
def lint(source, config, path):
|
|
diagnostics = []
|
|
|
|
try:
|
|
violations = tclint.lint(source, config, path)
|
|
except TclSyntaxError as e:
|
|
return [
|
|
lsp.Diagnostic(
|
|
message=str(e),
|
|
severity=lsp.DiagnosticSeverity.Error,
|
|
range=lsp.Range(
|
|
start=lsp.Position(e.start[0] - 1, e.start[1] - 1),
|
|
end=lsp.Position(e.end[0] - 1, e.end[1] - 1),
|
|
),
|
|
code="syntax error",
|
|
source=DIAGNOSTIC_SOURCE,
|
|
)
|
|
]
|
|
|
|
for violation in violations:
|
|
message = violation.message
|
|
severity = lsp.DiagnosticSeverity.Warning
|
|
start = lsp.Position(
|
|
line=violation.start[0] - 1, character=violation.start[1] - 1
|
|
)
|
|
end = lsp.Position(line=violation.end[0] - 1, character=violation.end[1] - 1)
|
|
|
|
diagnostics.append(
|
|
lsp.Diagnostic(
|
|
message=message,
|
|
severity=severity,
|
|
range=lsp.Range(
|
|
start=start,
|
|
end=end,
|
|
),
|
|
code=violation.id,
|
|
source=DIAGNOSTIC_SOURCE,
|
|
)
|
|
)
|
|
|
|
return diagnostics
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class ExtensionSettings:
|
|
# This path is expected to be absolute.
|
|
config_file: Optional[Path] = dataclasses.field(default=None)
|
|
|
|
|
|
class TclspServer(LanguageServer):
|
|
"""Main server class. Implements pull diagnostics using a method adapted from
|
|
https://pygls.readthedocs.io/en/latest/examples/pull-diagnostics.html."""
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self.diagnostics = {}
|
|
self.global_config: RunConfig = None
|
|
# Maps workspace roots to configs.
|
|
self.configs: Dict[Path, RunConfig] = {}
|
|
self.client_supports_refresh = False
|
|
|
|
self.global_settings = ExtensionSettings()
|
|
self.workspace_settings: Dict[Path, ExtensionSettings] = {}
|
|
|
|
def get_roots(self) -> List[Path]:
|
|
"""Returns root folders currently open in the workspace."""
|
|
roots = []
|
|
for uri in self.workspace.folders.keys():
|
|
path = to_fs_path(uri)
|
|
if path is not None:
|
|
roots.append(Path(path))
|
|
|
|
if len(roots) > 0:
|
|
return roots
|
|
|
|
if self.workspace.root_path is not None:
|
|
roots.append(Path(self.workspace.root_path))
|
|
|
|
return roots
|
|
|
|
def get_root(self, path: Path) -> Optional[Path]:
|
|
"""Returns workspace root folder that's closest to path.
|
|
|
|
Returns None if path is not in a workspace folder or if there are no workspace
|
|
folders.
|
|
"""
|
|
roots = self.get_roots()
|
|
closest_root = None
|
|
distance = float("inf")
|
|
for root in roots:
|
|
try:
|
|
relpath = path.relative_to(root)
|
|
except ValueError:
|
|
continue
|
|
if len(relpath.parts) < distance:
|
|
distance = len(relpath.parts)
|
|
closest_root = root
|
|
return closest_root
|
|
|
|
def get_config_file(self, workspace_root: Path) -> Optional[Path]:
|
|
if workspace_root in self.workspace_settings:
|
|
settings = self.workspace_settings[workspace_root]
|
|
return settings.config_file
|
|
return self.global_settings.config_file
|
|
|
|
def load_configs(self):
|
|
self.configs = {}
|
|
for root in self.get_roots():
|
|
try:
|
|
path = self.get_config_file(root)
|
|
config = get_config(path, root)
|
|
if config is not None:
|
|
self.configs[root] = config
|
|
except ConfigError as e:
|
|
self.show_message(f"Error loading config file: {e}")
|
|
|
|
# If a global config file exists, we apply it to any file not under a workspace
|
|
# folder.
|
|
global_path = self.global_settings.config_file
|
|
if global_path is not None:
|
|
try:
|
|
config = get_config(global_path, global_path.parent)
|
|
self.global_config = config
|
|
except ConfigError as e:
|
|
self.show_message(f"Error loading config file: {e}")
|
|
|
|
def get_config(self, path: Path, root: Optional[Path]) -> Config:
|
|
if root in self.configs:
|
|
return self.configs[root].get_for_path(path)
|
|
if self.global_config is not None:
|
|
return self.global_config.get_for_path(path)
|
|
return Config()
|
|
|
|
def _compute_diagnostics(self, document: TextDocument) -> List[lsp.Diagnostic]:
|
|
path = Path(document.path)
|
|
root = self.get_root(path)
|
|
config = self.get_config(path, root)
|
|
|
|
if root is None:
|
|
root = path.parent
|
|
|
|
is_excluded = utils.make_exclude_filter(config.exclude)
|
|
if is_excluded(path, root):
|
|
return []
|
|
|
|
return lint(document.source, config, path)
|
|
|
|
def compute_diagnostics(self, document: TextDocument):
|
|
# `None` sentinel ensures that `diagnostics` gets updated if the URI is not
|
|
# present.
|
|
_, previous = self.diagnostics.get(document.uri, (0, None))
|
|
|
|
diagnostics = self._compute_diagnostics(document)
|
|
|
|
# Only update if the list has changed
|
|
if previous != diagnostics:
|
|
self.diagnostics[document.uri] = (document.version, diagnostics)
|
|
|
|
def format(
|
|
self,
|
|
document: TextDocument,
|
|
options: lsp.FormattingOptions,
|
|
range: Optional[Tuple[int, int]] = None,
|
|
):
|
|
path = Path(document.path)
|
|
root = self.get_root(path)
|
|
config = self.get_config(path, root)
|
|
|
|
parser = Parser()
|
|
|
|
if config.style_indent is None:
|
|
indent = "\t" if not options.insert_spaces else " " * options.tab_size
|
|
else:
|
|
indent = config.get_indent()
|
|
|
|
formatter = Formatter(
|
|
FormatterOpts(
|
|
indent=indent,
|
|
spaces_in_braces=config.style_spaces_in_braces,
|
|
max_blank_lines=config.style_max_blank_lines,
|
|
indent_namespace_eval=config.style_indent_namespace_eval,
|
|
)
|
|
)
|
|
|
|
if range is not None:
|
|
start, end = range
|
|
return formatter.format_partial(document.source[start:end], parser)
|
|
|
|
return formatter.format_top(document.source, parser)
|
|
|
|
|
|
server = TclspServer("tclsp", __version__)
|
|
|
|
|
|
@server.feature(lsp.TEXT_DOCUMENT_DID_OPEN)
|
|
def did_open(ls: TclspServer, params: lsp.DidOpenTextDocumentParams):
|
|
"""Parse each document when it is opened"""
|
|
logging.debug("Received %s: %s", lsp.TEXT_DOCUMENT_DID_OPEN, params)
|
|
doc = ls.workspace.get_text_document(params.text_document.uri)
|
|
ls.compute_diagnostics(doc)
|
|
|
|
|
|
@server.feature(lsp.TEXT_DOCUMENT_DID_CHANGE)
|
|
def did_change(ls: TclspServer, params: lsp.DidOpenTextDocumentParams):
|
|
"""Parse each document when it is changed"""
|
|
logging.debug("Received %s: %s", lsp.TEXT_DOCUMENT_DID_CHANGE, params)
|
|
doc = ls.workspace.get_text_document(params.text_document.uri)
|
|
ls.compute_diagnostics(doc)
|
|
|
|
|
|
@server.feature(
|
|
lsp.TEXT_DOCUMENT_DIAGNOSTIC,
|
|
lsp.DiagnosticOptions(
|
|
identifier="pull-diagnostics",
|
|
inter_file_dependencies=False,
|
|
# We could support workspace diagnostics, although an implementation based on
|
|
# the pygls tutorial seems to add client-server noise for no benefit (it ends up
|
|
# replying to a frequent workspace diagnostics request with "unchanged"
|
|
# messages).
|
|
workspace_diagnostics=False,
|
|
),
|
|
)
|
|
def document_diagnostic(ls: TclspServer, params: lsp.DocumentDiagnosticParams):
|
|
"""Return diagnostics for the requested document"""
|
|
logging.debug("Received %s: %s", lsp.TEXT_DOCUMENT_DIAGNOSTIC, params)
|
|
|
|
was_cached = True
|
|
if (uri := params.text_document.uri) not in ls.diagnostics:
|
|
was_cached = False
|
|
doc = ls.workspace.get_text_document(uri)
|
|
ls.compute_diagnostics(doc)
|
|
|
|
version, diagnostics = ls.diagnostics[uri]
|
|
result_id = f"{uri}@{version}"
|
|
|
|
if was_cached and result_id == params.previous_result_id:
|
|
return lsp.UnchangedDocumentDiagnosticReport(result_id)
|
|
|
|
return lsp.FullDocumentDiagnosticReport(items=diagnostics, result_id=result_id)
|
|
|
|
|
|
@server.feature(lsp.WORKSPACE_DID_CHANGE_WATCHED_FILES)
|
|
def change_watched_files(ls: TclspServer, params: lsp.DidChangeWatchedFilesParams):
|
|
logging.debug("Received %s: %s", lsp.WORKSPACE_DID_CHANGE_WATCHED_FILES, params)
|
|
|
|
# Clear diagnostics cache so they get recalculated when requested
|
|
ls.diagnostics = {}
|
|
|
|
ls.load_configs()
|
|
if ls.client_supports_refresh:
|
|
ls.lsp.send_request(lsp.WORKSPACE_DIAGNOSTIC_REFRESH, None)
|
|
|
|
|
|
@server.feature(lsp.TEXT_DOCUMENT_FORMATTING)
|
|
def format_document(ls: TclspServer, params: lsp.DocumentFormattingParams):
|
|
"""Format the entire document"""
|
|
doc = ls.workspace.get_text_document(params.text_document.uri)
|
|
|
|
source = doc.source
|
|
start = lsp.Position(line=0, character=0)
|
|
last_line = source.rsplit("\n", 1)[-1]
|
|
end = lsp.Position(line=source.count("\n"), character=len(last_line))
|
|
|
|
formatted = ls.format(doc, params.options)
|
|
return [
|
|
lsp.TextEdit(
|
|
range=lsp.Range(start=start, end=end),
|
|
new_text=formatted,
|
|
)
|
|
]
|
|
|
|
|
|
@server.feature(lsp.TEXT_DOCUMENT_RANGE_FORMATTING)
|
|
def format_range(ls: TclspServer, params: lsp.DocumentRangeFormattingParams):
|
|
"""Format the given range with a document"""
|
|
doc = ls.workspace.get_text_document(params.text_document.uri)
|
|
|
|
# Round up range to full lines.
|
|
start_line = params.range.start.line
|
|
end_line = params.range.end.line
|
|
if params.range.end.character > 0:
|
|
end_line += 1
|
|
range = lsp.Range(
|
|
start=lsp.Position(line=start_line, character=0),
|
|
end=lsp.Position(line=end_line, character=0),
|
|
)
|
|
|
|
start = doc.offset_at_position(range.start)
|
|
end = doc.offset_at_position(range.end)
|
|
|
|
try:
|
|
formatted = ls.format(doc, params.options, range=(start, end))
|
|
except TclSyntaxError:
|
|
return None
|
|
|
|
return [
|
|
lsp.TextEdit(
|
|
range=range,
|
|
new_text=formatted,
|
|
)
|
|
]
|
|
|
|
|
|
@server.feature(lsp.INITIALIZE)
|
|
def initialize(ls: TclspServer, params: lsp.InitializeParams) -> None:
|
|
if params.initialization_options is None:
|
|
return
|
|
|
|
# Apply settings provided on initialization. The schema was copied from the template
|
|
# that the tclint-vscode extension is based on.
|
|
globalSettings = params.initialization_options.get("globalSettings", {})
|
|
if globalSettings.get("configPath"):
|
|
path = Path(globalSettings["configPath"]).expanduser()
|
|
if not path.is_absolute():
|
|
ls.show_message(
|
|
f"Warning: expected global config path to be absolute, got {path}"
|
|
)
|
|
else:
|
|
ls.global_settings.config_file = path
|
|
|
|
for settings in params.initialization_options.get("settings", []):
|
|
root = Path(settings["cwd"])
|
|
if root not in ls.workspace_settings:
|
|
ls.workspace_settings[root] = ExtensionSettings()
|
|
if settings.get("configPath"):
|
|
path = Path(settings["configPath"]).expanduser()
|
|
if not path.is_absolute():
|
|
path = root / path
|
|
ls.workspace_settings[root].config_file = path
|
|
|
|
|
|
@server.feature(lsp.INITIALIZED)
|
|
def init(ls: TclspServer, params: lsp.InitializeParams):
|
|
"""Registers file watchers on config filenames so that we can reload configs and
|
|
refresh diagnostics if they've changed.
|
|
|
|
Based on code snippet in
|
|
https://github.com/openlawlibrary/pygls/issues/376#issuecomment-1717656614.
|
|
"""
|
|
capabilities = ls.client_capabilities.workspace
|
|
|
|
try:
|
|
ls.client_supports_refresh = (
|
|
capabilities.diagnostics.refresh_support # type: ignore[union-attr]
|
|
)
|
|
except AttributeError:
|
|
ls.client_supports_refresh = False
|
|
|
|
try:
|
|
client_supports_watched_files_registration = (
|
|
capabilities.did_change_watched_files.dynamic_registration # type: ignore[union-attr] # noqa: E501
|
|
)
|
|
except AttributeError:
|
|
client_supports_watched_files_registration = False
|
|
|
|
if client_supports_watched_files_registration:
|
|
watchers = []
|
|
for filename in (*DEFAULT_CONFIGS, "pyproject.toml"):
|
|
pattern = f"**/{filename}"
|
|
watchers.append(lsp.FileSystemWatcher(glob_pattern=pattern))
|
|
|
|
for settings in (ls.global_settings, *ls.workspace_settings.values()):
|
|
if settings.config_file is not None:
|
|
watchers.append(
|
|
lsp.FileSystemWatcher(glob_pattern=settings.config_file)
|
|
)
|
|
|
|
ls.register_capability(
|
|
lsp.RegistrationParams(
|
|
registrations=[
|
|
lsp.Registration(
|
|
id=str(uuid.uuid4()),
|
|
method=lsp.WORKSPACE_DID_CHANGE_WATCHED_FILES,
|
|
register_options=lsp.DidChangeWatchedFilesRegistrationOptions(
|
|
watchers=watchers
|
|
),
|
|
)
|
|
]
|
|
)
|
|
)
|
|
|
|
ls.load_configs()
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser("tclsp")
|
|
log_levels = {
|
|
"debug": logging.DEBUG,
|
|
"info": logging.INFO,
|
|
"warning": logging.WARNING,
|
|
"error": logging.ERROR,
|
|
}
|
|
parser.add_argument(
|
|
"-l",
|
|
"--log-level",
|
|
default="info",
|
|
type=lambda x: x.lower(),
|
|
help="set the log level. defaults to info",
|
|
choices=log_levels.keys(),
|
|
)
|
|
args = parser.parse_args()
|
|
logging.basicConfig(level=log_levels[args.log_level], format="%(message)s")
|
|
|
|
server.start_io()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|