vibefpmgui.lsp
A small client for the Fortran language server, fortls.
fortls indexes the project's Fortran
sources and answers questions about them: what a name is (hover), where it
is defined, the outline of a file and the symbols in the whole project. It
runs as a background process speaking the Language Server Protocol (JSON
messages over stdin/stdout); LspClient handles that with a QProcess,
so the editor never waits for it.
Open editor tabs send their text to the server (didOpen/didChange),
so unsaved edits are seen. Unless the project has its own fortls settings
(.fortlsrc, .fortls.json or .fortls), fortls is told to index
just the fpm folders (src, app, test, example, include and
any source-dir) of the project and each downloaded dependency, rather
than every folder under the project (a Python environment, say).
1"""A small client for the Fortran language server, fortls. 2 3`fortls <https://fortls.fortran-lang.org>`_ indexes the project's Fortran 4sources and answers questions about them: what a name is (hover), where it 5is defined, the outline of a file and the symbols in the whole project. It 6runs as a background process speaking the Language Server Protocol (JSON 7messages over stdin/stdout); `LspClient` handles that with a QProcess, 8so the editor never waits for it. 9 10Open editor tabs send their text to the server (``didOpen``/``didChange``), 11so unsaved edits are seen. Unless the project has its own fortls settings 12(``.fortlsrc``, ``.fortls.json`` or ``.fortls``), fortls is told to index 13just the fpm folders (``src``, ``app``, ``test``, ``example``, ``include`` and 14any ``source-dir``) of the project and each downloaded dependency, rather 15than every folder under the project (a Python environment, say). 16""" 17 18from __future__ import annotations 19 20import html 21import json 22import re 23import shutil 24from collections.abc import Callable 25from pathlib import Path 26from typing import NamedTuple 27 28from qtpy.QtCore import QObject, QProcess, QUrl, Signal 29from qtpy.QtGui import QPalette, QTextDocument 30 31from . import manifest as m 32from .project_tree import fpm_folders 33from .wire import FrameReader, frame 34 35FORTRAN_EXTS = {".f", ".for", ".ftn", ".f77", ".f90", ".f95", ".f03", ".f08", ".f18", ".f23", ".fpp"} 36CONFIG_FILES = (".fortlsrc", ".fortls.json", ".fortls") 37 38# LSP SymbolKind numbers -> words (the ones fortls uses for Fortran). 39SYMBOL_KINDS = {2: "module", 5: "type", 6: "subroutine", 8: "component", 9: "constructor", 40 10: "enum", 11: "interface", 12: "function", 13: "variable", 14: "constant", 41 22: "enumerator", 23: "type", 26: "type parameter"} 42OUTLINE_KINDS = {2, 5, 6, 9, 10, 11, 12, 23} # what the outline lists (no variables) 43# fortls's outline calls subroutines "functions" and programs "modules"; the 44# keyword on the symbol's first line says which it is. 45KEYWORD_RE = re.compile(r"^(?:[^!'\"]*?\s)?(program|submodule|module|subroutine|function)\b", re.IGNORECASE) 46 47 48def executable() -> str | None: 49 return shutil.which("fortls") 50 51 52def is_fortran(path: Path) -> bool: 53 return path.suffix.lower() in FORTRAN_EXTS 54 55 56def to_uri(path: Path) -> str: 57 return QUrl.fromLocalFile(str(path)).toString() 58 59 60def from_uri(uri: str) -> Path: 61 return Path(QUrl(uri).toLocalFile()).resolve() 62 63 64def has_config(project_dir: Path) -> bool: 65 return any((project_dir / name).is_file() for name in CONFIG_FILES) 66 67 68def source_dirs(graph: m.DependencyGraph | None, project_dir: Path) -> list[str]: 69 """``folder/**`` globs for the fpm folders of the project and of every 70 dependency on disk (fortls doesn't look into subfolders otherwise).""" 71 packages = [project_dir] 72 if graph is not None: 73 packages += [p.directory for p in graph.packages.values() 74 if p.name != graph.root and p.kind not in (m.META, m.MISSING) 75 and p.directory and p.directory.is_dir()] 76 folders: set[str] = set() 77 for directory in packages: 78 manifest = directory / "fpm.toml" 79 found = fpm_folders(manifest) if manifest.is_file() else {} 80 folders.update(found or {str(directory.resolve()): ""}) 81 # A folder inside another listed one is already covered by its "/**". 82 kept = [f for f in sorted(folders) 83 if not any(other != f and Path(other) in Path(f).parents for other in folders)] 84 return [f"{folder}/**" for folder in kept] 85 86 87def _packages(graph: m.DependencyGraph | None, project_dir: Path) -> list[tuple[Path, str]]: 88 """(folder, version) of the project and of every dependency on disk.""" 89 out = [(project_dir, "")] 90 if graph is not None: 91 root = graph.packages.get(graph.root) 92 out[0] = (project_dir, (root.version if root else "") or "") 93 out += [(p.directory, p.version or "") for p in graph.packages.values() 94 if p.name != graph.root and p.kind not in (m.META, m.MISSING) 95 and p.directory and p.directory.is_dir()] 96 return out 97 98 99def preprocessor_args(graph: m.DependencyGraph | None, project_dir: Path) -> list[str]: 100 """fortls options for the C preprocessor settings in the fpm.toml files 101 (``[preprocess.cpp]``: macros, suffixes) and the packages' include 102 folders, so #ifdef branches are read as fpm compiles them. [] when no 103 package has preprocessor settings (fortls's own defaults then).""" 104 macros: dict[str, str] = {} 105 suffixes: list[str] = [] 106 includes: list[str] = [] 107 found = False 108 for directory, version in reversed(_packages(graph, project_dir)): # (the project's own win) 109 manifest = directory / "fpm.toml" 110 try: 111 data = m.load_toml(manifest) if manifest.is_file() else {} 112 except Exception: # noqa: BLE001 - an unreadable manifest has no settings 113 data = {} 114 cpp = (data.get("preprocess") or {}).get("cpp") if isinstance(data.get("preprocess"), dict) else None 115 if not isinstance(cpp, dict): 116 continue 117 found = True 118 for macro in cpp.get("macros") or []: 119 name, _, value = str(macro).partition("=") 120 if name.strip(): 121 macros[name.strip()] = value.strip().replace("{version}", version) 122 for suffix in cpp.get("suffixes") or []: 123 suffix = "." + str(suffix).lstrip(".") 124 if suffix not in suffixes: 125 suffixes.append(suffix) 126 includes += [f for f, role in fpm_folders(manifest).items() if role == "include" and f not in includes] 127 if not found: 128 return [] 129 args = [] 130 if macros: 131 args += ["--pp_defs", json.dumps(macros)] 132 if suffixes: 133 args += ["--pp_suffixes", *suffixes] 134 if includes: 135 args += ["--include_dirs", *includes] 136 return args 137 138 139def server_args(graph: m.DependencyGraph | None, project_dir: Path) -> list[str]: 140 args = ["--disable_autoupdate", # it would otherwise ask PyPI for updates 141 "--hover_signature", 142 "--use_signature_help", # complete names only; arguments are hinted on "(" 143 "--lowercase_intrinsics"] # complete "size", not "SIZE" 144 if not has_config(project_dir): # (a .fortlsrc says it all) 145 args += ["--source_dirs", *source_dirs(graph, project_dir)] 146 args += preprocessor_args(graph, project_dir) 147 return args 148 149 150# --------------------------------------------------------- diagnostics 151class LspIssue(NamedTuple): 152 """A problem fortls found in a file (it checks a file when it's opened 153 or saved: the saved text), shown like fortitude's issues.""" 154 line: int # 1-based 155 column: int # 1-based 156 end_line: int 157 end_column: int 158 message: str 159 level: int = 2 # LSP: 1 error, 2 warning, 3 information, 4 hint 160 code: str = "fortls" 161 fix: str = "" 162 fix_safe: bool = False 163 edits: tuple = () 164 source: str = "fortls" 165 166 @property 167 def fixable(self) -> bool: 168 return False 169 170 @property 171 def severity(self) -> str: 172 return "fortls error" if self.level == 1 else "fortls" 173 174 def describe(self) -> str: 175 return f"fortls: {self.message}" 176 177 178def diagnostic_issues(diagnostics) -> list[LspIssue]: 179 """publishDiagnostics' list -> issues (anything malformed is skipped).""" 180 out = [] 181 for d in diagnostics or []: 182 try: 183 r = d["range"] 184 out.append(LspIssue(r["start"]["line"] + 1, r["start"]["character"] + 1, 185 r["end"]["line"] + 1, r["end"]["character"] + 1, 186 " ".join(str(d.get("message", "")).split()), int(d.get("severity") or 2))) 187 except (KeyError, TypeError, ValueError): 188 continue 189 return out 190 191 192# fortls's hover text for an intrinsic is a manual page in markdown that 193# starts with a rule and the name as a heading ("-----\n## sum\n\n### **Name**"); 194# for the project's own names it starts with a ```fortran block. 195INTRINSIC_RE = re.compile(r"^\s*-{3,}[ \t]*\n\s*##[ \t]+(\w+)[ \t]*\n(?:\s*###[ \t]*\**Name\**[ \t]*\n)?") 196 197 198def hover_text(contents) -> str: 199 """The markdown text of a hover answer (MarkupContent, string or list).""" 200 if isinstance(contents, dict): 201 return contents.get("value", "") 202 if isinstance(contents, list): 203 return "\n\n".join(c.get("value", "") if isinstance(c, dict) else str(c) for c in contents) 204 return str(contents or "") 205 206 207def _accent(dark: bool) -> str: 208 """The highlight color, made lighter on a dark background so it reads.""" 209 from qtpy.QtWidgets import QApplication 210 color = QApplication.palette().color(QPalette.Active, QPalette.Highlight) 211 if dark: 212 color = color.lighter(170) 213 elif color.lightness() > 150: 214 color = color.darker(140) 215 return color.name() 216 217 218def code_html(code: str, language: str = "fortran", style: str | None = None) -> str: 219 """A block of code, syntax-highlighted with the editor's Pygments style, 220 on a panel in that style's background color.""" 221 from pygments import highlight 222 from pygments.formatters import HtmlFormatter 223 from pygments.lexers import TextLexer 224 from pygments.styles import get_style_by_name 225 226 from .highlighter import FortranSourceLexer 227 228 style = style or _current_style() 229 lexer = FortranSourceLexer() if language.lower().startswith("fortran") or not language else TextLexer() 230 spans = highlight(code.rstrip("\n"), lexer, HtmlFormatter(nowrap=True, noclasses=True, style=style)) 231 # Synopses aren't always valid Fortran ("[,dim] | [mask]"): no error boxes. 232 spans = re.sub(r"(background-color|border):\s*[^;\"]+;?\s*", "", spans) 233 background = get_style_by_name(style).background_color 234 return (f"<table width='100%' cellspacing='0' cellpadding='6' style='margin:3px 0; " 235 f"background-color:{background}'><tr><td><pre style='margin:0'>{spans.rstrip()}</pre>" 236 "</td></tr></table>") 237 238 239def _current_style() -> str: 240 from qtpy.QtWidgets import QApplication 241 242 from . import highlighter as hl 243 return hl.style_name(QApplication.palette()) 244 245 246def markdown_html(text: str, style: str | None = None) -> str: 247 """Markdown as HTML for a rich-text label (rendered by Qt): code blocks 248 syntax-highlighted, headings as small colored labels.""" 249 out = [] 250 for i, part in enumerate(re.split(r"```([\w+-]*)[ \t]*\n(.*?)```", text, flags=re.DOTALL)): 251 kind = i % 3 252 if kind == 1: 253 language = part 254 elif kind == 2: 255 out.append(code_html(part, language, style)) 256 elif part.strip(): 257 doc = QTextDocument() 258 doc.setMarkdown(part, QTextDocument.MarkdownDialectGitHub) 259 body = re.search(r"<body[^>]*>(.*)</body>", doc.toHtml(), re.DOTALL) 260 out.append(body.group(1).strip() if body else "") 261 result = "".join(out) 262 from . import highlighter as hl 263 accent = _accent(hl.style_is_dark(style or _current_style())) 264 265 def heading(match): 266 title = html.unescape(re.sub(r"<[^>]+>", "", match.group(2))).strip() 267 return (f"<p style='margin:9px 0 2px 0'><span style='color:{accent}; font-weight:600; " 268 f"font-size:small'>{html.escape(title.upper())}</span></p>") 269 270 return re.sub(r"<h(\d)[^>]*>(.*?)</h\1>", heading, result, flags=re.DOTALL) 271 272 273class HoverInfo(NamedTuple): 274 kind: str # "intrinsic", "function", "subroutine", "variable", "parameter", … or "" 275 category: str # an intrinsic's group, e.g. "ARRAY · REDUCTION" 276 summary: str # an intrinsic's one-line description 277 body_html: str # the rest, rendered 278 279 280def _clean(text: str) -> str: 281 # "-----" between the declaration and a doc comment: just a gap (the code 282 # panel already sets them apart; markdown would make the line above a heading). 283 text = re.sub(r"^[ \t]*-{3,}[ \t]*$", "\n", text, flags=re.MULTILINE) 284 # Qt shows _word_ underlined (like a link); the manual pages mean italics. 285 parts = re.split(r"(```.*?```|`[^`]*`)", text, flags=re.DOTALL) 286 return "".join(part if i % 2 else re.sub(r"(?<![\w\\*])_([^_\n]+?)_(?!\w)", r"*\1*", part) 287 for i, part in enumerate(parts)) 288 289 290DECL_KINDS = [("subroutine", r"\bSUBROUTINE\b"), ("function", r"\bFUNCTION\b"), 291 ("program", r"^\s*PROGRAM\b"), ("submodule", r"^\s*SUBMODULE\b"), 292 ("module", r"^\s*MODULE\b(?!\s+PROCEDURE)"), ("interface", r"\bINTERFACE\b"), 293 ("type", r"^\s*TYPE\b(?!\s*\()")] 294 295 296def hover_info(contents, style: str | None = None) -> HoverInfo | None: 297 """fortls's hover answer, taken apart for the hover card (None if empty).""" 298 text = hover_text(contents) 299 if not text.strip(): 300 return None 301 match = INTRINSIC_RE.match(text) 302 if match: 303 rest = text[match.end():] 304 category = summary = "" 305 line = re.match(r"\s*\*\*\w+\*\*\s*-\s*(?:\\?\[([^\]\\]*)\\?\])?\s*(.*)\n", rest) 306 if line: 307 category = " · ".join(p.strip() for p in (line.group(1) or "").split(":") if p.strip()) 308 summary = line.group(2).strip() 309 rest = rest[line.end():] 310 return HoverInfo("intrinsic", category, summary, markdown_html(_clean(rest), style)) 311 kind = "" 312 code = re.match(r"\s*```[\w+-]*[ \t]*\n(.*?)```", text, re.DOTALL) 313 if code: 314 first = code.group(1).strip().splitlines()[0] if code.group(1).strip() else "" 315 if "::" in first and not re.match(r"^\s*TYPE\b\s*(,|::)", first, re.IGNORECASE): 316 kind = "parameter" if re.search(r"\bPARAMETER\b", first, re.IGNORECASE) else "variable" 317 else: 318 kind = next((k for k, pattern in DECL_KINDS if re.search(pattern, first, re.IGNORECASE)), "") 319 if not kind and re.match(r"^\s*TYPE\b", first, re.IGNORECASE): 320 kind = "type" 321 return HoverInfo(kind, "", "", markdown_html(_clean(text), style)) 322 323 324def hover_html(contents) -> str: 325 """fortls's hover text (markdown) as HTML, e.g. for an argument's 326 documentation in the argument hints; "" if there is nothing.""" 327 info = hover_info(contents) 328 if info is None: 329 return "" 330 lead = f"<p style='margin:0 0 4px 0'>{html.escape(info.summary)}</p>" if info.summary else "" 331 return lead + info.body_html 332 333 334def first_location(result) -> tuple[Path, int, int] | None: 335 """(file, 1-based line, 0-based column) from a definition answer 336 (a Location, a list of them, or LocationLinks).""" 337 if isinstance(result, list): 338 result = result[0] if result else None 339 if not isinstance(result, dict): 340 return None 341 uri = result.get("uri") or result.get("targetUri") 342 rng = result.get("targetSelectionRange") or result.get("range") or result.get("targetRange") 343 if not uri or not rng: 344 return None 345 return from_uri(uri), rng["start"]["line"] + 1, rng["start"]["character"] 346 347 348class LspClient(QObject): 349 """One fortls process for the open project.""" 350 351 stopped = Signal(str) # the server exited on its own (with why) 352 message = Signal(str) # something the server wants the user to know 353 diagnostics = Signal(object, object) # file (Path), [LspIssue]: fortls's check of a file 354 355 def __init__(self, parent=None): 356 super().__init__(parent) 357 self.process: QProcess | None = None 358 self.root: Path | None = None 359 self.args: list[str] = [] 360 self._reader = FrameReader() 361 self._next_id = 0 362 self._callbacks: dict[int, Callable] = {} 363 self._ready = False 364 self._queue: list[dict] = [] # sent once initialize has answered 365 self._versions: dict[str, int] = {} # open document uri -> version 366 367 # ---------------------------------------------------------- lifecycle 368 def is_running(self) -> bool: 369 return self.process is not None 370 371 def start(self, root: Path, args: list[str]) -> bool: 372 self.stop() 373 program = executable() 374 if program is None: 375 return False 376 self.root, self.args = root, list(args) 377 process = QProcess(self) 378 process.setWorkingDirectory(str(root)) 379 process.setStandardErrorFile(QProcess.nullDevice()) 380 process.readyReadStandardOutput.connect(self._read) 381 # (Methods, not lambdas holding the process: when Qt deletes the 382 # process, PySide drops those lambdas, and with them possibly the last 383 # reference to the process's Python wrapper, mid-delete: a crash.) 384 process.finished.connect(self._process_finished) 385 process.errorOccurred.connect(self._process_error) 386 self.process = process 387 process.start(program, self.args) 388 self._send({"jsonrpc": "2.0", "id": self._new_id(self._initialized), "method": "initialize", 389 "params": {"processId": None, "rootUri": to_uri(root), "capabilities": {}}}) 390 return True 391 392 def stop(self): 393 process, self.process = self.process, None 394 self._reset() 395 if process is not None: 396 process.kill() 397 process.waitForFinished(1000) 398 process.deleteLater() 399 400 def _reset(self): 401 self._reader.clear() 402 self._callbacks.clear() 403 self._queue.clear() 404 self._versions.clear() 405 self._ready = False 406 407 def _process_finished(self, *_): 408 if isinstance(self.sender(), QProcess): 409 self._finished(self.sender()) 410 411 def _process_error(self, error): 412 if error == QProcess.FailedToStart and isinstance(self.sender(), QProcess): 413 self._finished(self.sender()) 414 415 def _finished(self, process: QProcess): 416 if process is not self.process: 417 return # one we stopped 418 self.process = None 419 self._reset() 420 try: 421 process.deleteLater() 422 except RuntimeError: 423 return # the app is shutting down (Qt has already deleted it) 424 self.stopped.emit("fortls could not be started" if process.error() == QProcess.FailedToStart 425 else f"fortls stopped (exit code {process.exitCode()})") 426 427 def _initialized(self, _result): 428 self._ready = True 429 self._send({"jsonrpc": "2.0", "method": "initialized", "params": {}}) 430 queue, self._queue = self._queue, [] 431 for message in queue: 432 self._send(message) 433 434 # ---------------------------------------------------------- messages 435 def _new_id(self, callback: Callable) -> int: 436 self._next_id += 1 437 self._callbacks[self._next_id] = callback 438 return self._next_id 439 440 def _send(self, message: dict): 441 if self.process is None: 442 return 443 self.process.write(frame(message)) 444 445 def _post(self, message: dict): 446 if self._ready: 447 self._send(message) 448 else: 449 self._queue.append(message) 450 451 def request(self, method: str, params: dict, callback: Callable) -> bool: 452 """Ask the server something; ``callback(result)`` gets the answer 453 (None on an error). False if no server is running.""" 454 if self.process is None: 455 return False 456 self._post({"jsonrpc": "2.0", "id": self._new_id(callback), "method": method, "params": params}) 457 return True 458 459 def notify(self, method: str, params: dict): 460 if self.process is not None: 461 self._post({"jsonrpc": "2.0", "method": method, "params": params}) 462 463 def _read(self): 464 if self.process is None: 465 return 466 for message in self._reader.feed(bytes(self.process.readAllStandardOutput())): 467 self._dispatch(message) 468 469 def _dispatch(self, message: dict): 470 if "method" in message: 471 if "id" in message: # a request from the server: we don't support any 472 self._send({"jsonrpc": "2.0", "id": message["id"], "result": None}) 473 elif message["method"] == "window/showMessage": # e.g. why a rename failed 474 text = (message.get("params") or {}).get("message") 475 if text: 476 self.message.emit(str(text)) 477 elif message["method"] == "textDocument/publishDiagnostics": 478 params = message.get("params") or {} 479 if params.get("uri"): 480 self.diagnostics.emit(from_uri(params["uri"]), diagnostic_issues(params.get("diagnostics"))) 481 return # other notifications (logs) aren't used 482 callback = self._callbacks.pop(message.get("id"), None) 483 if callback is not None: 484 try: 485 callback(message.get("result") if "error" not in message else None) 486 except RuntimeError: 487 pass # the editor that asked has been closed 488 489 # ---------------------------------------------------------- documents 490 def open_document(self, path: Path, text: str): 491 uri = to_uri(path) 492 if self.process is None: 493 return 494 if uri in self._versions: 495 self.change_document(path, text) 496 return 497 self._versions[uri] = 1 498 self.notify("textDocument/didOpen", {"textDocument": { 499 "uri": uri, "languageId": "fortran", "version": 1, "text": text}}) 500 # fortls reads the file from disk on didOpen; send the editor's text 501 # too, in case it has unsaved changes. 502 self.change_document(path, text) 503 504 def change_document(self, path: Path, text: str): 505 uri = to_uri(path) 506 if uri not in self._versions: 507 self.open_document(path, text) 508 return 509 self._versions[uri] += 1 510 self.notify("textDocument/didChange", { 511 "textDocument": {"uri": uri, "version": self._versions[uri]}, 512 "contentChanges": [{"text": text}]}) 513 514 def save_document(self, path: Path): 515 self.notify("textDocument/didSave", {"textDocument": {"uri": to_uri(path)}}) 516 517 def file_changed_on_disk(self, path: Path): 518 """A file that isn't open was rewritten (e.g. by Rename): fortls 519 re-reads a file from disk when told it was saved.""" 520 if to_uri(path) not in self._versions: 521 self.save_document(path) 522 523 def close_document(self, path: Path): 524 uri = to_uri(path) 525 if self._versions.pop(uri, None) is not None: 526 self.notify("textDocument/didClose", {"textDocument": {"uri": uri}}) 527 528 # ---------------------------------------------------------- questions 529 @staticmethod 530 def _at(path: Path, line: int, column: int) -> dict: 531 """Position parameters; ``line`` is 1-based, ``column`` 0-based.""" 532 return {"textDocument": {"uri": to_uri(path)}, 533 "position": {"line": line - 1, "character": column}} 534 535 def hover(self, path: Path, line: int, column: int, callback: Callable) -> bool: 536 return self.request("textDocument/hover", self._at(path, line, column), callback) 537 538 def definition(self, path: Path, line: int, column: int, callback: Callable) -> bool: 539 return self.request("textDocument/definition", self._at(path, line, column), callback) 540 541 def implementation(self, path: Path, line: int, column: int, callback: Callable) -> bool: 542 """For a type-bound procedure, the procedure it binds; for a generic 543 interface, its first module procedure.""" 544 return self.request("textDocument/implementation", self._at(path, line, column), callback) 545 546 def document_symbols(self, path: Path, callback: Callable) -> bool: 547 return self.request("textDocument/documentSymbol", 548 {"textDocument": {"uri": to_uri(path)}}, callback) 549 550 def workspace_symbols(self, query: str, callback: Callable) -> bool: 551 return self.request("workspace/symbol", {"query": query}, callback) 552 553 def completion(self, path: Path, line: int, column: int, callback: Callable) -> bool: 554 return self.request("textDocument/completion", self._at(path, line, column), callback) 555 556 def signature_help(self, path: Path, line: int, column: int, callback: Callable) -> bool: 557 return self.request("textDocument/signatureHelp", self._at(path, line, column), callback) 558 559 def references(self, path: Path, line: int, column: int, callback: Callable) -> bool: 560 return self.request("textDocument/references", 561 {**self._at(path, line, column), "context": {"includeDeclaration": True}}, 562 callback) 563 564 def rename(self, path: Path, line: int, column: int, new_name: str, callback: Callable) -> bool: 565 return self.request("textDocument/rename", 566 {**self._at(path, line, column), "newName": new_name}, callback) 567 568 569def outline(symbols, line_text: Callable[[int], str] | None = None) -> list[tuple[int, str, str, int, int]]: 570 """documentSymbol answer -> [(depth, kind word, name, first line, last 571 line)] for modules, types and procedures, in file order (lines 1-based). 572 ``line_text(line)`` (the file's text) refines the kind words.""" 573 rows = [] 574 575 def kind_word(s, first): 576 word = SYMBOL_KINDS.get(s["kind"], "") 577 if line_text is not None and s["kind"] in (2, 6, 12): 578 match = KEYWORD_RE.match(line_text(first).strip()) 579 if match: 580 word = match.group(1).lower() 581 return word 582 583 def visit(items, depth, container_depth): 584 for s in items or []: 585 if not isinstance(s, dict) or s.get("kind") not in OUTLINE_KINDS: 586 continue 587 rng = s.get("range") or s.get("location", {}).get("range") 588 if not rng: 589 continue 590 level = depth if "children" in s else container_depth.get(s.get("containerName"), -1) + 1 591 rows.append((level, kind_word(s, rng["start"]["line"] + 1), s.get("name", ""), 592 rng["start"]["line"] + 1, rng["end"]["line"] + 1)) 593 container_depth[s.get("name")] = level 594 visit(s.get("children"), depth + 1, container_depth) 595 596 visit(symbols, 0, {}) 597 rows.sort(key=lambda r: (r[3], r[0])) 598 return rows 599 600 601# ------------------------------------------------------------- completion 602# LSP CompletionItemKind numbers -> words. 603COMPLETION_KINDS = {1: "text", 2: "method", 3: "function", 4: "constructor", 5: "component", 604 6: "variable", 7: "type", 8: "interface", 9: "module", 10: "property", 605 13: "enum", 14: "keyword", 15: "snippet", 20: "enumerator", 21: "constant", 606 22: "type", 25: "type parameter"} 607 608 609class Completion(NamedTuple): 610 label: str 611 insert: str # the text to insert (snippet placeholders removed) 612 kind: str 613 detail: str 614 doc: str # markdown (fortls: the declaration and doc comment) 615 616 617def snippet_text(text: str) -> str: 618 """A snippet ("foo(${1:x}, $2)$0") as plain text ("foo(x, )").""" 619 text = re.sub(r"\$\{\d+:([^}]*)\}", r"\1", text) 620 text = re.sub(r"\$\{\d+\}|\$\d+", "", text) 621 return text.replace("\\$", "$").replace("\\}", "}") 622 623 624def completions(result) -> list[Completion]: 625 items = result.get("items", []) if isinstance(result, dict) else result or [] 626 out = [] 627 for item in items: 628 if not isinstance(item, dict) or not item.get("label"): 629 continue 630 insert = item.get("insertText") or item.get("label") 631 edit = item.get("textEdit") 632 if isinstance(edit, dict) and edit.get("newText"): 633 insert = edit["newText"] 634 if item.get("insertTextFormat") == 2: 635 insert = snippet_text(insert) 636 doc = item.get("documentation") or "" 637 if isinstance(doc, dict): 638 doc = doc.get("value", "") 639 out.append(Completion(str(item["label"]), insert, COMPLETION_KINDS.get(item.get("kind"), ""), 640 str(item.get("detail") or ""), str(doc))) 641 return out 642 643 644def code_spans(line: str) -> list[tuple[int, int]]: 645 """The parts of a line that are code: not in a string or a "!" comment.""" 646 spans, start, quote = [], 0, None 647 for i, ch in enumerate(line): 648 if quote: 649 if ch == quote: 650 quote, start = None, i + 1 651 elif ch in "'\"": 652 spans.append((start, i)) 653 quote = ch 654 elif ch == "!": 655 spans.append((start, i)) 656 return spans 657 if quote is None: 658 spans.append((start, len(line))) 659 return spans 660 661 662def in_comment_or_string(line_text: str, column: int) -> bool: 663 """Is ``column`` of a line inside a "!" comment or a quoted string?""" 664 return not any(start <= column <= end for start, end in code_spans(line_text)) 665 666# --------------------------------------------------------- signature help 667def signature_html(result, active_hint: int | None = None) -> str: 668 """signatureHelp answer -> tooltip HTML: the call with the current 669 argument in bold, then that argument's declaration and doc comment.""" 670 if not isinstance(result, dict) or not result.get("signatures"): 671 return "" 672 signatures = result["signatures"] 673 sig = signatures[min(int(result.get("activeSignature") or 0), len(signatures) - 1)] 674 params = sig.get("parameters") or [] 675 active = sig.get("activeParameter", result.get("activeParameter")) 676 if active is None: 677 active = active_hint 678 label = str(sig.get("label", "")) 679 shown = html.escape(label) 680 doc = "" 681 if active is not None and 0 <= int(active) < len(params): 682 param = params[int(active)] 683 name = param.get("label") 684 if isinstance(name, list) and len(name) == 2: # [start, end] offsets 685 start, end = name 686 shown = (html.escape(label[:start]) + "<b><u>" + html.escape(label[start:end]) + "</u></b>" 687 + html.escape(label[end:])) 688 elif isinstance(name, str): 689 # The argument's name inside the parentheses (not the routine's name). 690 open_paren = label.find("(") 691 match = re.search(rf"\b{re.escape(name)}\b", label[open_paren + 1:]) if open_paren >= 0 else None 692 if match: 693 start, end = open_paren + 1 + match.start(), open_paren + 1 + match.end() 694 shown = (html.escape(label[:start]) + "<b><u>" + html.escape(label[start:end]) + "</u></b>" 695 + html.escape(label[end:])) 696 doc = hover_html(param.get("documentation") or "") 697 return f"<pre style='margin:0'>{shown}</pre>" + doc 698 699 700def argument_index(text_before: str) -> int | None: 701 """Which argument (0-based) the cursor is in, from the code before it on 702 the line: the commas at the level of the innermost open "(" (None if the 703 cursor isn't inside parentheses, or the innermost is a "[").""" 704 open_brackets: list[list] = [] # [bracket, commas] of each one still open 705 for start, end in code_spans(text_before): 706 for ch in text_before[start:end]: 707 if ch in "([": 708 open_brackets.append([ch, 0]) 709 elif ch in ")]": 710 if open_brackets: 711 open_brackets.pop() 712 elif ch == "," and open_brackets: 713 open_brackets[-1][1] += 1 714 if not open_brackets: 715 return None 716 bracket, commas = open_brackets[-1] 717 return commas if bracket == "(" else None 718 719# ------------------------------------------------------- references / rename 720class TextEdit(NamedTuple): 721 line: int # 0-based, as in LSP 722 column: int 723 end_line: int 724 end_column: int 725 new_text: str 726 727 728def locations(result) -> list[tuple[Path, int, int, int]]: 729 """references answer -> [(file, 1-based line, start column, end column)].""" 730 out = [] 731 for loc in result or []: 732 try: 733 rng = loc["range"] 734 out.append((from_uri(loc["uri"]), rng["start"]["line"] + 1, rng["start"]["character"], 735 rng["end"]["character"] if rng["end"]["line"] == rng["start"]["line"] 736 else rng["start"]["character"])) 737 except (KeyError, TypeError): 738 continue 739 return sorted(set(out), key=lambda r: (str(r[0]), r[1], r[2])) 740 741 742def workspace_edits(result) -> dict[Path, list[TextEdit]]: 743 """rename answer (a WorkspaceEdit) -> {file: [edits]}.""" 744 if not isinstance(result, dict): 745 return {} 746 pairs = list((result.get("changes") or {}).items()) 747 for change in result.get("documentChanges") or []: 748 if isinstance(change, dict) and "textDocument" in change: 749 pairs.append((change["textDocument"]["uri"], change.get("edits", []))) 750 out: dict[Path, list[TextEdit]] = {} 751 for uri, edits in pairs: 752 for e in edits or []: 753 try: 754 r = e["range"] 755 out.setdefault(from_uri(uri), []).append(TextEdit( 756 r["start"]["line"], r["start"]["character"], r["end"]["line"], r["end"]["character"], 757 e["newText"])) 758 except (KeyError, TypeError): 759 continue 760 return out 761 762 763def apply_edits(text: str, edits: list[TextEdit]) -> str: 764 """Apply LSP text edits to a text ("\\n" line ends), last first so the 765 earlier positions stay valid.""" 766 lines = text.split("\n") 767 starts = [0] 768 for line in lines: 769 starts.append(starts[-1] + len(line) + 1) 770 771 def offset(line: int, column: int) -> int: 772 line = min(max(line, 0), len(lines) - 1) 773 return starts[line] + min(column, len(lines[line])) 774 775 for e in sorted(edits, key=lambda e: (e.line, e.column), reverse=True): 776 text = text[:offset(e.line, e.column)] + e.new_text + text[offset(e.end_line, e.end_column):] 777 return text 778 779 780def add_result_variable_edits(edits: dict[Path, list[TextEdit]], old: str, new: str, 781 file_text: Callable[[Path], str]) -> dict[Path, list[TextEdit]]: 782 """fortls's rename misses a function's result variable: in a function 783 without a ``result(...)`` clause, ``old = ...`` inside the body also 784 names the function and must be renamed too. Adds those edits.""" 785 header = re.compile(rf"\bfunction\s+{re.escape(old)}\s*\(", re.IGNORECASE) 786 word = re.compile(rf"\b{re.escape(old)}\b", re.IGNORECASE) 787 end = re.compile(r"^\s*(end\s*(function\b|$)|contains\b)", re.IGNORECASE) 788 out = {} 789 for path, file_edits in edits.items(): 790 try: 791 lines = file_text(path).split("\n") 792 except OSError: 793 out[path] = file_edits 794 continue 795 have = {(e.line, e.column) for e in file_edits} 796 added = list(file_edits) 797 for i, line in enumerate(lines): 798 code = "".join(line[a:b] for a, b in code_spans(line)) 799 if not header.search(code): 800 continue 801 # The whole statement (it may continue over "&" lines). 802 statement, j = code, i 803 while statement.rstrip().endswith("&") and j + 1 < len(lines): 804 j += 1 805 statement += "".join(lines[j][a:b] for a, b in code_spans(lines[j])) 806 if re.search(r"\bresult\s*\(", statement, re.IGNORECASE): 807 continue # the result has its own name 808 for k in range(j + 1, len(lines)): 809 if end.match(lines[k]): 810 break 811 for a, b in code_spans(lines[k]): 812 for match in word.finditer(lines[k], a, b): 813 if (k, match.start()) not in have: 814 have.add((k, match.start())) 815 added.append(TextEdit(k, match.start(), k, match.end(), new)) 816 out[path] = added 817 return out
69def source_dirs(graph: m.DependencyGraph | None, project_dir: Path) -> list[str]: 70 """``folder/**`` globs for the fpm folders of the project and of every 71 dependency on disk (fortls doesn't look into subfolders otherwise).""" 72 packages = [project_dir] 73 if graph is not None: 74 packages += [p.directory for p in graph.packages.values() 75 if p.name != graph.root and p.kind not in (m.META, m.MISSING) 76 and p.directory and p.directory.is_dir()] 77 folders: set[str] = set() 78 for directory in packages: 79 manifest = directory / "fpm.toml" 80 found = fpm_folders(manifest) if manifest.is_file() else {} 81 folders.update(found or {str(directory.resolve()): ""}) 82 # A folder inside another listed one is already covered by its "/**". 83 kept = [f for f in sorted(folders) 84 if not any(other != f and Path(other) in Path(f).parents for other in folders)] 85 return [f"{folder}/**" for folder in kept]
folder/** globs for the fpm folders of the project and of every
dependency on disk (fortls doesn't look into subfolders otherwise).
100def preprocessor_args(graph: m.DependencyGraph | None, project_dir: Path) -> list[str]: 101 """fortls options for the C preprocessor settings in the fpm.toml files 102 (``[preprocess.cpp]``: macros, suffixes) and the packages' include 103 folders, so #ifdef branches are read as fpm compiles them. [] when no 104 package has preprocessor settings (fortls's own defaults then).""" 105 macros: dict[str, str] = {} 106 suffixes: list[str] = [] 107 includes: list[str] = [] 108 found = False 109 for directory, version in reversed(_packages(graph, project_dir)): # (the project's own win) 110 manifest = directory / "fpm.toml" 111 try: 112 data = m.load_toml(manifest) if manifest.is_file() else {} 113 except Exception: # noqa: BLE001 - an unreadable manifest has no settings 114 data = {} 115 cpp = (data.get("preprocess") or {}).get("cpp") if isinstance(data.get("preprocess"), dict) else None 116 if not isinstance(cpp, dict): 117 continue 118 found = True 119 for macro in cpp.get("macros") or []: 120 name, _, value = str(macro).partition("=") 121 if name.strip(): 122 macros[name.strip()] = value.strip().replace("{version}", version) 123 for suffix in cpp.get("suffixes") or []: 124 suffix = "." + str(suffix).lstrip(".") 125 if suffix not in suffixes: 126 suffixes.append(suffix) 127 includes += [f for f, role in fpm_folders(manifest).items() if role == "include" and f not in includes] 128 if not found: 129 return [] 130 args = [] 131 if macros: 132 args += ["--pp_defs", json.dumps(macros)] 133 if suffixes: 134 args += ["--pp_suffixes", *suffixes] 135 if includes: 136 args += ["--include_dirs", *includes] 137 return args
fortls options for the C preprocessor settings in the fpm.toml files
([preprocess.cpp]: macros, suffixes) and the packages' include
folders, so #ifdef branches are read as fpm compiles them. [] when no
package has preprocessor settings (fortls's own defaults then).
140def server_args(graph: m.DependencyGraph | None, project_dir: Path) -> list[str]: 141 args = ["--disable_autoupdate", # it would otherwise ask PyPI for updates 142 "--hover_signature", 143 "--use_signature_help", # complete names only; arguments are hinted on "(" 144 "--lowercase_intrinsics"] # complete "size", not "SIZE" 145 if not has_config(project_dir): # (a .fortlsrc says it all) 146 args += ["--source_dirs", *source_dirs(graph, project_dir)] 147 args += preprocessor_args(graph, project_dir) 148 return args
152class LspIssue(NamedTuple): 153 """A problem fortls found in a file (it checks a file when it's opened 154 or saved: the saved text), shown like fortitude's issues.""" 155 line: int # 1-based 156 column: int # 1-based 157 end_line: int 158 end_column: int 159 message: str 160 level: int = 2 # LSP: 1 error, 2 warning, 3 information, 4 hint 161 code: str = "fortls" 162 fix: str = "" 163 fix_safe: bool = False 164 edits: tuple = () 165 source: str = "fortls" 166 167 @property 168 def fixable(self) -> bool: 169 return False 170 171 @property 172 def severity(self) -> str: 173 return "fortls error" if self.level == 1 else "fortls" 174 175 def describe(self) -> str: 176 return f"fortls: {self.message}"
A problem fortls found in a file (it checks a file when it's opened or saved: the saved text), shown like fortitude's issues.
Create new instance of LspIssue(line, column, end_line, end_column, message, level, code, fix, fix_safe, edits, source)
179def diagnostic_issues(diagnostics) -> list[LspIssue]: 180 """publishDiagnostics' list -> issues (anything malformed is skipped).""" 181 out = [] 182 for d in diagnostics or []: 183 try: 184 r = d["range"] 185 out.append(LspIssue(r["start"]["line"] + 1, r["start"]["character"] + 1, 186 r["end"]["line"] + 1, r["end"]["character"] + 1, 187 " ".join(str(d.get("message", "")).split()), int(d.get("severity") or 2))) 188 except (KeyError, TypeError, ValueError): 189 continue 190 return out
publishDiagnostics' list -> issues (anything malformed is skipped).
199def hover_text(contents) -> str: 200 """The markdown text of a hover answer (MarkupContent, string or list).""" 201 if isinstance(contents, dict): 202 return contents.get("value", "") 203 if isinstance(contents, list): 204 return "\n\n".join(c.get("value", "") if isinstance(c, dict) else str(c) for c in contents) 205 return str(contents or "")
The markdown text of a hover answer (MarkupContent, string or list).
219def code_html(code: str, language: str = "fortran", style: str | None = None) -> str: 220 """A block of code, syntax-highlighted with the editor's Pygments style, 221 on a panel in that style's background color.""" 222 from pygments import highlight 223 from pygments.formatters import HtmlFormatter 224 from pygments.lexers import TextLexer 225 from pygments.styles import get_style_by_name 226 227 from .highlighter import FortranSourceLexer 228 229 style = style or _current_style() 230 lexer = FortranSourceLexer() if language.lower().startswith("fortran") or not language else TextLexer() 231 spans = highlight(code.rstrip("\n"), lexer, HtmlFormatter(nowrap=True, noclasses=True, style=style)) 232 # Synopses aren't always valid Fortran ("[,dim] | [mask]"): no error boxes. 233 spans = re.sub(r"(background-color|border):\s*[^;\"]+;?\s*", "", spans) 234 background = get_style_by_name(style).background_color 235 return (f"<table width='100%' cellspacing='0' cellpadding='6' style='margin:3px 0; " 236 f"background-color:{background}'><tr><td><pre style='margin:0'>{spans.rstrip()}</pre>" 237 "</td></tr></table>")
A block of code, syntax-highlighted with the editor's Pygments style, on a panel in that style's background color.
247def markdown_html(text: str, style: str | None = None) -> str: 248 """Markdown as HTML for a rich-text label (rendered by Qt): code blocks 249 syntax-highlighted, headings as small colored labels.""" 250 out = [] 251 for i, part in enumerate(re.split(r"```([\w+-]*)[ \t]*\n(.*?)```", text, flags=re.DOTALL)): 252 kind = i % 3 253 if kind == 1: 254 language = part 255 elif kind == 2: 256 out.append(code_html(part, language, style)) 257 elif part.strip(): 258 doc = QTextDocument() 259 doc.setMarkdown(part, QTextDocument.MarkdownDialectGitHub) 260 body = re.search(r"<body[^>]*>(.*)</body>", doc.toHtml(), re.DOTALL) 261 out.append(body.group(1).strip() if body else "") 262 result = "".join(out) 263 from . import highlighter as hl 264 accent = _accent(hl.style_is_dark(style or _current_style())) 265 266 def heading(match): 267 title = html.unescape(re.sub(r"<[^>]+>", "", match.group(2))).strip() 268 return (f"<p style='margin:9px 0 2px 0'><span style='color:{accent}; font-weight:600; " 269 f"font-size:small'>{html.escape(title.upper())}</span></p>") 270 271 return re.sub(r"<h(\d)[^>]*>(.*?)</h\1>", heading, result, flags=re.DOTALL)
Markdown as HTML for a rich-text label (rendered by Qt): code blocks syntax-highlighted, headings as small colored labels.
274class HoverInfo(NamedTuple): 275 kind: str # "intrinsic", "function", "subroutine", "variable", "parameter", … or "" 276 category: str # an intrinsic's group, e.g. "ARRAY · REDUCTION" 277 summary: str # an intrinsic's one-line description 278 body_html: str # the rest, rendered
HoverInfo(kind, category, summary, body_html)
297def hover_info(contents, style: str | None = None) -> HoverInfo | None: 298 """fortls's hover answer, taken apart for the hover card (None if empty).""" 299 text = hover_text(contents) 300 if not text.strip(): 301 return None 302 match = INTRINSIC_RE.match(text) 303 if match: 304 rest = text[match.end():] 305 category = summary = "" 306 line = re.match(r"\s*\*\*\w+\*\*\s*-\s*(?:\\?\[([^\]\\]*)\\?\])?\s*(.*)\n", rest) 307 if line: 308 category = " · ".join(p.strip() for p in (line.group(1) or "").split(":") if p.strip()) 309 summary = line.group(2).strip() 310 rest = rest[line.end():] 311 return HoverInfo("intrinsic", category, summary, markdown_html(_clean(rest), style)) 312 kind = "" 313 code = re.match(r"\s*```[\w+-]*[ \t]*\n(.*?)```", text, re.DOTALL) 314 if code: 315 first = code.group(1).strip().splitlines()[0] if code.group(1).strip() else "" 316 if "::" in first and not re.match(r"^\s*TYPE\b\s*(,|::)", first, re.IGNORECASE): 317 kind = "parameter" if re.search(r"\bPARAMETER\b", first, re.IGNORECASE) else "variable" 318 else: 319 kind = next((k for k, pattern in DECL_KINDS if re.search(pattern, first, re.IGNORECASE)), "") 320 if not kind and re.match(r"^\s*TYPE\b", first, re.IGNORECASE): 321 kind = "type" 322 return HoverInfo(kind, "", "", markdown_html(_clean(text), style))
fortls's hover answer, taken apart for the hover card (None if empty).
325def hover_html(contents) -> str: 326 """fortls's hover text (markdown) as HTML, e.g. for an argument's 327 documentation in the argument hints; "" if there is nothing.""" 328 info = hover_info(contents) 329 if info is None: 330 return "" 331 lead = f"<p style='margin:0 0 4px 0'>{html.escape(info.summary)}</p>" if info.summary else "" 332 return lead + info.body_html
fortls's hover text (markdown) as HTML, e.g. for an argument's documentation in the argument hints; "" if there is nothing.
335def first_location(result) -> tuple[Path, int, int] | None: 336 """(file, 1-based line, 0-based column) from a definition answer 337 (a Location, a list of them, or LocationLinks).""" 338 if isinstance(result, list): 339 result = result[0] if result else None 340 if not isinstance(result, dict): 341 return None 342 uri = result.get("uri") or result.get("targetUri") 343 rng = result.get("targetSelectionRange") or result.get("range") or result.get("targetRange") 344 if not uri or not rng: 345 return None 346 return from_uri(uri), rng["start"]["line"] + 1, rng["start"]["character"]
(file, 1-based line, 0-based column) from a definition answer (a Location, a list of them, or LocationLinks).
349class LspClient(QObject): 350 """One fortls process for the open project.""" 351 352 stopped = Signal(str) # the server exited on its own (with why) 353 message = Signal(str) # something the server wants the user to know 354 diagnostics = Signal(object, object) # file (Path), [LspIssue]: fortls's check of a file 355 356 def __init__(self, parent=None): 357 super().__init__(parent) 358 self.process: QProcess | None = None 359 self.root: Path | None = None 360 self.args: list[str] = [] 361 self._reader = FrameReader() 362 self._next_id = 0 363 self._callbacks: dict[int, Callable] = {} 364 self._ready = False 365 self._queue: list[dict] = [] # sent once initialize has answered 366 self._versions: dict[str, int] = {} # open document uri -> version 367 368 # ---------------------------------------------------------- lifecycle 369 def is_running(self) -> bool: 370 return self.process is not None 371 372 def start(self, root: Path, args: list[str]) -> bool: 373 self.stop() 374 program = executable() 375 if program is None: 376 return False 377 self.root, self.args = root, list(args) 378 process = QProcess(self) 379 process.setWorkingDirectory(str(root)) 380 process.setStandardErrorFile(QProcess.nullDevice()) 381 process.readyReadStandardOutput.connect(self._read) 382 # (Methods, not lambdas holding the process: when Qt deletes the 383 # process, PySide drops those lambdas, and with them possibly the last 384 # reference to the process's Python wrapper, mid-delete: a crash.) 385 process.finished.connect(self._process_finished) 386 process.errorOccurred.connect(self._process_error) 387 self.process = process 388 process.start(program, self.args) 389 self._send({"jsonrpc": "2.0", "id": self._new_id(self._initialized), "method": "initialize", 390 "params": {"processId": None, "rootUri": to_uri(root), "capabilities": {}}}) 391 return True 392 393 def stop(self): 394 process, self.process = self.process, None 395 self._reset() 396 if process is not None: 397 process.kill() 398 process.waitForFinished(1000) 399 process.deleteLater() 400 401 def _reset(self): 402 self._reader.clear() 403 self._callbacks.clear() 404 self._queue.clear() 405 self._versions.clear() 406 self._ready = False 407 408 def _process_finished(self, *_): 409 if isinstance(self.sender(), QProcess): 410 self._finished(self.sender()) 411 412 def _process_error(self, error): 413 if error == QProcess.FailedToStart and isinstance(self.sender(), QProcess): 414 self._finished(self.sender()) 415 416 def _finished(self, process: QProcess): 417 if process is not self.process: 418 return # one we stopped 419 self.process = None 420 self._reset() 421 try: 422 process.deleteLater() 423 except RuntimeError: 424 return # the app is shutting down (Qt has already deleted it) 425 self.stopped.emit("fortls could not be started" if process.error() == QProcess.FailedToStart 426 else f"fortls stopped (exit code {process.exitCode()})") 427 428 def _initialized(self, _result): 429 self._ready = True 430 self._send({"jsonrpc": "2.0", "method": "initialized", "params": {}}) 431 queue, self._queue = self._queue, [] 432 for message in queue: 433 self._send(message) 434 435 # ---------------------------------------------------------- messages 436 def _new_id(self, callback: Callable) -> int: 437 self._next_id += 1 438 self._callbacks[self._next_id] = callback 439 return self._next_id 440 441 def _send(self, message: dict): 442 if self.process is None: 443 return 444 self.process.write(frame(message)) 445 446 def _post(self, message: dict): 447 if self._ready: 448 self._send(message) 449 else: 450 self._queue.append(message) 451 452 def request(self, method: str, params: dict, callback: Callable) -> bool: 453 """Ask the server something; ``callback(result)`` gets the answer 454 (None on an error). False if no server is running.""" 455 if self.process is None: 456 return False 457 self._post({"jsonrpc": "2.0", "id": self._new_id(callback), "method": method, "params": params}) 458 return True 459 460 def notify(self, method: str, params: dict): 461 if self.process is not None: 462 self._post({"jsonrpc": "2.0", "method": method, "params": params}) 463 464 def _read(self): 465 if self.process is None: 466 return 467 for message in self._reader.feed(bytes(self.process.readAllStandardOutput())): 468 self._dispatch(message) 469 470 def _dispatch(self, message: dict): 471 if "method" in message: 472 if "id" in message: # a request from the server: we don't support any 473 self._send({"jsonrpc": "2.0", "id": message["id"], "result": None}) 474 elif message["method"] == "window/showMessage": # e.g. why a rename failed 475 text = (message.get("params") or {}).get("message") 476 if text: 477 self.message.emit(str(text)) 478 elif message["method"] == "textDocument/publishDiagnostics": 479 params = message.get("params") or {} 480 if params.get("uri"): 481 self.diagnostics.emit(from_uri(params["uri"]), diagnostic_issues(params.get("diagnostics"))) 482 return # other notifications (logs) aren't used 483 callback = self._callbacks.pop(message.get("id"), None) 484 if callback is not None: 485 try: 486 callback(message.get("result") if "error" not in message else None) 487 except RuntimeError: 488 pass # the editor that asked has been closed 489 490 # ---------------------------------------------------------- documents 491 def open_document(self, path: Path, text: str): 492 uri = to_uri(path) 493 if self.process is None: 494 return 495 if uri in self._versions: 496 self.change_document(path, text) 497 return 498 self._versions[uri] = 1 499 self.notify("textDocument/didOpen", {"textDocument": { 500 "uri": uri, "languageId": "fortran", "version": 1, "text": text}}) 501 # fortls reads the file from disk on didOpen; send the editor's text 502 # too, in case it has unsaved changes. 503 self.change_document(path, text) 504 505 def change_document(self, path: Path, text: str): 506 uri = to_uri(path) 507 if uri not in self._versions: 508 self.open_document(path, text) 509 return 510 self._versions[uri] += 1 511 self.notify("textDocument/didChange", { 512 "textDocument": {"uri": uri, "version": self._versions[uri]}, 513 "contentChanges": [{"text": text}]}) 514 515 def save_document(self, path: Path): 516 self.notify("textDocument/didSave", {"textDocument": {"uri": to_uri(path)}}) 517 518 def file_changed_on_disk(self, path: Path): 519 """A file that isn't open was rewritten (e.g. by Rename): fortls 520 re-reads a file from disk when told it was saved.""" 521 if to_uri(path) not in self._versions: 522 self.save_document(path) 523 524 def close_document(self, path: Path): 525 uri = to_uri(path) 526 if self._versions.pop(uri, None) is not None: 527 self.notify("textDocument/didClose", {"textDocument": {"uri": uri}}) 528 529 # ---------------------------------------------------------- questions 530 @staticmethod 531 def _at(path: Path, line: int, column: int) -> dict: 532 """Position parameters; ``line`` is 1-based, ``column`` 0-based.""" 533 return {"textDocument": {"uri": to_uri(path)}, 534 "position": {"line": line - 1, "character": column}} 535 536 def hover(self, path: Path, line: int, column: int, callback: Callable) -> bool: 537 return self.request("textDocument/hover", self._at(path, line, column), callback) 538 539 def definition(self, path: Path, line: int, column: int, callback: Callable) -> bool: 540 return self.request("textDocument/definition", self._at(path, line, column), callback) 541 542 def implementation(self, path: Path, line: int, column: int, callback: Callable) -> bool: 543 """For a type-bound procedure, the procedure it binds; for a generic 544 interface, its first module procedure.""" 545 return self.request("textDocument/implementation", self._at(path, line, column), callback) 546 547 def document_symbols(self, path: Path, callback: Callable) -> bool: 548 return self.request("textDocument/documentSymbol", 549 {"textDocument": {"uri": to_uri(path)}}, callback) 550 551 def workspace_symbols(self, query: str, callback: Callable) -> bool: 552 return self.request("workspace/symbol", {"query": query}, callback) 553 554 def completion(self, path: Path, line: int, column: int, callback: Callable) -> bool: 555 return self.request("textDocument/completion", self._at(path, line, column), callback) 556 557 def signature_help(self, path: Path, line: int, column: int, callback: Callable) -> bool: 558 return self.request("textDocument/signatureHelp", self._at(path, line, column), callback) 559 560 def references(self, path: Path, line: int, column: int, callback: Callable) -> bool: 561 return self.request("textDocument/references", 562 {**self._at(path, line, column), "context": {"includeDeclaration": True}}, 563 callback) 564 565 def rename(self, path: Path, line: int, column: int, new_name: str, callback: Callable) -> bool: 566 return self.request("textDocument/rename", 567 {**self._at(path, line, column), "newName": new_name}, callback)
One fortls process for the open project.
356 def __init__(self, parent=None): 357 super().__init__(parent) 358 self.process: QProcess | None = None 359 self.root: Path | None = None 360 self.args: list[str] = [] 361 self._reader = FrameReader() 362 self._next_id = 0 363 self._callbacks: dict[int, Callable] = {} 364 self._ready = False 365 self._queue: list[dict] = [] # sent once initialize has answered 366 self._versions: dict[str, int] = {} # open document uri -> version
__init__(self, /, parent: PySide6.QtCore.QObject | None = None, *, objectName: str | None = None) -> None
Initialize self. See help(type(self)) for accurate signature.
372 def start(self, root: Path, args: list[str]) -> bool: 373 self.stop() 374 program = executable() 375 if program is None: 376 return False 377 self.root, self.args = root, list(args) 378 process = QProcess(self) 379 process.setWorkingDirectory(str(root)) 380 process.setStandardErrorFile(QProcess.nullDevice()) 381 process.readyReadStandardOutput.connect(self._read) 382 # (Methods, not lambdas holding the process: when Qt deletes the 383 # process, PySide drops those lambdas, and with them possibly the last 384 # reference to the process's Python wrapper, mid-delete: a crash.) 385 process.finished.connect(self._process_finished) 386 process.errorOccurred.connect(self._process_error) 387 self.process = process 388 process.start(program, self.args) 389 self._send({"jsonrpc": "2.0", "id": self._new_id(self._initialized), "method": "initialize", 390 "params": {"processId": None, "rootUri": to_uri(root), "capabilities": {}}}) 391 return True
452 def request(self, method: str, params: dict, callback: Callable) -> bool: 453 """Ask the server something; ``callback(result)`` gets the answer 454 (None on an error). False if no server is running.""" 455 if self.process is None: 456 return False 457 self._post({"jsonrpc": "2.0", "id": self._new_id(callback), "method": method, "params": params}) 458 return True
Ask the server something; callback(result) gets the answer
(None on an error). False if no server is running.
491 def open_document(self, path: Path, text: str): 492 uri = to_uri(path) 493 if self.process is None: 494 return 495 if uri in self._versions: 496 self.change_document(path, text) 497 return 498 self._versions[uri] = 1 499 self.notify("textDocument/didOpen", {"textDocument": { 500 "uri": uri, "languageId": "fortran", "version": 1, "text": text}}) 501 # fortls reads the file from disk on didOpen; send the editor's text 502 # too, in case it has unsaved changes. 503 self.change_document(path, text)
505 def change_document(self, path: Path, text: str): 506 uri = to_uri(path) 507 if uri not in self._versions: 508 self.open_document(path, text) 509 return 510 self._versions[uri] += 1 511 self.notify("textDocument/didChange", { 512 "textDocument": {"uri": uri, "version": self._versions[uri]}, 513 "contentChanges": [{"text": text}]})
518 def file_changed_on_disk(self, path: Path): 519 """A file that isn't open was rewritten (e.g. by Rename): fortls 520 re-reads a file from disk when told it was saved.""" 521 if to_uri(path) not in self._versions: 522 self.save_document(path)
A file that isn't open was rewritten (e.g. by Rename): fortls re-reads a file from disk when told it was saved.
542 def implementation(self, path: Path, line: int, column: int, callback: Callable) -> bool: 543 """For a type-bound procedure, the procedure it binds; for a generic 544 interface, its first module procedure.""" 545 return self.request("textDocument/implementation", self._at(path, line, column), callback)
For a type-bound procedure, the procedure it binds; for a generic interface, its first module procedure.
570def outline(symbols, line_text: Callable[[int], str] | None = None) -> list[tuple[int, str, str, int, int]]: 571 """documentSymbol answer -> [(depth, kind word, name, first line, last 572 line)] for modules, types and procedures, in file order (lines 1-based). 573 ``line_text(line)`` (the file's text) refines the kind words.""" 574 rows = [] 575 576 def kind_word(s, first): 577 word = SYMBOL_KINDS.get(s["kind"], "") 578 if line_text is not None and s["kind"] in (2, 6, 12): 579 match = KEYWORD_RE.match(line_text(first).strip()) 580 if match: 581 word = match.group(1).lower() 582 return word 583 584 def visit(items, depth, container_depth): 585 for s in items or []: 586 if not isinstance(s, dict) or s.get("kind") not in OUTLINE_KINDS: 587 continue 588 rng = s.get("range") or s.get("location", {}).get("range") 589 if not rng: 590 continue 591 level = depth if "children" in s else container_depth.get(s.get("containerName"), -1) + 1 592 rows.append((level, kind_word(s, rng["start"]["line"] + 1), s.get("name", ""), 593 rng["start"]["line"] + 1, rng["end"]["line"] + 1)) 594 container_depth[s.get("name")] = level 595 visit(s.get("children"), depth + 1, container_depth) 596 597 visit(symbols, 0, {}) 598 rows.sort(key=lambda r: (r[3], r[0])) 599 return rows
documentSymbol answer -> [(depth, kind word, name, first line, last
line)] for modules, types and procedures, in file order (lines 1-based).
line_text(line) (the file's text) refines the kind words.
610class Completion(NamedTuple): 611 label: str 612 insert: str # the text to insert (snippet placeholders removed) 613 kind: str 614 detail: str 615 doc: str # markdown (fortls: the declaration and doc comment)
Completion(label, insert, kind, detail, doc)
618def snippet_text(text: str) -> str: 619 """A snippet ("foo(${1:x}, $2)$0") as plain text ("foo(x, )").""" 620 text = re.sub(r"\$\{\d+:([^}]*)\}", r"\1", text) 621 text = re.sub(r"\$\{\d+\}|\$\d+", "", text) 622 return text.replace("\\$", "$").replace("\\}", "}")
A snippet ("foo(${1:x}, $2)$0") as plain text ("foo(x, )").
625def completions(result) -> list[Completion]: 626 items = result.get("items", []) if isinstance(result, dict) else result or [] 627 out = [] 628 for item in items: 629 if not isinstance(item, dict) or not item.get("label"): 630 continue 631 insert = item.get("insertText") or item.get("label") 632 edit = item.get("textEdit") 633 if isinstance(edit, dict) and edit.get("newText"): 634 insert = edit["newText"] 635 if item.get("insertTextFormat") == 2: 636 insert = snippet_text(insert) 637 doc = item.get("documentation") or "" 638 if isinstance(doc, dict): 639 doc = doc.get("value", "") 640 out.append(Completion(str(item["label"]), insert, COMPLETION_KINDS.get(item.get("kind"), ""), 641 str(item.get("detail") or ""), str(doc))) 642 return out
645def code_spans(line: str) -> list[tuple[int, int]]: 646 """The parts of a line that are code: not in a string or a "!" comment.""" 647 spans, start, quote = [], 0, None 648 for i, ch in enumerate(line): 649 if quote: 650 if ch == quote: 651 quote, start = None, i + 1 652 elif ch in "'\"": 653 spans.append((start, i)) 654 quote = ch 655 elif ch == "!": 656 spans.append((start, i)) 657 return spans 658 if quote is None: 659 spans.append((start, len(line))) 660 return spans
The parts of a line that are code: not in a string or a "!" comment.
663def in_comment_or_string(line_text: str, column: int) -> bool: 664 """Is ``column`` of a line inside a "!" comment or a quoted string?""" 665 return not any(start <= column <= end for start, end in code_spans(line_text))
Is column of a line inside a "!" comment or a quoted string?
668def signature_html(result, active_hint: int | None = None) -> str: 669 """signatureHelp answer -> tooltip HTML: the call with the current 670 argument in bold, then that argument's declaration and doc comment.""" 671 if not isinstance(result, dict) or not result.get("signatures"): 672 return "" 673 signatures = result["signatures"] 674 sig = signatures[min(int(result.get("activeSignature") or 0), len(signatures) - 1)] 675 params = sig.get("parameters") or [] 676 active = sig.get("activeParameter", result.get("activeParameter")) 677 if active is None: 678 active = active_hint 679 label = str(sig.get("label", "")) 680 shown = html.escape(label) 681 doc = "" 682 if active is not None and 0 <= int(active) < len(params): 683 param = params[int(active)] 684 name = param.get("label") 685 if isinstance(name, list) and len(name) == 2: # [start, end] offsets 686 start, end = name 687 shown = (html.escape(label[:start]) + "<b><u>" + html.escape(label[start:end]) + "</u></b>" 688 + html.escape(label[end:])) 689 elif isinstance(name, str): 690 # The argument's name inside the parentheses (not the routine's name). 691 open_paren = label.find("(") 692 match = re.search(rf"\b{re.escape(name)}\b", label[open_paren + 1:]) if open_paren >= 0 else None 693 if match: 694 start, end = open_paren + 1 + match.start(), open_paren + 1 + match.end() 695 shown = (html.escape(label[:start]) + "<b><u>" + html.escape(label[start:end]) + "</u></b>" 696 + html.escape(label[end:])) 697 doc = hover_html(param.get("documentation") or "") 698 return f"<pre style='margin:0'>{shown}</pre>" + doc
signatureHelp answer -> tooltip HTML: the call with the current argument in bold, then that argument's declaration and doc comment.
701def argument_index(text_before: str) -> int | None: 702 """Which argument (0-based) the cursor is in, from the code before it on 703 the line: the commas at the level of the innermost open "(" (None if the 704 cursor isn't inside parentheses, or the innermost is a "[").""" 705 open_brackets: list[list] = [] # [bracket, commas] of each one still open 706 for start, end in code_spans(text_before): 707 for ch in text_before[start:end]: 708 if ch in "([": 709 open_brackets.append([ch, 0]) 710 elif ch in ")]": 711 if open_brackets: 712 open_brackets.pop() 713 elif ch == "," and open_brackets: 714 open_brackets[-1][1] += 1 715 if not open_brackets: 716 return None 717 bracket, commas = open_brackets[-1] 718 return commas if bracket == "(" else None
Which argument (0-based) the cursor is in, from the code before it on the line: the commas at the level of the innermost open "(" (None if the cursor isn't inside parentheses, or the innermost is a "[").
721class TextEdit(NamedTuple): 722 line: int # 0-based, as in LSP 723 column: int 724 end_line: int 725 end_column: int 726 new_text: str
TextEdit(line, column, end_line, end_column, new_text)
729def locations(result) -> list[tuple[Path, int, int, int]]: 730 """references answer -> [(file, 1-based line, start column, end column)].""" 731 out = [] 732 for loc in result or []: 733 try: 734 rng = loc["range"] 735 out.append((from_uri(loc["uri"]), rng["start"]["line"] + 1, rng["start"]["character"], 736 rng["end"]["character"] if rng["end"]["line"] == rng["start"]["line"] 737 else rng["start"]["character"])) 738 except (KeyError, TypeError): 739 continue 740 return sorted(set(out), key=lambda r: (str(r[0]), r[1], r[2]))
references answer -> [(file, 1-based line, start column, end column)].
743def workspace_edits(result) -> dict[Path, list[TextEdit]]: 744 """rename answer (a WorkspaceEdit) -> {file: [edits]}.""" 745 if not isinstance(result, dict): 746 return {} 747 pairs = list((result.get("changes") or {}).items()) 748 for change in result.get("documentChanges") or []: 749 if isinstance(change, dict) and "textDocument" in change: 750 pairs.append((change["textDocument"]["uri"], change.get("edits", []))) 751 out: dict[Path, list[TextEdit]] = {} 752 for uri, edits in pairs: 753 for e in edits or []: 754 try: 755 r = e["range"] 756 out.setdefault(from_uri(uri), []).append(TextEdit( 757 r["start"]["line"], r["start"]["character"], r["end"]["line"], r["end"]["character"], 758 e["newText"])) 759 except (KeyError, TypeError): 760 continue 761 return out
rename answer (a WorkspaceEdit) -> {file: [edits]}.
764def apply_edits(text: str, edits: list[TextEdit]) -> str: 765 """Apply LSP text edits to a text ("\\n" line ends), last first so the 766 earlier positions stay valid.""" 767 lines = text.split("\n") 768 starts = [0] 769 for line in lines: 770 starts.append(starts[-1] + len(line) + 1) 771 772 def offset(line: int, column: int) -> int: 773 line = min(max(line, 0), len(lines) - 1) 774 return starts[line] + min(column, len(lines[line])) 775 776 for e in sorted(edits, key=lambda e: (e.line, e.column), reverse=True): 777 text = text[:offset(e.line, e.column)] + e.new_text + text[offset(e.end_line, e.end_column):] 778 return text
Apply LSP text edits to a text ("\n" line ends), last first so the earlier positions stay valid.
781def add_result_variable_edits(edits: dict[Path, list[TextEdit]], old: str, new: str, 782 file_text: Callable[[Path], str]) -> dict[Path, list[TextEdit]]: 783 """fortls's rename misses a function's result variable: in a function 784 without a ``result(...)`` clause, ``old = ...`` inside the body also 785 names the function and must be renamed too. Adds those edits.""" 786 header = re.compile(rf"\bfunction\s+{re.escape(old)}\s*\(", re.IGNORECASE) 787 word = re.compile(rf"\b{re.escape(old)}\b", re.IGNORECASE) 788 end = re.compile(r"^\s*(end\s*(function\b|$)|contains\b)", re.IGNORECASE) 789 out = {} 790 for path, file_edits in edits.items(): 791 try: 792 lines = file_text(path).split("\n") 793 except OSError: 794 out[path] = file_edits 795 continue 796 have = {(e.line, e.column) for e in file_edits} 797 added = list(file_edits) 798 for i, line in enumerate(lines): 799 code = "".join(line[a:b] for a, b in code_spans(line)) 800 if not header.search(code): 801 continue 802 # The whole statement (it may continue over "&" lines). 803 statement, j = code, i 804 while statement.rstrip().endswith("&") and j + 1 < len(lines): 805 j += 1 806 statement += "".join(lines[j][a:b] for a, b in code_spans(lines[j])) 807 if re.search(r"\bresult\s*\(", statement, re.IGNORECASE): 808 continue # the result has its own name 809 for k in range(j + 1, len(lines)): 810 if end.match(lines[k]): 811 break 812 for a, b in code_spans(lines[k]): 813 for match in word.finditer(lines[k], a, b): 814 if (k, match.start()) not in have: 815 have.add((k, match.start())) 816 added.append(TextEdit(k, match.start(), k, match.end(), new)) 817 out[path] = added 818 return out
fortls's rename misses a function's result variable: in a function
without a result(...) clause, old = ... inside the body also
names the function and must be renamed too. Adds those edits.