"""IDE state management -- central signal bus for cross-component communication.
All IDE components connect to these signals rather than referencing each other.
Uses simvx.core.Signal (not Qt signals).
"""
import logging
from collections.abc import Callable
from dataclasses import dataclass
from pathlib import Path
from simvx.core import Signal
from simvx.core.document import BufferRegistry
from simvx.core.file_state import FileStateMixin
log = logging.getLogger(__name__)
[docs]
@dataclass
class Diagnostic:
"""Single diagnostic (error/warning) from LSP or linter."""
path: str
line: int
col_start: int
col_end: int
severity: int # 1=Error, 2=Warning, 3=Info, 4=Hint
message: str
source: str = ""
code: str = ""
[docs]
class State(FileStateMixin):
"""Single source of truth for IDE-wide state.
Signals:
file_opened(path: str)
file_closed(path: str)
file_saved(path: str)
active_file_changed(path: str)
cursor_moved(line: int, col: int)
diagnostics_updated(path: str, diagnostics: list[Diagnostic])
goto_requested(path: str, line: int, col: int)
status_message(message: str)
sidebar_toggled(visible: bool)
bottom_panel_toggled(visible: bool)
run_requested(path: str)
debug_started()
debug_stopped()
breakpoint_toggled(path: str, line: int)
completion_requested(path: str, line: int, col: int)
completion_received(items: list)
hover_received(text: str, line: int, col: int)
format_requested(path: str)
"""
def __init__(
self,
*,
file_opened: Signal | None = None,
file_closed: Signal | None = None,
file_saved: Signal | None = None,
active_file_changed: Signal | None = None,
):
# File lifecycle signals -- shared with State when embedded
self._init_file_signals(
file_opened=file_opened,
file_closed=file_closed,
file_saved=file_saved,
active_file_changed=active_file_changed,
)
# Editor events
self.cursor_moved = Signal()
# LSP / Diagnostics
self.diagnostics_updated = Signal()
self.completion_requested = Signal()
self.completion_received = Signal()
self.hover_received = Signal()
self.definition_received = Signal()
self.references_received = Signal()
self.format_requested = Signal()
self.rename_edits_received = Signal() # emits dict[str, list[tuple]]
self.formatting_edits_received = Signal() # emits (path: str, edits: list[tuple])
# Navigation
self.goto_requested = Signal()
# Run / Debug
self.run_requested = Signal()
self.debug_started = Signal()
self.debug_stopped = Signal()
self.debug_output = Signal()
self.debug_state_changed = Signal()
self.breakpoint_toggled = Signal()
self.bookmark_toggled = Signal()
self.exception_occurred = Signal()
# UI state
self.status_message = Signal()
self.sidebar_toggled = Signal()
self.bottom_panel_toggled = Signal()
# Document management
self.buffers = BufferRegistry()
# Optional callback returning the set of currently-dirty file paths.
# Wired by the standalone IDE app to ``CodeEditorPanel.modified_files``;
# consumed by the file browser's GitStatusProvider to colour unsaved files.
self.dirty_paths_provider: Callable[[], set[Path]] | None = None
# Private state
self._project_root: str = ""
self._active_file: str = ""
self._cursor_line: int = 0
self._cursor_col: int = 0
self._diagnostics: dict[str, list[Diagnostic]] = {}
self._breakpoints: dict[str, set[int]] = {}
self._bookmarks: dict[str, set[int]] = {}
self._cursor_history: list[tuple[str, int, int]] = []
self._history_pos: int = -1
# -- Properties ------------------------------------------------------------
@property
def project_root(self) -> str:
return self._project_root
[docs]
@project_root.setter
def project_root(self, path: str):
self._project_root = path
@property
def active_file(self) -> str:
return self._active_file
[docs]
@active_file.setter
def active_file(self, path: str):
if path != self._active_file:
self._active_file = path
self.active_file_changed.emit(path)
[docs]
@property
def cursor_line(self) -> int:
return self._cursor_line
[docs]
@property
def cursor_col(self) -> int:
return self._cursor_col
# -- Cursor ----------------------------------------------------------------
[docs]
def set_cursor(self, line: int, col: int):
self._cursor_line = line
self._cursor_col = col
self.cursor_moved.emit(line, col)
# -- Diagnostics -----------------------------------------------------------
[docs]
def set_diagnostics(self, path: str, diagnostics: list[Diagnostic]):
self._diagnostics[path] = diagnostics
self.diagnostics_updated.emit(path, diagnostics)
[docs]
def get_diagnostics(self, path: str) -> list[Diagnostic]:
return self._diagnostics.get(path, [])
[docs]
@property
def all_diagnostics(self) -> dict[str, list[Diagnostic]]:
return dict(self._diagnostics)
# -- Breakpoints -----------------------------------------------------------
[docs]
def toggle_breakpoint(self, path: str, line: int):
bp = self._breakpoints.setdefault(path, set())
if line in bp:
bp.discard(line)
else:
bp.add(line)
self.breakpoint_toggled.emit(path, line)
[docs]
def get_breakpoints(self, path: str) -> set[int]:
return self._breakpoints.get(path, set())
[docs]
@property
def all_breakpoints(self) -> dict[str, set[int]]:
return {p: set(lines) for p, lines in self._breakpoints.items() if lines}
# -- Bookmarks -------------------------------------------------------------
[docs]
def toggle_bookmark(self, path: str, line: int):
bm = self._bookmarks.setdefault(path, set())
if line in bm:
bm.discard(line)
else:
bm.add(line)
self.bookmark_toggled.emit(path, line)
[docs]
def get_bookmarks(self, path: str) -> set[int]:
return self._bookmarks.get(path, set())
[docs]
@property
def all_bookmarks(self) -> dict[str, set[int]]:
return {p: set(lines) for p, lines in self._bookmarks.items() if lines}
# -- Cursor History --------------------------------------------------------
[docs]
def push_cursor_history(self, path: str, line: int, col: int = 0):
"""Push current location onto history stack (for Alt+Left/Right navigation)."""
entry = (path, line, col)
# Don't push duplicates
if self._cursor_history and self._history_pos >= 0:
if self._cursor_history[self._history_pos] == entry:
return
# Truncate forward history when pushing new entry
self._cursor_history = self._cursor_history[: self._history_pos + 1]
self._cursor_history.append(entry)
if len(self._cursor_history) > 100:
self._cursor_history = self._cursor_history[-100:]
self._history_pos = len(self._cursor_history) - 1
[docs]
def history_back(self) -> tuple[str, int, int] | None:
"""Navigate backward in cursor history. Returns (path, line, col) or None."""
if self._history_pos > 0:
self._history_pos -= 1
return self._cursor_history[self._history_pos]
return None
[docs]
def history_forward(self) -> tuple[str, int, int] | None:
"""Navigate forward in cursor history. Returns (path, line, col) or None."""
if self._history_pos < len(self._cursor_history) - 1:
self._history_pos += 1
return self._cursor_history[self._history_pos]
return None
# -- Navigation ------------------------------------------------------------
[docs]
def goto(self, path: str, line: int, col: int = 0):
self.goto_requested.emit(path, line, col)
# -- Dirty buffers ---------------------------------------------------------
[docs]
def dirty_paths(self) -> set[Path]:
"""Return resolved paths of open buffers with unsaved changes.
Routes through ``dirty_paths_provider`` (wired by the IDE app to
``CodeEditorPanel.modified_files``). Wrapped in try/except: this
runs in the GitStatusProvider polling thread; an exception there would
kill the poller, so we swallow and log instead.
"""
provider = self.dirty_paths_provider
if provider is None:
return set()
try:
return {Path(p).resolve() for p in provider()}
except Exception:
log.warning("dirty_paths: provider raised", exc_info=True)
return set()
# -- Helpers ---------------------------------------------------------------
[docs]
def relative_path(self, path: str) -> str:
if self._project_root and path.startswith(self._project_root):
return str(Path(path).relative_to(self._project_root))
return Path(path).name