Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 27 additions & 2 deletions src/vulcanai/console/console.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,13 @@

from vulcanai.console.logger import VulcanAILogger
from vulcanai.console.modal_screens import CheckListModal, RadioListModal, ReverseSearchModal
from vulcanai.console.utils import SpinnerHook, StreamToTextual, attach_ros_logger_to_console, common_prefix
from vulcanai.console.terminal_session import TerminalSession
from vulcanai.console.utils import (
SpinnerHook,
StreamToTextual,
attach_ros_logger_to_console,
common_prefix,
)
from vulcanai.console.widget_custom_log_text_area import CustomLogTextArea
from vulcanai.console.widget_spinner import SpinnerStatus

Expand All @@ -53,25 +59,36 @@ class VulcanConsole(App):
CSS = """
Screen {
layout: horizontal;
overflow: hidden hidden;
}

#root {
width: 100%;
height: 100%;
overflow: hidden hidden;
}

#left {
width: 1fr;
layout: vertical;
overflow: hidden hidden;
}

#right {
width: 48;
layout: vertical;
border: tall #56AA08;
padding: 0;
overflow: hidden hidden;
}

#logcontent {
height: auto;
min-height: 1;
max-height: 1fr;
border: tall #333333;
scrollbar-size-vertical: 0;
scrollbar-size-horizontal: 0;
}

#llm_spinner {
Expand All @@ -94,6 +111,8 @@ class VulcanConsole(App):
#history_scroll {
height: 1fr;
margin: 1;
scrollbar-size-vertical: 0;
scrollbar-size-horizontal: 0;
}

#history {
Expand Down Expand Up @@ -170,6 +189,9 @@ def __init__(
self.suggestion_index = -1
self.suggestion_index_changed = threading.Event()

self._gnome_profile_schema: str | None = None
self._gnome_scrollbar_policy_backup: str | None = None

async def on_mouse_down(self, event: MouseEvent) -> None:
"""
Function used to paste the string for the user clipboard
Expand Down Expand Up @@ -1075,7 +1097,10 @@ def run_console(self) -> None:
"""
Function used to run VulcanAI.
"""
self.run()

session = TerminalSession()
with session:
self.run()

def init_manager(self) -> None:
"""
Expand Down
230 changes: 230 additions & 0 deletions src/vulcanai/console/terminal_session.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,230 @@
# Copyright 2026 Proyectos y Sistemas de Mantenimiento SL (eProsima).
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import os
import subprocess
import sys
from dataclasses import dataclass
from typing import Any, Optional, Protocol


def write_terminal_sequence(sequence: str) -> None:
"""
Write a raw escape sequence to the active terminal.

