"""Console Panel -- Interactive Python console with log capture.
Provides a read-only output area for log messages and print statements,
a command line for executing Python commands, with history on Up and
Down, a filter toolbar and a clear button.
"""
import logging
import sys
import time
from dataclasses import dataclass
from simvx.core import (
Button,
CheckBox,
CommandLine,
Container,
Label,
MultiLineTextEdit,
Pointer,
PointerFilter,
Property,
SceneTree,
ScrollEvent,
Signal,
TextEdit,
Vec2,
)
from simvx.core.decorators import on_gui_input
#: How many lines one wheel notch over the prompt or the toolbar scrolls the output.
_WHEEL_LINES = 3.0
#: The output view's top padding, where its first row starts.
_OUTPUT_PADDING_TOP = 4.0
# ---------------------------------------------------------------------------
# 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 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"``);
the filter toolbar surfaces 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
# ---------------------------------------------------------------------------
class _ConsoleOutput(MultiLineTextEdit):
"""The console's read-only output, where a click on a traceback's "Open" line opens its source.
*open_row* is called with the clicked row and returns whether that row
was a link it followed; a click on any other row selects text as usual.
"""
def __init__(self, open_row, **kwargs):
super().__init__(read_only=True, show_line_numbers=False, **kwargs)
self._open_row = open_row
def row_at(self, position) -> int | None:
"""The rendered row under the screen *position*, or None past the last one."""
lh = self._line_height()
row = int((self.to_local(position).y - _OUTPUT_PADDING_TOP) / lh + self._scroll_y)
return row if 0 <= row < len(self._lines) else None
@on_gui_input(Pointer.PRIMARY)
def _follow_link(self, event) -> bool:
row = self.row_at(event.position)
return row is not None and self._open_row(row)
[docs]
class ConsolePanel(Container):
"""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
"""
pointer_filter = Property(PointerFilter.STOP, hint="How this control takes part in the pointer hit test")
size_x = Property(600.0, range=(0, 10000), hint="Control width", on_change="_on_size_changed")
size_y = Property(200.0, range=(0, 10000), hint="Control height", on_change="_on_size_changed")
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)
# 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]] = {}
# 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): a click selects, Ctrl+C copies,
# and a click on a traceback's "Open" line jumps to the source.
self._output = self.add_child(_ConsoleOutput(self._open_row))
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)
self._prompt = self.add_child(Label(">>>", font_size=11.0))
self._prompt.text_colour = (0.5, 0.5, 0.5, 1.0)
self._input = self.add_child(CommandLine(placeholder="Enter command..."))
self._input.text_submitted.connect(self._on_input_submitted)
self._clear_btn = self.add_child(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 = self.add_child(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 = self.add_child(CheckBox("Stdout", checked=self._show_stdout))
self._stdout_checkbox.toggled.connect(self._on_stdout_toggled)
self._search_box = self.add_child(TextEdit(placeholder="Search..."))
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()
# Script errors come from the running game's tree, which play mode
# replaces on every start: _follow_game_tree tracks it while the panel
# is in the editor's tree.
self._error_tree: SceneTree | None = None
# 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 _follow_game_tree(self) -> None:
"""Listen for script errors on the game tree play mode is running, if any."""
play_mode = getattr(self.state, "play_mode", None)
self._watch_errors(play_mode.game_tree if play_mode is not None else None)
def _watch_errors(self, tree: SceneTree | None) -> None:
if tree is self._error_tree:
return
if self._error_tree is not None:
self._error_tree.script_error.disconnect(self._on_script_error)
self._error_tree = tree
if tree is not None:
tree.script_error.connect(self._on_script_error)
def _on_script_error(self, node, method_name: str, tb: str):
"""Report a script error raised in the game tree."""
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 _open_row(self, row: int) -> bool:
"""Open the source a traceback's "Open" line names, if rendered *row* is one; True when it was."""
if not 0 <= row < len(self._visible_to_original):
return False
target = self._clickable_lines.get(self._visible_to_original[row])
if target is None:
return False
self._open_source(*target)
return True
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._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
# ------------------------------------------------------------------
def _on_input_submitted(self, text: str):
"""Run the command the input line submitted."""
cmd = text.strip()
if cmd:
self._execute_command(cmd)
@on_gui_input(ScrollEvent)
def _scroll_output(self, event) -> bool:
"""The wheel over the prompt or the toolbar scrolls the output, as it does over the text."""
self._output.scroll_y -= _WHEEL_LINES if event.scroll[1] > 0 else -_WHEEL_LINES
return True
# ------------------------------------------------------------------
# Layout and drawing
# ------------------------------------------------------------------
_TITLE_H = 26.0
_TOOLBAR_H = 26.0
_INPUT_H = 30.0
_PAD = 2.0
def _update_layout(self):
"""Title strip with the Clear button, the filter toolbar, the output, then the prompt and input line."""
w, h = self.size_x, self.size_y
title_h, toolbar_h, input_h, pad = self._TITLE_H, self._TOOLBAR_H, self._INPUT_H, self._PAD
header_h = title_h + toolbar_h
btn_w, btn_h = self._clear_btn.size_x, self._clear_btn.size_y
Container._place(self._clear_btn, w - btn_w - pad, (title_h - btn_h) / 2, btn_w, btn_h)
# Filter toolbar: each widget keeps its content-driven size and is placed left to right.
cursor_x = 4.0
for cb in (*self._level_checkboxes.values(), self._stdout_checkbox):
if cb is self._stdout_checkbox:
cursor_x += 4.0
cw, ch = cb.size_x, cb.size_y
Container._place(cb, cursor_x, title_h + (toolbar_h - ch) / 2, cw, ch)
cursor_x += cw + (8.0 if cb is self._stdout_checkbox else 6.0)
# The search box takes the rest of the row, capped at 240.
search_w = max(80.0, min(240.0, w - pad - cursor_x))
sh = self._search_box.size_y
Container._place(self._search_box, cursor_x, title_h + (toolbar_h - sh) / 2, search_w, sh)
Container._place(self._output, 0.0, header_h, w, max(h - header_h - input_h - pad, 20.0))
prompt_w = self._prompt.get_minimum_size().x
Container._place(self._prompt, 4.0, h - input_h, prompt_w, input_h)
line_x = 4.0 + prompt_w + 4.0
Container._place(self._input, line_x, h - input_h, max(0.0, w - line_x), input_h)
[docs]
def on_draw(self, renderer):
x, y, w, h = self._local_rect()
title_h = self._TITLE_H
header_h = title_h + self._TOOLBAR_H
renderer.rect((x, y), (w, h), fill=self.bg_colour)
renderer.rect((x, y), (w, title_h), fill=(0.16, 0.16, 0.16, 1.0))
renderer.text("Console", (x + 8, y + 5), fill=(0.7, 0.7, 0.7, 1.0), scale=11.0 / 16.0)
renderer.rect((x, y + title_h), (w, self._TOOLBAR_H), fill=(0.13, 0.13, 0.13, 1.0))
renderer.line((x, y + header_h), (x + w, y + header_h), stroke=(0.25, 0.25, 0.25, 1.0))
# ------------------------------------------------------------------
# Cleanup
# ------------------------------------------------------------------
[docs]
def on_enter_tree(self):
"""Re-attach the log handler and follow the game tree when (re-)entering the tree."""
super().on_enter_tree()
play_state_changed = getattr(self.state, "play_state_changed", None)
if play_state_changed is not None:
play_state_changed.connect(self._follow_game_tree)
self._follow_game_tree()
# 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 on_exit_tree(self):
"""Remove log handler, disconnect error signal, and restore stdout on removal from tree."""
super().on_exit_tree()
logging.getLogger().removeHandler(self._log_handler)
play_state_changed = getattr(self.state, "play_state_changed", None)
if play_state_changed is not None:
play_state_changed.disconnect(self._follow_game_tree)
self._watch_errors(None)
self.stop_stdout_capture()