"""LSP integration controller -- server lifecycle, document sync, and response handling."""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from .app import IDERoot
log = logging.getLogger(__name__)
[docs]
class LSPController:
"""Manages the LSP client lifecycle and document synchronisation.
Also handles lint runner startup and format/lint-on-save hooks.
"""
def __init__(self, ide: IDERoot) -> None:
self._ide = ide
# -- Convenience accessors ------------------------------------------------
@property
def _lsp_client(self):
return self._ide._lsp_client
@_lsp_client.setter
def _lsp_client(self, value):
self._ide._lsp_client = value
@property
def _editor_panel(self):
return self._ide._editor_panel
@property
def _status_bar(self):
return self._ide._status_bar
@property
def _lint_runner(self):
return self._ide._lint_runner
@_lint_runner.setter
def _lint_runner(self, value):
self._ide._lint_runner = value
@property
def _output_panel(self):
return self._ide._output_panel
@property
def state(self):
return self._ide.state
@property
def config(self):
return self._ide.config
# -- Server lifecycle -----------------------------------------------------
[docs]
def start_lsp(self):
if not self.config.lsp_enabled:
if self._status_bar:
self._status_bar.set_lsp_status("off")
return
try:
from .lsp.client import LSPClient
env = self.config.get_env(self.state.project_root)
self._lsp_client = LSPClient(
self.state,
command=self.config.lsp_command,
args=self.config.lsp_args,
env=env,
)
if self._status_bar:
self._status_bar.set_lsp_status("starting")
self._lsp_client.server_started.connect(self.on_lsp_ready)
self._lsp_client.server_stopped.connect(self.on_lsp_stopped)
self._lsp_client.start()
except (ImportError, Exception) as e:
log.debug("LSP client not available: %s", e)
if self._status_bar:
self._status_bar.set_lsp_status("error")
[docs]
def start_lint_runner(self):
try:
from .lint.runner import LintRunner
self._lint_runner = LintRunner(self.state, self.config)
except ImportError:
log.debug("LintRunner not available")
[docs]
def on_lsp_ready(self):
"""Called when LSP server completes initialization."""
if self._status_bar:
self._status_bar.set_lsp_status("ready")
self.state.status_message.emit(f"LSP ready ({self.config.lsp_command})")
[docs]
def on_lsp_stopped(self):
"""Called when LSP server shuts down."""
if self._status_bar:
self._status_bar.set_lsp_status("off" if not self.config.lsp_enabled else "error")
# -- Document notifications -----------------------------------------------
[docs]
def lsp_notify_open(self, path: str):
"""Notify LSP server when a file is opened."""
if not self._lsp_client or not path or path.startswith("untitled"):
return
if self._editor_panel:
text = self._editor_panel.get_file_text(path)
if text:
self._lsp_client.notify_open(path, text)
[docs]
def lsp_notify_close(self, path: str):
"""Notify LSP server when a file is closed."""
if self._lsp_client and path and not path.startswith("untitled"):
self._lsp_client.notify_close(path)
[docs]
def lsp_notify_save(self, path: str):
"""Notify LSP server when a file is saved."""
if not self._lsp_client or not path or path.startswith("untitled"):
return
if self._editor_panel:
text = self._editor_panel.get_file_text(path) or None
self._lsp_client.notify_save(path, text)
[docs]
def lsp_request_completion(self, path: str, line: int, col: int):
"""Request code completion from LSP server."""
if self._lsp_client and path and not path.startswith("untitled"):
self._lsp_client.request_completion(path, line, col)
# -- LSP response handlers ------------------------------------------------
[docs]
def apply_rename_edits(self, file_edits: dict[str, list[tuple[int, int, int, int, str]]]):
"""Apply rename edits across files. Opens files not yet open."""
if not self._editor_panel:
return
for path, edits in file_edits.items():
if not edits:
continue
if path not in self._editor_panel.get_open_files():
self._editor_panel.open_file(path)
editor = self._editor_panel.get_open_files().get(path, {}).get("editor")
if editor:
editor.apply_text_edits(edits)
[docs]
def on_references_received(self, locations: list):
"""Show references in the search/output panel."""
if not locations:
self.state.status_message.emit("No references found")
return
from .lsp.protocol import uri_to_path
self._ide._show_bottom_panel()
self._ide._switch_to_tab("Output")
if self._output_panel:
self._output_panel.clear()
self._output_panel.append(f"\033[1m{len(locations)} references found\033[0m\n")
for loc in locations:
path = uri_to_path(loc.uri)
short = self.state.relative_path(path)
line = loc.range.start.line + 1
col = loc.range.start.character
self._output_panel.append(f" {short}:{line}:{col}\n")
[docs]
def on_hover_received(self, text: str, line: int, col: int):
"""Show LSP hover info as a tooltip on the active editor."""
if not text or not self._editor_panel:
return
editor = self._editor_panel.get_current_editor()
if editor and hasattr(editor, "set_hover_tooltip"):
editor.set_hover_tooltip(text, line, col)
# -- Tools menu handlers --------------------------------------------------
[docs]
def on_toggle_lsp(self):
"""Enable/disable LSP."""
if self._lsp_client:
self._lsp_client.stop()
self._lsp_client = None
self.config.lsp_enabled = False
self.state.status_message.emit("LSP disabled")
else:
self.config.lsp_enabled = True
self.start_lsp()
self.state.status_message.emit("LSP enabled")
[docs]
def on_restart_lsp(self):
"""Stop and restart the LSP server."""
if self._lsp_client:
self._lsp_client.stop()
self._lsp_client = None
self.config.lsp_enabled = True
self.start_lsp()
self.state.status_message.emit("LSP server restarting...")
# -- Lint / format signal handlers ----------------------------------------
[docs]
def on_file_opened_lint(self, path: str):
if self._lint_runner and self.config.lint_enabled and path and path.endswith(".py"):
self._lint_runner.lint_file(path)
[docs]
def on_file_saved_signal(self, path: str):
if self._lint_runner and self.config.lint_enabled and path.endswith(".py"):
if self.config.format_on_save:
self._lint_runner.format_on_save(path)
if self.config.lint_on_save:
self._lint_runner.lint_on_save(path)
[docs]
def on_lint_file(self):
"""Manually lint the current file."""
path = self.state.active_file
if not path or path.startswith("untitled"):
self.state.status_message.emit("Save file first to lint")
return
if not self._lint_runner:
self.state.status_message.emit("Linter not available")
return
diags = self._lint_runner.lint_file(path)
count = len(diags) if diags else 0
self.state.status_message.emit(f"Lint: {count} issue{'s' if count != 1 else ''} found")
[docs]
def on_toggle_linting(self):
"""Enable/disable automatic linting."""
self.config.lint_enabled = not self.config.lint_enabled
status = "enabled" if self.config.lint_enabled else "disabled"
self.state.status_message.emit(f"Linting {status}")
if not self.config.lint_enabled:
for path in list(self.state.get_all_diagnostics()):
self.state.set_diagnostics(path, [])