Used to change the color of the terminal.
- Change the terminal color using the same color of VulcanAI
- Restore the terminal color
"""
if not sys.stdout.isatty():
return
try:
sys.stdout.write(sequence)
sys.stdout.flush()
except Exception:
pass


class TerminalAdapter(Protocol):
"""
Abstract parent class to enhance VulcanAI visualization in each terminal.
Currently supported: Gnome
Not yet implemented: Terminator, Zsh
"""
Comment thread
cferreiragonz marked this conversation as resolved.

name: str

def detect(self) -> bool: ...

def apply(self) -> Any: ...

def restore(self, state: Any) -> None: ...


# region TERMINALS

# region gnome


def _run_gsettings(*args: str) -> Optional[str]:
"""
@brief Run gsettings and return trimmed stdout on success.
@param args Positional arguments forwarded to ``gsettings``.
@return Command stdout without trailing whitespace, or ``None`` on failure.
"""
try:
completed = subprocess.run(
["gsettings", *args],
check=False,
capture_output=True,
text=True,
)
except Exception:
return None
if completed.returncode != 0:
return None
return completed.stdout.strip()


@dataclass
class GnomeState:
"""@brief State required to restore GNOME Terminal settings."""

schema: str
scrollbar_policy_backup: str


class GnomeTerminalAdapter:
"""@brief GNOME Terminal adapter that hides and restores the scrollbar."""

name = "gnome-terminal"

def detect(self) -> bool:
"""
@brief Detect whether the current terminal is GNOME Terminal.
@return ``True`` when GNOME Terminal environment markers are found.
"""
is_gnome = (
"GNOME_TERMINAL_SCREEN" in os.environ
or "gnome-terminal" in os.environ.get("TERMINAL_EMULATOR", "").lower()
or "gnome-terminal" in os.environ.get("TERM_PROGRAM", "").lower()
)
return is_gnome

def apply(self) -> Optional[GnomeState]:
"""
@brief Hide GNOME scrollbar and return state for later restoration.
@return ``GnomeState`` when the change is applied/confirmed, else ``None``.
"""
# The return value could be None, empty string or string with just single quotes
profile_id = _run_gsettings("get", "org.gnome.Terminal.ProfilesList", "default")
if not profile_id:
Comment thread
cferreiragonz marked this conversation as resolved.
return None
profile_id = profile_id.strip("'")
if not profile_id:
return None

# GNOME stores per-profile keys under this dynamic schema path.
schema = f"org.gnome.Terminal.Legacy.Profile:/org/gnome/terminal/legacy/profiles:/:{profile_id}/"
current_policy = _run_gsettings("get", schema, "scrollbar-policy")
if not current_policy:
return None

# set only if needed
if current_policy != "'never'":
_run_gsettings("set", schema, "scrollbar-policy", "never")

return GnomeState(schema=schema, scrollbar_policy_backup=current_policy)

def restore(self, state: Optional[GnomeState]) -> None:
"""
@brief Restore the scrollbar policy captured by ``apply``.
@param state Previously saved state; no-op when ``None``.
@return None
"""
if not state:
return
restore_value = state.scrollbar_policy_backup.strip("'")
if restore_value:
_run_gsettings("set", state.schema, "scrollbar-policy", restore_value)


# endregion

# endregion


# region SESSION


@dataclass
class TerminalSessionConfig:
"""@brief Runtime options controlling generic terminal tweaks."""

# Background color used by OSC 11 (set default background color).
bg_color: str = "#121212"
# Emit OSC sequences to set and later reset background color.
force_bg: bool = True
# Emit DEC private mode sequence to hide/show scrollbar.
hide_scrollbar: bool = True


class TerminalSession:
"""
Session helper that applies terminal tweaks and safely restores them.
"""

def __init__(
self,
config: Optional[TerminalSessionConfig] = None,
adapters: Optional[list[TerminalAdapter]] = None,
):
self.config = config if config is not None else TerminalSessionConfig()
self.adapters = adapters if adapters is not None else [GnomeTerminalAdapter()]
self._active: list[tuple[TerminalAdapter, Any]] = []

def start(self) -> None:
"""
Apply generic and adapter-specific terminal tweaks.
"""
# Generic sequences (independent from specific emulators)
if self.config.force_bg:
# OSC 11: set default background color.
write_terminal_sequence(f"\x1b]11;{self.config.bg_color}\x07")
if self.config.hide_scrollbar:
# DECSET private mode 30: hide scrollbar where supported.
write_terminal_sequence("\x1b[?30l")

# Terminal-specific adapters
for adapter in self.adapters:
if adapter.detect():
state = adapter.apply()
self._active.append((adapter, state))

def end(self) -> None:
"""
Restore adapter state and generic terminal tweaks.
"""
# Restore adapters in reverse order
for adapter, state in reversed(self._active):
try:
adapter.restore(state)
except Exception:
pass
self._active.clear()

# Restore generic sequences
if self.config.hide_scrollbar:
# DECRST private mode 30: show scrollbar again.
write_terminal_sequence("\x1b[?30h")
if self.config.force_bg:
# OSC 111: reset default background color.
write_terminal_sequence("\x1b]111\x07")

def __enter__(self):
"""
Context-manager entrypoint.
"""
self.start()
return self

def __exit__(self, exc_type, exc, tb):
"""
Context-manager exitpoint; always restores terminal state.
"""
self.end()
return False


# endregion
45 changes: 28 additions & 17 deletions src/vulcanai/console/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,26 @@
from textual.markup import escape # To remove potential errors in textual terminal


class SpinnerHook:
"""
Single entrant spinner controller for console.
- Starts the spinner on the LLM request.
- Stops the spinner when LLM request is over.
"""

def __init__(self, spinner_status):
self.spinner_status = spinner_status

def on_request_start(self, text="Querying LLM..."):
self.spinner_status.start(text)

def on_request_end(self):
self.spinner_status.stop()


# region CONSOLE_REDIRECT


class StreamToTextual:
"""
Class used to redirect the stdout/stderr streams in the textual terminal
Expand All @@ -46,23 +66,6 @@ def flush(self):
self.real_stream.flush()


class SpinnerHook:
"""
Single entrant spinner controller for console.
- Starts the spinner on the LLM request.
- Stops the spinner when LLM request is over.
"""

def __init__(self, spinner_status):
self.spinner_status = spinner_status

def on_request_start(self, text="Querying LLM..."):
self.spinner_status.start(text)

def on_request_end(self):
self.spinner_status.stop()


def attach_ros_logger_to_console(console):
"""
Redirect ALL rclpy RcutilsLogger output (nodes + executor + rclpy internals)
Expand Down Expand Up @@ -123,6 +126,11 @@ def patched_log(self, msg, level, *args, **kwargs):
RcutilsLogger._textual_patched = True


# endregion

# region TEXTUAL


def common_prefix(strings: str) -> str:
if not strings:
return ""
Expand Down Expand Up @@ -349,3 +357,6 @@ def _get_suggestions(real_string_list_comp: list[str], string_comp: str) -> tupl
console.suggestion_index_changed.clear()

return ret


# endregion