"""Console Panel -- Interactive Python console with log capture.
Provides a read-only output area for log messages and print statements,
a single-line input for executing Python commands, command history
navigation via up/down arrows, and a clear button.
"""
import logging
import sys
import time
from dataclasses import dataclass
from simvx.core import (
Button,
CheckBox,
Control,
Node,
Signal,
TextEdit,
Vec2,
)
from simvx.core.ui.core import FocusMode
from simvx.core.ui.multiline import MultiLineTextEdit
# ---------------------------------------------------------------------------
# ConsoleRecord -- canonical per-line entry retained by the panel
# ---------------------------------------------------------------------------
[docs]
@dataclass(slots=True)
class ConsoleRecord:
"""Canonical record stored alongside each rendered output line.
One record exists per entry in ``ConsolePanel._output_lines`` so the
upcoming filter UI (G2) can decide visibility per line. Multi-line log
messages (e.g. tracebacks) produce one record per visible line, all
sharing the originating ``level``/``source``/``created`` values.
``source`` distinguishes between the standard ``logging`` channel
(``"log"``) and writes that came through ``sys.stdout`` (``"stdout"``);
G2 will surface this as a filter facet. ``stdout`` entries are stored
as synthetic ``INFO``-level records -- there is no real ``LogRecord``
for them, so we use a single dataclass for all entries to keep one
canonical representation.
"""
level: str # logging level name: DEBUG/INFO/WARNING/ERROR/CRITICAL
name: str # logger name, or "stdout" / "console"
message: str # the rendered line text (already prefixed)
created: float # epoch seconds, matching logging.LogRecord.created
source: str # "log" | "stdout" | "console" | "script_error"
# ---------------------------------------------------------------------------
# Log handler that feeds into the console output
# ---------------------------------------------------------------------------
class _ConsoleLogHandler(logging.Handler):
"""Logging handler that appends formatted records to the console panel."""
def __init__(self, console: ConsolePanel):
super().__init__()
self._console = console
def emit(self, record: logging.LogRecord):
try:
level = record.levelname
if level in ("ERROR", "CRITICAL"):
prefix = "[ERROR]"
elif level == "WARNING":
prefix = "[WARN]"
else:
prefix = "[INFO]"
msg = self.format(record)
text = f"{prefix} {msg}"
template = ConsoleRecord(
level=level,
name=record.name,
message=text,
created=record.created,
source="log",
)
self._console._append_output(text, record=template)
except Exception:
self.handleError(record)
# ---------------------------------------------------------------------------
# Stdout redirect that copies writes into the console output
# ---------------------------------------------------------------------------
class _StdoutCapture:
"""File-like wrapper that tees writes to both the real stdout and a console."""
def __init__(self, console: ConsolePanel, real_stdout):
self._console = console
self._real = real_stdout
def write(self, text: str):
if self._real:
self._real.write(text)
if text and text.strip():
for line in text.rstrip("\n").split("\n"):
self._console._append_output(
line,
record=ConsoleRecord(
level="INFO",
name="stdout",
message=line,
created=time.time(),
source="stdout",
),
)
def flush(self):
if self._real:
self._real.flush()
# Needed so `sys.stdout` acts file-like
def fileno(self):
return self._real.fileno() if self._real else -1
def isatty(self):
return False
# ---------------------------------------------------------------------------
# ConsolePanel
# ---------------------------------------------------------------------------
[docs]
class ConsolePanel(Control):
"""Interactive Python console with log output and command input.
Features:
- Read-only output area showing log and print output
- Single-line input with enter-to-execute
- Command history (up/down arrow)
- eval/exec execution with shared namespace
- Logging handler that captures Python log messages
- Optional sys.stdout redirect for print() capture
- 1000-line output cap (oldest lines discarded)
- Clear button to reset output
"""
def __init__(self, editor_state=None, capture_stdout: bool = False,
**kwargs):
super().__init__(**kwargs)
self.state = editor_state
self.bg_colour = (0.12, 0.12, 0.12, 1.0)
self.size = Vec2(600, 200)
# Output buffer. ``_records`` is kept strictly in lockstep with
# ``_output_lines`` -- one ConsoleRecord per visible line. The
# filter toolbar reads ``records`` to decide visibility per line.
self._output_lines: list[str] = []
self._records: list[ConsoleRecord] = []
self._max_lines = 1000
# Filter state -- applied at draw time (one canonical filter):
# * level checkboxes (default DEBUG off, others on);
# the "ERROR" checkbox covers both ``ERROR`` and ``CRITICAL``
# records since users do not distinguish those two in practice.
# * a "Stdout" checkbox toggles ``source == "stdout"`` lines;
# script-error lines stay visible because their source differs.
# * a search box does case-insensitive substring matching on the
# rendered line text. Empty string disables the search filter.
# Filtering never mutates ``_output_lines`` / ``_records``; the
# rendered ``_output.text`` is rebuilt from a derived index list.
self._filter_levels: dict[str, bool] = {
"DEBUG": False,
"INFO": True,
"WARNING": True,
"ERROR": True,
}
self._show_stdout: bool = True
self._search_text: str = ""
# Maps a row index in the rendered output back to the original
# ``_output_lines`` / ``_records`` index. Rebuilt by ``_apply_filter``.
self._visible_to_original: list[int] = []
# Navigation: output-line index -> (file_path, line_number) so error
# tracebacks can be clicked in the console to jump to source.
self._clickable_lines: dict[int, tuple[str, int]] = {}
# Command history
self._history: list[str] = []
self._history_index = 0
# Input text tracked separately for history navigation
self._input_text = ""
# Severity tracking: 0=none, 1=info, 2=warn, 3=error
self._worst_severity: int = 0
self.severity_changed = Signal()
# Execution namespace
self._namespace: dict = {"__builtins__": __builtins__}
if self.state:
self._namespace["editor"] = self.state
if hasattr(self.state, "edited_scene"):
self._namespace["scene"] = self.state.edited_scene
# Child widgets ------------------------------------------------
# Output area (read-only multi-line)
self._output = MultiLineTextEdit()
self._output.read_only = True
self._output.show_line_numbers = False
self._output.bg_colour = (0.08, 0.08, 0.08, 1.0)
self._output.text_colour = (0.85, 0.85, 0.85, 1.0)
self._output.border_colour = (0.2, 0.2, 0.2, 1.0)
# Explicit focus_mode=CLICK so mouse clicks establish focus, which
# enables keyboard shortcuts (ctrl+c/ctrl+a) forwarded from the panel
# and shift+arrow keyboard selection.
self._output.focus_mode = FocusMode.CLICK
# Input line (single-line)
self._input = TextEdit(placeholder="Enter command...")
self._input.bg_colour = (0.1, 0.1, 0.1, 1.0)
self._input.text_colour = (1.0, 1.0, 1.0, 1.0)
self._input.border_colour = (0.3, 0.3, 0.3, 1.0)
self._input.text_submitted.connect(self._on_input_submitted)
# Clear button
self._clear_btn = Button(text="Clear", on_press=self.clear)
self._clear_btn.size = Vec2(48, 24)
# Filter toolbar (level checkboxes + search box). The "WARN" label
# matches the engine-wide compact prefix; the underlying record
# level is "WARNING". The ERROR checkbox folds CRITICAL records
# under the same toggle (see ``_record_passes_filter``).
self._level_checkboxes: dict[str, CheckBox] = {}
for label, level_key in (
("DEBUG", "DEBUG"),
("INFO", "INFO"),
("WARN", "WARNING"),
("ERROR", "ERROR"),
):
cb = CheckBox(label, checked=self._filter_levels[level_key])
cb.toggled.connect(self._make_level_toggler(level_key))
self._level_checkboxes[level_key] = cb
self._stdout_checkbox = CheckBox("Stdout", checked=self._show_stdout)
self._stdout_checkbox.toggled.connect(self._on_stdout_toggled)
self._search_box = TextEdit(placeholder="Search...")
self._search_box.bg_colour = (0.1, 0.1, 0.1, 1.0)
self._search_box.text_colour = (1.0, 1.0, 1.0, 1.0)
self._search_box.border_colour = (0.3, 0.3, 0.3, 1.0)
self._search_box.size = Vec2(160, 22)
self._search_box.text_changed.connect(self._on_search_changed)
# Logging handler
self._log_handler = _ConsoleLogHandler(self)
self._log_handler.setFormatter(logging.Formatter("%(name)s: %(message)s"))
logging.getLogger().addHandler(self._log_handler)
# Optional stdout capture
self._real_stdout = sys.stdout
self._capturing_stdout = False
if capture_stdout:
self.start_stdout_capture()
# Connection to Node.script_error_raised is established in enter_tree()
# and disconnected in exit_tree() to survive IDE bridge panel re-injection.
self._error_connected = False
# Welcome message
self._append_output("[INFO] Console ready. Type Python expressions below.")
# ------------------------------------------------------------------
# Public helpers
# ------------------------------------------------------------------
[docs]
@property
def worst_severity(self) -> int:
return self._worst_severity
[docs]
def reset_severity(self):
"""Reset severity tracking (call when Console tab is viewed)."""
self._worst_severity = 0
self.severity_changed.emit()
def _on_script_error(self, node, method_name: str, tb: str):
"""Handle runtime script errors from Node._safe_call."""
err_template = ConsoleRecord(
level="ERROR",
name=node.name,
message="",
created=time.time(),
source="script_error",
)
self._append_output(
f"[ERROR] {node.name}.{method_name}() — script error:",
record=err_template,
)
for line in tb.strip().splitlines():
self._append_output(f" {line}", record=err_template)
# Extract the last user frame so the reader can click to open it.
from ..error_nav import parse_traceback_source
project_path = None
if self.state is not None:
project_path = getattr(self.state, "project_path", None)
source = parse_traceback_source(tb, project_path)
if source is not None:
file_path, line_num = source
display_path = file_path
if project_path is not None:
try:
from pathlib import Path
display_path = str(Path(file_path).resolve().relative_to(
Path(project_path).resolve()))
except (OSError, ValueError):
pass
nav_text = f" \u00bb Open {display_path}:{line_num} (click to jump)"
self._append_output(nav_text, record=err_template)
# Record the clickable line index (the line we just appended).
nav_index = len(self._output_lines) - 1
self._clickable_lines[nav_index] = (file_path, line_num)
if self._worst_severity < 3:
self._worst_severity = 3
self.severity_changed.emit()
[docs]
def start_stdout_capture(self):
"""Begin capturing sys.stdout into the console output."""
if not self._capturing_stdout:
self._real_stdout = sys.stdout
sys.stdout = _StdoutCapture(self, self._real_stdout)
self._capturing_stdout = True
[docs]
def stop_stdout_capture(self):
"""Restore sys.stdout to its original value."""
if self._capturing_stdout:
sys.stdout = self._real_stdout
self._capturing_stdout = False
[docs]
def inject(self, name: str, value):
"""Add a name into the console execution namespace."""
self._namespace[name] = value
[docs]
def clear(self):
"""Clear all output lines."""
self._output_lines.clear()
self._records.clear()
self._clickable_lines.clear()
self._visible_to_original = []
self._output.text = ""
[docs]
@property
def records(self) -> list[ConsoleRecord]:
"""Return the parallel list of ConsoleRecord entries.
Always the same length as ``_output_lines``; index ``i`` describes
the origin/level of the line at ``_output_lines[i]``.
"""
return self._records
[docs]
def record_at(self, index: int) -> ConsoleRecord:
"""Return the ConsoleRecord for output line *index*."""
return self._records[index]
# ------------------------------------------------------------------
# Filter
# ------------------------------------------------------------------
def _make_level_toggler(self, level_key: str):
def _toggle(checked: bool) -> None:
self._filter_levels[level_key] = checked
self._apply_filter()
return _toggle
def _on_stdout_toggled(self, checked: bool) -> None:
self._show_stdout = checked
self._apply_filter()
def _on_search_changed(self, text: str) -> None:
self._search_text = text
self._apply_filter()
def _record_passes_filter(self, record: ConsoleRecord) -> bool:
"""Return True if *record* should be visible under the current filter."""
# Level: CRITICAL is folded under the ERROR checkbox; unknown
# levels (e.g. ``console`` synthetic INFO records) are governed by
# their own ``record.level`` key when present in the table.
level = record.level
if level == "CRITICAL":
level_key = "ERROR"
elif level in self._filter_levels:
level_key = level
else:
level_key = "INFO"
if not self._filter_levels.get(level_key, True):
return False
# Source: stdout-only toggle. Other sources are always allowed.
if record.source == "stdout" and not self._show_stdout:
return False
# Search: case-insensitive substring on the rendered line text.
if self._search_text and self._search_text.casefold() not in record.message.casefold():
return False
return True
def _apply_filter(self) -> None:
"""Recompute the visible-rows projection and refresh the output text."""
visible: list[int] = []
rendered: list[str] = []
for i, record in enumerate(self._records):
if self._record_passes_filter(record):
visible.append(i)
rendered.append(self._output_lines[i])
self._visible_to_original = visible
self._output.text = "\n".join(rendered)
[docs]
@property
def visible_records(self) -> list[ConsoleRecord]:
"""Return records currently passing the filter (in display order)."""
return [self._records[i] for i in self._visible_to_original]
# ------------------------------------------------------------------
# Error navigation
# ------------------------------------------------------------------
def _line_index_at(self, screen_pos) -> int | None:
"""Return the original output-line index under *screen_pos*, or None.
The rendered output only shows lines that pass the current filter,
so the row index computed from screen Y is mapped back through
``_visible_to_original`` to get the underlying ``_output_lines``
index. Returns ``None`` when no row is hit.
"""
if not self._visible_to_original:
return None
out = self._output
_, y, _, _ = out.get_global_rect()
py = screen_pos.y if hasattr(screen_pos, "y") else screen_pos[1]
# Matches MultiLineTextEdit._pos_to_cursor: y-offset / line_height + scroll
_PADDING_TOP = 4.0
lh = out._line_height()
if lh <= 0:
return None
row = int((py - y - _PADDING_TOP) / lh + out._scroll_y)
if row < 0 or row >= len(self._visible_to_original):
return None
return self._visible_to_original[row]
def _open_source(self, file_path: str, line_num: int) -> None:
"""Open *file_path* in the script workspace at *line_num* (1-indexed)."""
workspace = getattr(self.state, "workspace", None) if self.state else None
if workspace is None or not hasattr(workspace, "open_file"):
return
workspace.open_file(file_path, line=line_num)
# ------------------------------------------------------------------
# Output management
# ------------------------------------------------------------------
def _append_output(self, text: str, *, record: ConsoleRecord | None = None):
"""Append a line to the output buffer, enforcing max_lines.
If ``record`` is omitted a synthetic ``console`` record is generated
so ``self._records`` stays in lockstep with ``self._output_lines``.
Multi-line ``text`` produces one record per resulting line, all
sharing the same metadata (level/source/created/name).
"""
for line in text.split("\n"):
self._output_lines.append(line)
if record is None:
line_record = ConsoleRecord(
level="INFO",
name="console",
message=line,
created=time.time(),
source="console",
)
else:
# Re-use the supplied metadata but store the per-line message.
line_record = ConsoleRecord(
level=record.level,
name=record.name,
message=line,
created=record.created,
source=record.source,
)
self._records.append(line_record)
# When rolling off old lines, shift clickable-line indices to match
# and drop the corresponding records.
while len(self._output_lines) > self._max_lines:
self._output_lines.pop(0)
self._records.pop(0)
if self._clickable_lines:
shifted: dict[int, tuple[str, int]] = {}
for idx, payload in self._clickable_lines.items():
if idx > 0:
shifted[idx - 1] = payload
self._clickable_lines = shifted
self._apply_filter()
# ------------------------------------------------------------------
# Command execution
# ------------------------------------------------------------------
def _execute_command(self, cmd: str):
"""Execute a command string via eval/exec in the console namespace."""
self._history.append(cmd)
self._history_index = len(self._history)
self._append_output(f">>> {cmd}")
try:
result = eval(cmd, self._namespace)
if result is not None:
self._append_output(repr(result))
except SyntaxError:
try:
exec(cmd, self._namespace)
except Exception as e:
self._append_output(
f"[ERROR] {e}",
record=ConsoleRecord(
level="ERROR", name="console", message="",
created=time.time(), source="console",
),
)
except Exception as e:
self._append_output(
f"[ERROR] {e}",
record=ConsoleRecord(
level="ERROR", name="console", message="",
created=time.time(), source="console",
),
)
# ------------------------------------------------------------------
# Input callbacks
# ------------------------------------------------------------------
def _on_input_submitted(self, text: str):
"""Called when the user presses Enter in the input line."""
cmd = text.strip()
if cmd:
self._execute_command(cmd)
self._input.text = ""
self._input.cursor_pos = 0
self._input_text = ""
# ------------------------------------------------------------------
# GUI input (history navigation via up/down)
# ------------------------------------------------------------------
def _on_gui_input(self, event):
"""Handle scroll, up/down arrow for command history, delegate rest to children."""
# Forward scroll events to the output area
if event.key in ("scroll_up", "scroll_down"):
self._output._on_gui_input(event)
return
# Forward mouse events to the output area for text selection
if self._output._dragging_text or self._output._dragging_scrollbar:
self._output._on_gui_input(event)
return
if event.button == 1 and event.pressed and event.position:
# Filter toolbar checkboxes consume the click first.
for cb in (*self._level_checkboxes.values(), self._stdout_checkbox):
if cb.is_point_inside(event.position):
cb._on_gui_input(event)
return
# Search box gains focus and handles caret placement.
if self._search_box.is_point_inside(event.position):
self._search_box.focused = True
self._input.focused = False
self._output.focused = False
self._search_box._on_gui_input(event)
return
if self._output.is_point_inside(event.position):
# Check for error-navigation click: tracebacks include a "»
# Open …:N" line that maps to (file, line). Navigate if hit.
line_idx = self._line_index_at(event.position)
if line_idx is not None and line_idx in self._clickable_lines:
file_path, line_num = self._clickable_lines[line_idx]
self._open_source(file_path, line_num)
return
self._search_box.focused = False
self._output._on_gui_input(event)
return
elif self._input.is_point_inside(event.position):
self._search_box.focused = False
self._input._on_gui_input(event)
return
if event.button == 1 and not event.pressed:
self._output._on_gui_input(event)
self._input._on_gui_input(event)
self._search_box._on_gui_input(event)
return
# Keyboard / character events while the search box has focus.
if self._search_box.focused:
self._search_box._on_gui_input(event)
return
# Forward selection/copy shortcuts to the output area when it has focus
if self._output.focused and event.key in ("ctrl+c", "ctrl+a", "ctrl+d"):
self._output._on_gui_input(event)
return
# Arrow keys / shift+arrow for output text selection when output is focused
if self._output.focused and event.key in ("left", "right", "up", "down", "home", "end"):
self._output._on_gui_input(event)
return
if not self._input.focused:
return
if event.key == "up" and event.pressed:
if self._history and self._history_index > 0:
# Save current input if starting to navigate
if self._history_index == len(self._history):
self._input_text = self._input.text
self._history_index -= 1
self._input.text = self._history[self._history_index]
self._input.cursor_pos = len(self._input.text)
return
if event.key == "down" and event.pressed:
if self._history_index < len(self._history):
self._history_index += 1
if self._history_index == len(self._history):
self._input.text = self._input_text
else:
self._input.text = self._history[self._history_index]
self._input.cursor_pos = len(self._input.text)
return
# ------------------------------------------------------------------
# Layout
# ------------------------------------------------------------------
# Header geometry shared by ``_layout_children`` and ``draw`` so both
# see the same Y offsets when the toolbar height changes.
_TITLE_H = 26.0
_TOOLBAR_H = 26.0
def _layout_children(self):
"""Position output area, input line, clear button, and filter toolbar."""
x, y, w, h = self.get_global_rect()
title_h = self._TITLE_H
toolbar_h = self._TOOLBAR_H
header_h = title_h + toolbar_h
input_h = 30.0
pad = 2.0
# Clear button in title strip (right-aligned)
btn_w, btn_h = self._clear_btn.size[0], self._clear_btn.size[1]
self._clear_btn.position = Vec2(
x + w - btn_w - pad, y + (title_h - btn_h) / 2
)
# Filter toolbar row (just below the title strip, left-aligned).
# Each widget keeps its content-driven size; we just place them.
cursor_x = x + 4.0
toolbar_y = y + title_h
for level_key in ("DEBUG", "INFO", "WARNING", "ERROR"):
cb = self._level_checkboxes[level_key]
cb.position = Vec2(cursor_x, toolbar_y + (toolbar_h - cb.size[1]) / 2)
cursor_x += cb.size[0] + 6.0
# Small gap, then stdout toggle.
cursor_x += 4.0
self._stdout_checkbox.position = Vec2(
cursor_x, toolbar_y + (toolbar_h - self._stdout_checkbox.size[1]) / 2
)
cursor_x += self._stdout_checkbox.size[0] + 8.0
# Search box: stretches to fill the rest of the row, capped at 240px.
search_w = max(80.0, min(240.0, x + w - pad - cursor_x))
self._search_box.size = Vec2(search_w, self._search_box.size[1])
self._search_box.position = Vec2(
cursor_x, toolbar_y + (toolbar_h - self._search_box.size[1]) / 2
)
# Output area fills the middle
out_y = y + header_h
out_h = h - header_h - input_h - pad
self._output.position = Vec2(x, out_y)
self._output.size = Vec2(w, max(out_h, 20.0))
# Input line at the bottom
self._input.position = Vec2(x, y + h - input_h)
self._input.size = Vec2(w, input_h)
# ------------------------------------------------------------------
# Process / Draw
# ------------------------------------------------------------------
[docs]
def process(self, dt: float):
self._output.process(dt)
self._input.process(dt)
self._search_box.process(dt)
[docs]
def draw(self, renderer):
self._layout_children()
x, y, w, h = self.get_global_rect()
# Panel background
renderer.draw_rect((x, y), (w, h), colour=self.bg_colour, filled=True)
# Title strip (top of header)
title_h = self._TITLE_H
toolbar_h = self._TOOLBAR_H
header_h = title_h + toolbar_h
renderer.draw_rect((x, y), (w, title_h), colour=(0.16, 0.16, 0.16, 1.0), filled=True)
scale = 11.0 / 14.0
renderer.draw_text("Console", (x + 8, y + 5), colour=(0.7, 0.7, 0.7, 1.0), scale=scale)
# Filter toolbar strip (slightly darker than the title)
renderer.draw_rect(
(x, y + title_h), (w, toolbar_h), colour=(0.13, 0.13, 0.13, 1.0), filled=True
)
# Header bottom border
renderer.draw_line(
(x, y + header_h), (x + w, y + header_h), colour=(0.25, 0.25, 0.25, 1.0)
)
# Clear button
self._clear_btn.draw(renderer)
# Filter toolbar widgets
for level_key in ("DEBUG", "INFO", "WARNING", "ERROR"):
self._level_checkboxes[level_key].draw(renderer)
self._stdout_checkbox.draw(renderer)
self._search_box.draw(renderer)
# Output area
self._output.draw(renderer)
# Input prompt indicator
input_y = y + h - 30.0
renderer.draw_text(">>>", (x + 4, input_y + 7), colour=(0.5, 0.5, 0.5, 1.0), scale=scale)
# Input line (offset to leave room for prompt)
self._input.draw(renderer)
# ------------------------------------------------------------------
# Cleanup
# ------------------------------------------------------------------
[docs]
def enter_tree(self):
"""Re-attach log handler and error signal when (re-)entering the tree."""
if not self._error_connected:
Node.script_error_raised.connect(self._on_script_error)
self._error_connected = True
# Re-add log handler (may have been removed by exit_tree during panel re-injection)
root_logger = logging.getLogger()
if self._log_handler not in root_logger.handlers:
root_logger.addHandler(self._log_handler)
[docs]
def exit_tree(self):
"""Remove log handler, disconnect error signal, and restore stdout on removal from tree."""
logging.getLogger().removeHandler(self._log_handler)
if self._error_connected:
Node.script_error_raised.disconnect(self._on_script_error)
self._error_connected = False
self.stop_stdout_capture()