bec/bec_widgets
1
0
Fork 0
mirror of https://github.com/bec-project/bec_widgets.git synced 2026-05-04 05:44:23 +02:00
Code Activity

feat: add qtermwidget plugin and replace web term

Browse Source
This commit is contained in:
perl_d
2026-04-07 17:52:40 +02:00
committed by Jan Wyzula
parent 1d3e0214fd
commit 02cb393bb0
23 changed files with 890 additions and 911 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+3 -1
View File
@@ -177,4 +177,6 @@ cython_debug/
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
# and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
#.idea/
#.idea/
#
tombi.toml
bec_widgets/applications/views/developer_view/developer_widget.py
+2 -22
View File
@@ -16,9 +16,9 @@ from bec_widgets.utils.toolbars.toolbar import ModularToolBar
from bec_widgets.widgets.containers.dock_area.basic_dock_area import DockAreaWidget
from bec_widgets.widgets.containers.dock_area.dock_area import BECDockArea
from bec_widgets.widgets.containers.qt_ads import CDockWidget
from bec_widgets.widgets.editors.bec_console.bec_console import BecConsole, BECShell
from bec_widgets.widgets.editors.monaco.monaco_dock import MonacoDock
from bec_widgets.widgets.editors.monaco.monaco_widget import MonacoWidget
from bec_widgets.widgets.editors.web_console.web_console import BECShell, WebConsole
from bec_widgets.widgets.utility.ide_explorer.ide_explorer import IDEExplorer
@@ -96,7 +96,7 @@ class DeveloperWidget(DockAreaWidget):
self.console = BECShell(self, rpc_exposed=False)
self.console.setObjectName("BEC Shell")
self.terminal = WebConsole(self, rpc_exposed=False)
self.terminal = BecConsole(self, rpc_exposed=False)
self.terminal.setObjectName("Terminal")
self.monaco = MonacoDock(self, rpc_exposed=False, rpc_passthrough_children=False)
self.monaco.setObjectName("MonacoEditor")
@@ -410,23 +410,3 @@ class DeveloperWidget(DockAreaWidget):
"""Clean up resources used by the developer widget."""
self.delete_all()
return super().cleanup()
if __name__ == "__main__":
import sys
from bec_qthemes import apply_theme
from qtpy.QtWidgets import QApplication
from bec_widgets.applications.main_app import BECMainApp
app = QApplication(sys.argv)
apply_theme("dark")
_app = BECMainApp()
_app.show()
# developer_view.show()
# developer_view.setWindowTitle("Developer View")
# developer_view.resize(1920, 1080)
# developer_view.set_stretch(horizontal=[1, 3, 2], vertical=[5, 5]) #can be set during runtime
sys.exit(app.exec_())
bec_widgets/cli/client.py
+24 -24
View File
@@ -32,6 +32,7 @@ _Widgets = {
"BECQueue": "BECQueue",
"BECShell": "BECShell",
"BECStatusBox": "BECStatusBox",
"BecConsole": "BecConsole",
"DapComboBox": "DapComboBox",
"DeviceBrowser": "DeviceBrowser",
"Heatmap": "Heatmap",
@@ -56,7 +57,6 @@ _Widgets = {
"SignalLabel": "SignalLabel",
"TextBox": "TextBox",
"Waveform": "Waveform",
"WebConsole": "WebConsole",
"WebsiteWidget": "WebsiteWidget",
}
@@ -506,7 +506,7 @@ class BECQueue(RPCBase):
class BECShell(RPCBase):
"""A WebConsole pre-configured to run the BEC shell."""
"""A BecConsole pre-configured to run the BEC shell."""
@rpc_call
def remove(self):
@@ -691,6 +691,28 @@ class BaseROI(RPCBase):
"""
class BecConsole(RPCBase):
"""A console widget with access to a shared registry of terminals, such that instances can be moved around."""
@rpc_call
def remove(self):
"""
Cleanup the BECConnector
"""
@rpc_call
def attach(self):
"""
None
"""
@rpc_call
def detach(self):
"""
Detach the widget from its parent dock widget (if widget is in the dock), making it a floating widget.
"""
class CircularROI(RPCBase):
"""Circular Region of Interest with center/diameter tracking and auto-labeling."""
@@ -6417,28 +6439,6 @@ class WaveformViewPopup(RPCBase):
"""
class WebConsole(RPCBase):
"""A simple widget to display a website"""
@rpc_call
def remove(self):
"""
Cleanup the BECConnector
"""
@rpc_call
def attach(self):
"""
None
"""
@rpc_call
def detach(self):
"""
Detach the widget from its parent dock widget (if widget is in the dock), making it a floating widget.
"""
class WebsiteWidget(RPCBase):
"""A simple widget to display a website"""
bec_widgets/widgets/containers/dock_area/dock_area.py
+2 -2
View File
@@ -69,7 +69,7 @@ from bec_widgets.widgets.containers.main_window.main_window import BECMainWindow
from bec_widgets.widgets.containers.qt_ads import CDockWidget
from bec_widgets.widgets.control.device_control.positioner_box import PositionerBox, PositionerBox2D
from bec_widgets.widgets.control.scan_control import ScanControl
from bec_widgets.widgets.editors.web_console.web_console import BECShell, WebConsole
from bec_widgets.widgets.editors.bec_console.bec_console import BecConsole, BECShell
from bec_widgets.widgets.plots.heatmap.heatmap import Heatmap
from bec_widgets.widgets.plots.image.image import Image
from bec_widgets.widgets.plots.motor_map.motor_map import MotorMap
@@ -372,7 +372,7 @@ class BECDockArea(DockAreaWidget):
"Add Circular ProgressBar",
"RingProgressBar",
),
"terminal": (WebConsole.ICON_NAME, "Add Terminal", "WebConsole"),
"terminal": (BecConsole.ICON_NAME, "Add Terminal", "BecConsole"),
"bec_shell": (BECShell.ICON_NAME, "Add BEC Shell", "BECShell"),
"log_panel": (LogPanel.ICON_NAME, "Add LogPanel - Disabled", "LogPanel"),
"sbb_monitor": ("train", "Add SBB Monitor", "SBBMonitor"),
bec_widgets/widgets/editors/web_console/__init__.py → bec_widgets/widgets/editors/bec_console/__init__.py
View File
bec_widgets/widgets/editors/bec_console/bec_console.py
+431
View File
@@ -0,0 +1,431 @@
from __future__ import annotations
import enum
from uuid import uuid4
from weakref import WeakValueDictionary
from bec_lib.logger import bec_logger
from pydantic import BaseModel
from qtpy.QtCore import Qt, Signal
from qtpy.QtGui import QMouseEvent, QResizeEvent
from qtpy.QtWidgets import (
QApplication,
QHBoxLayout,
QLabel,
QStackedLayout,
QTabWidget,
QVBoxLayout,
QWidget,
)
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.widgets.utility.bec_term.protocol import BecTerminal
from bec_widgets.widgets.utility.bec_term.util import get_current_bec_term_class
logger = bec_logger.logger
_BecTermClass = get_current_bec_term_class()
# Note on definitions:
# Terminal: an instance of a terminal widget with a system shell
# Console: one of possibly several widgets which may share ownership of one single terminal
# Shell: a Console set to start the BEC IPython client in its terminal
class ConsoleMode(str, enum.Enum):
ACTIVE = "active"
INACTIVE = "inactive"
HIDDEN = "hidden"
class _TerminalOwnerInfo(BaseModel):
"""Should be managed only by the BecConsoleRegistry. Consoles should ask the registry for
necessary ownership info."""
owner_console_id: str | None = None
registered_console_ids: set[str] = set()
instance: BecTerminal
terminal_id: str
initialized: bool = False
model_config = {"arbitrary_types_allowed": True}
class BecConsoleRegistry:
"""
A registry for the BecConsole class to manage its instances.
"""
def __init__(self):
"""
Initialize the registry.
"""
self._consoles: WeakValueDictionary[str, BecConsole] = WeakValueDictionary()
self._terminal_registry: dict[str, _TerminalOwnerInfo] = {}
def register(self, console: BecConsole):
"""
Register an instance of BecConsole. If there is already a terminal with the associated
terminal_id, this does not automatically grant ownership.
Args:
console (BecConsole): The instance to register.
"""
self._consoles[console.console_id] = console
console_id, terminal_id = console.console_id, console.terminal_id
if (term_info := self._terminal_registry.get(terminal_id)) is None:
term = _BecTermClass()
self._terminal_registry[terminal_id] = _TerminalOwnerInfo(
registered_console_ids={console_id},
owner_console_id=console_id,
instance=term,
terminal_id=terminal_id,
)
return
logger.info(f"Registered new console {console_id} for terminal {terminal_id}")
term_info.registered_console_ids.add(console_id)
def unregister(self, console: BecConsole):
"""
Unregister an instance of BecConsole.
Args:
instance (BecConsole): The instance to unregister.
"""
console_id, terminal_id = console.console_id, console.terminal_id
if console_id in self._consoles:
del self._consoles[console_id]
if (term_info := self._terminal_registry.get(console_id)) is None:
return
if console_id in term_info.registered_console_ids:
term_info.registered_console_ids.remove(console_id)
if term_info.owner_console_id == console_id:
term_info.owner_console_id = None
if not term_info.registered_console_ids:
term_info.instance.deleteLater()
del self._terminal_registry[terminal_id]
logger.info(f"Unregistered console {console_id} for terminal {terminal_id}")
def is_owner(self, console: BecConsole):
"""Returns true if the given console is the owner of its terminal"""
if console not in self._consoles.values():
return False
if (info := self._terminal_registry.get(console.terminal_id)) is None:
logger.warning(f"Console {console.console_id} references an unknown terminal!")
return False
return info.owner_console_id == console.console_id
def take_ownership(self, console: BecConsole) -> BecTerminal | None:
"""
Transfer ownership of a terminal to the given console.
Args:
console: the console which wishes to take ownership of its associated terminal.
Returns:
BecTerminal | None: The instance if ownership transfer was successful, None otherwise.
"""
console_id, terminal_id = console.console_id, console.terminal_id
if terminal_id not in self._terminal_registry:
logger.warning(f"Terminal {terminal_id} not found in registry")
return None
instance_info = self._terminal_registry[terminal_id]
if (old_owner_console_ide := instance_info.owner_console_id) is not None:
if (old_owner := self._consoles.get(old_owner_console_ide)) is not None:
old_owner.yield_ownership() # call this on the old owner to make sure it is updated
instance_info.owner_console_id = console_id
logger.info(f"Transferred ownership of terminal {terminal_id} to {console_id}")
return instance_info.instance
def try_get_term(self, console: BecConsole) -> BecTerminal | None:
"""
Return the terminal instance if the requesting console is the owner
Args:
console: the requesting console.
Returns:
BecTerminal | None: The instance if the console is the owner, None otherwise.
"""
console_id, terminal_id = console.console_id, console.terminal_id
logger.debug(f"checking term for {console_id}")
if terminal_id not in self._terminal_registry:
logger.warning(f"Terminal {terminal_id} not found in registry")
return None
instance_info = self._terminal_registry[terminal_id]
if instance_info.owner_console_id == console_id:
return instance_info.instance
def yield_ownership(self, console: BecConsole):
"""
Yield ownership of a instance without destroying it. The instance remains in the
registry with no owner, available for another widget to claim.
Args:
gui_id (str): The GUI ID of the widget yielding ownership.
"""
console_id, terminal_id = console.console_id, console.terminal_id
logger.debug(f"Console {console_id} attempted to yield ownership")
if console_id not in self._consoles or terminal_id not in self._terminal_registry:
return
term_info = self._terminal_registry[terminal_id]
if term_info.owner_console_id != console_id:
logger.debug(f"But it was not the owner, which was {term_info.owner_console_id}!")
return
term_info.owner_console_id = None
term_info.instance.setParent(None)
def owner_is_visible(self, term_id: str) -> bool:
"""
Check if the owner of a instance is currently visible.
Args:
unique_id (str): The unique identifier for the instance.
Returns:
bool: True if the owner is visible, False otherwise.
"""
instance_info = self._terminal_registry.get(term_id)
if instance_info is None or instance_info.owner_console_id is None:
return False
if (owner := self._consoles.get(instance_info.owner_console_id)) is None:
return False
return owner.isVisible()
_bec_console_registry = BecConsoleRegistry()
class _Overlay(QWidget):
def __init__(self, console: BecConsole):
super().__init__(parent=console)
self._console = console
def mousePressEvent(self, event: QMouseEvent) -> None:
if event.button() == Qt.MouseButton.LeftButton:
self._console.take_terminal_ownership()
event.accept()
return
return super().mousePressEvent(event)
class BecConsole(BECWidget, QWidget):
"""A console widget with access to a shared registry of terminals, such that instances can be moved around."""
_js_callback = Signal(bool)
initialized = Signal()
PLUGIN = True
ICON_NAME = "terminal"
def __init__(
self,
parent=None,
config=None,
client=None,
gui_id=None,
startup_cmd: str | None = None,
is_bec_shell: bool = False,
terminal_id: str | None = None,
**kwargs,
):
super().__init__(parent=parent, client=client, gui_id=gui_id, config=config, **kwargs)
self._mode = ConsoleMode.INACTIVE
self._is_bec_shell = is_bec_shell
self._startup_cmd = startup_cmd
self._is_initialized = False
self.terminal_id = terminal_id or str(uuid4())
self.console_id = self.gui_id
self.term: BecTerminal | None = None # Will be set in _set_up_instance
self._set_up_instance()
def _set_up_instance(self):
"""
Set up the web instance and UI elements.
"""
self._stacked_layout = QStackedLayout()
# self._stacked_layout.setStackingMode(QStackedLayout.StackingMode.StackAll)
self._term_holder = QWidget()
self._term_layout = QVBoxLayout()
self._term_layout.setContentsMargins(0, 0, 0, 0)
self._term_holder.setLayout(self._term_layout)
self.setLayout(self._stacked_layout)
# prepare overlay
self._overlay = _Overlay(self)
layout = QVBoxLayout(self._overlay)
layout.setAlignment(Qt.AlignmentFlag.AlignCenter)
label = QLabel("Click to activate terminal", self._overlay)
layout.addWidget(label)
self._stacked_layout.addWidget(self._term_holder)
self._stacked_layout.addWidget(self._overlay)
# will create a new terminal instance if there isn't already one for this ID
_bec_console_registry.register(self)
self._infer_mode()
if self.startup_cmd:
self.write(self.startup_cmd, True) # will have no effect if not the owner
def _infer_mode(self):
self.term = _bec_console_registry.try_get_term(self)
if self.term:
self._set_mode(ConsoleMode.ACTIVE)
elif self.isHidden:
self._set_mode(ConsoleMode.HIDDEN)
else:
self._set_mode(ConsoleMode.INACTIVE)
def _set_mode(self, mode: ConsoleMode):
"""
Set the mode of the web console.
Args:
mode (ConsoleMode): The mode to set.
"""
match mode:
case ConsoleMode.ACTIVE:
if self.term:
if self.term not in (self._term_layout.children()):
self._term_layout.addWidget(self.term) # type: ignore # BecTerminal is QWidget
self._stacked_layout.setCurrentIndex(0)
self._mode = mode
else:
self._stacked_layout.setCurrentIndex(1)
self._mode = ConsoleMode.INACTIVE
case ConsoleMode.INACTIVE:
self._stacked_layout.setCurrentIndex(1)
self._mode = mode
case ConsoleMode.HIDDEN:
self._stacked_layout.setCurrentIndex(1)
self._mode = mode
@property
def startup_cmd(self):
"""
Get the startup command for the web console.
"""
return self._startup_cmd
@startup_cmd.setter
def startup_cmd(self, cmd: str | None):
"""
Set the startup command for the console.
logger.info(f"{self._console_id} inferred mode active through ownerp)
"""
self._startup_cmd = cmd
def write(self, data: str, send_return: bool = True):
"""
Send data to the console
Args:
data (str): The data to send.
send_return (bool): Whether to send a return after the data.
"""
if self.term:
self.term.write(data, send_return)
def take_terminal_ownership(self):
"""
Take ownership of a web instance from the registry. This will transfer the instance
from its current owner (if any) to this widget.
"""
# Get the instance from registry
self.term = _bec_console_registry.take_ownership(self)
self._infer_mode()
if self._mode == ConsoleMode.ACTIVE:
logger.debug(f"Widget {self.gui_id} took ownership of instance {self.terminal_id}")
def yield_ownership(self):
"""
Yield ownership of the instance. The instance remains in the registry with no owner,
available for another widget to claim. This is automatically called when the
widget becomes hidden.
"""
_bec_console_registry.yield_ownership(self)
self._infer_mode()
if self._mode != ConsoleMode.ACTIVE:
logger.debug(f"Widget {self.gui_id} yielded ownership of instance {self.terminal_id}")
def hideEvent(self, event):
"""Called when the widget is hidden. Automatically yields ownership."""
self.yield_ownership()
super().hideEvent(event)
def showEvent(self, event):
"""Called when the widget is shown. Updates UI state based on ownership."""
super().showEvent(event)
if not _bec_console_registry.is_owner(self):
if not _bec_console_registry.owner_is_visible(self.terminal_id):
self.take_terminal_ownership()
def cleanup(self):
"""Unregister this console on destruction."""
_bec_console_registry.unregister(self)
super().cleanup()
class BECShell(BecConsole):
"""
A BecConsole pre-configured to run the BEC shell.
We cannot simply expose the web console properties to Qt as we need to have a deterministic
startup behavior for sharing the same shell instance across multiple widgets.
"""
ICON_NAME = "hub"
def __init__(self, parent=None, config=None, client=None, gui_id=None, **kwargs):
super().__init__(
parent=parent,
config=config,
client=client,
gui_id=gui_id,
terminal_id="bec_shell",
**kwargs,
)
@property
def startup_cmd(self):
"""
Get the startup command for the BEC shell.
"""
if self.bec_dispatcher.cli_server is None:
return "bec --nogui"
return f"bec --gui-id {self.bec_dispatcher.cli_server.gui_id}"
@startup_cmd.setter
def startup_cmd(self, cmd: str | None): ...
if __name__ == "__main__": # pragma: no cover
import sys
app = QApplication(sys.argv)
widget = QTabWidget()
# Create two consoles with different unique_ids
bec_console_1a = BecConsole(startup_cmd="htop", gui_id="console_1_a", terminal_id="terminal_1")
bec_console_1b = BecConsole(startup_cmd="htop", gui_id="console_1_b", terminal_id="terminal_1")
bec_console_1 = QWidget()
bec_console_1_layout = QHBoxLayout(bec_console_1)
bec_console_1_layout.addWidget(bec_console_1a)
bec_console_1_layout.addWidget(bec_console_1b)
bec_console2 = BECShell()
bec_console3 = BecConsole(gui_id="console_3", terminal_id="terminal_1")
widget.addTab(bec_console_1, "Console 1")
widget.addTab(bec_console2, "Console 2 - BEC Shell")
widget.addTab(bec_console3, "Console 3 -- mirror of Console 1")
widget.show()
widget.resize(800, 600)
sys.exit(app.exec_())
bec_widgets/widgets/editors/bec_console/bec_console.pyproject
+1
View File
@@ -0,0 +1 @@
{'files': ['bec_console.py']}
bec_widgets/widgets/editors/web_console/web_console_plugin.py → bec_widgets/widgets/editors/bec_console/bec_console_plugin.py
+10 -10
View File
@@ -5,38 +5,38 @@ from qtpy.QtDesigner import QDesignerCustomWidgetInterface
from qtpy.QtWidgets import QWidget
from bec_widgets.utils.bec_designer import designer_material_icon
from bec_widgets.widgets.editors.web_console.web_console import WebConsole
from bec_widgets.widgets.editors.bec_console.bec_console import BecConsole
DOM_XML = """
<ui language='c++'>
<widget class='WebConsole' name='web_console'>
<widget class='BecConsole' name='bec_console'>
</widget>
</ui>
"""
class WebConsolePlugin(QDesignerCustomWidgetInterface): # pragma: no cover
class BecConsolePlugin(QDesignerCustomWidgetInterface): # pragma: no cover
def __init__(self):
super().__init__()
self._form_editor = None
def createWidget(self, parent):
if parent is None:
return QWidget()
t = WebConsole(parent)
return QWidget()
t = BecConsole(parent)
return t
def domXml(self):
return DOM_XML
def group(self):
return "BEC Developer"
return ""
def icon(self):
return designer_material_icon(WebConsole.ICON_NAME)
return designer_material_icon(BecConsole.ICON_NAME)
def includeFile(self):
return "web_console"
return "bec_console"
def initialize(self, form_editor):
self._form_editor = form_editor
@@ -48,10 +48,10 @@ class WebConsolePlugin(QDesignerCustomWidgetInterface): # pragma: no cover
return self._form_editor is not None
def name(self):
return "WebConsole"
return "BecConsole"
def toolTip(self):
return ""
return "A console widget with access to a shared registry of terminals, such that instances can be moved around."
def whatsThis(self):
return self.toolTip()
bec_widgets/widgets/editors/bec_console/bec_shell.pyproject
+1
View File
@@ -0,0 +1 @@
{'files': ['bec_console.py']}
bec_widgets/widgets/editors/web_console/bec_shell_plugin.py → bec_widgets/widgets/editors/bec_console/bec_shell_plugin.py
+2 -2
View File
@@ -5,7 +5,7 @@ from qtpy.QtDesigner import QDesignerCustomWidgetInterface
from qtpy.QtWidgets import QWidget
from bec_widgets.utils.bec_designer import designer_material_icon
from bec_widgets.widgets.editors.web_console.web_console import BECShell
from bec_widgets.widgets.editors.bec_console.bec_console import BECShell
DOM_XML = """
<ui language='c++'>
@@ -22,7 +22,7 @@ class BECShellPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
def createWidget(self, parent):
if parent is None:
return QWidget()
return QWidget()
t = BECShell(parent)
return t
bec_widgets/widgets/editors/web_console/register_web_console.py → bec_widgets/widgets/editors/bec_console/register_bec_console.py
+2 -2
View File
@@ -6,9 +6,9 @@ def main(): # pragma: no cover
return
from PySide6.QtDesigner import QPyDesignerCustomWidgetCollection
from bec_widgets.widgets.editors.web_console.web_console_plugin import WebConsolePlugin
from bec_widgets.widgets.editors.bec_console.bec_console_plugin import BecConsolePlugin
QPyDesignerCustomWidgetCollection.addCustomWidget(WebConsolePlugin())
QPyDesignerCustomWidgetCollection.addCustomWidget(BecConsolePlugin())
if __name__ == "__main__": # pragma: no cover
bec_widgets/widgets/editors/web_console/register_bec_shell.py → bec_widgets/widgets/editors/bec_console/register_bec_shell.py
+1 -1
View File
@@ -6,7 +6,7 @@ def main(): # pragma: no cover
return
from PySide6.QtDesigner import QPyDesignerCustomWidgetCollection
from bec_widgets.widgets.editors.web_console.bec_shell_plugin import BECShellPlugin
from bec_widgets.widgets.editors.bec_console.bec_shell_plugin import BECShellPlugin
QPyDesignerCustomWidgetCollection.addCustomWidget(BECShellPlugin())
bec_widgets/widgets/editors/web_console/bec_shell.pyproject
-1
View File
@@ -1 +0,0 @@
{'files': ['web_console.py']}
bec_widgets/widgets/editors/web_console/web_console.py
-705
View File
@@ -1,705 +0,0 @@
from __future__ import annotations
import enum
import json
import secrets
import subprocess
import time
from bec_lib.logger import bec_logger
from louie.saferef import safe_ref
from pydantic import BaseModel
from qtpy.QtCore import Qt, QTimer, QUrl, Signal, qInstallMessageHandler
from qtpy.QtGui import QMouseEvent, QResizeEvent
from qtpy.QtWebEngineWidgets import QWebEnginePage, QWebEngineView
from qtpy.QtWidgets import QApplication, QLabel, QTabWidget, QVBoxLayout, QWidget
from bec_widgets.utils.bec_widget import BECWidget
logger = bec_logger.logger
class ConsoleMode(str, enum.Enum):
ACTIVE = "active"
INACTIVE = "inactive"
HIDDEN = "hidden"
class PageOwnerInfo(BaseModel):
owner_gui_id: str | None = None
widget_ids: list[str] = []
page: QWebEnginePage | None = None
initialized: bool = False
model_config = {"arbitrary_types_allowed": True}
class WebConsoleRegistry:
"""
A registry for the WebConsole class to manage its instances.
"""
def __init__(self):
"""
Initialize the registry.
"""
self._instances = {}
self._server_process = None
self._server_port = None
self._token = secrets.token_hex(16)
self._page_registry: dict[str, PageOwnerInfo] = {}
def register(self, instance: WebConsole):
"""
Register an instance of WebConsole.
Args:
instance (WebConsole): The instance to register.
"""
self._instances[instance.gui_id] = safe_ref(instance)
self.cleanup()
if instance._unique_id:
self._register_page(instance)
if self._server_process is None:
# Start the ttyd server if not already running
self.start_ttyd()
def start_ttyd(self, use_zsh: bool | None = None):
"""
Start the ttyd server
ttyd -q -W -t 'theme={"background": "black"}' zsh
Args:
use_zsh (bool): Whether to use zsh or bash. If None, it will try to detect if zsh is available.
"""
# First, check if ttyd is installed
try:
subprocess.run(["ttyd", "--version"], check=True, stdout=subprocess.PIPE)
except FileNotFoundError:
# pylint: disable=raise-missing-from
raise RuntimeError("ttyd is not installed. Please install it first.")
if use_zsh is None:
# Check if we can use zsh
try:
subprocess.run(["zsh", "--version"], check=True, stdout=subprocess.PIPE)
use_zsh = True
except FileNotFoundError:
use_zsh = False
command = [
"ttyd",
"-p",
"0",
"-W",
"-t",
'theme={"background": "black"}',
"-c",
f"user:{self._token}",
]
if use_zsh:
command.append("zsh")
else:
command.append("bash")
# Start the ttyd server
self._server_process = subprocess.Popen(
command, stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
self._wait_for_server_port()
self._server_process.stdout.close()
self._server_process.stderr.close()
def _wait_for_server_port(self, timeout: float = 10):
"""
Wait for the ttyd server to start and get the port number.
Args:
timeout (float): The timeout in seconds to wait for the server to start.
"""
start_time = time.time()
while True:
output = self._server_process.stderr.readline()
if output == b"" and self._server_process.poll() is not None:
break
if not output:
continue
output = output.decode("utf-8").strip()
if "Listening on" in output:
# Extract the port number from the output
self._server_port = int(output.split(":")[-1])
logger.info(f"ttyd server started on port {self._server_port}")
break
if time.time() - start_time > timeout:
raise TimeoutError(
"Timeout waiting for ttyd server to start. Please check if ttyd is installed and available in your PATH."
)
def cleanup(self):
"""
Clean up the registry by removing any instances that are no longer valid.
"""
for gui_id, weak_ref in list(self._instances.items()):
if weak_ref() is None:
del self._instances[gui_id]
if not self._instances and self._server_process:
# If no instances are left, terminate the server process
self._server_process.terminate()
self._server_process = None
self._server_port = None
logger.info("ttyd server terminated")
def unregister(self, instance: WebConsole):
"""
Unregister an instance of WebConsole.
Args:
instance (WebConsole): The instance to unregister.
"""
if instance.gui_id in self._instances:
del self._instances[instance.gui_id]
if instance._unique_id:
self._unregister_page(instance._unique_id, instance.gui_id)
self.cleanup()
def _register_page(self, instance: WebConsole):
"""
Register a page in the registry. Please note that this does not transfer ownership
for already existing pages; it simply records which widget currently owns the page.
Use transfer_page_ownership to change ownership.
Args:
instance (WebConsole): The instance to register.
"""
unique_id = instance._unique_id
gui_id = instance.gui_id
if unique_id is None:
return
if unique_id not in self._page_registry:
page = BECWebEnginePage()
page.authenticationRequired.connect(instance._authenticate)
self._page_registry[unique_id] = PageOwnerInfo(
owner_gui_id=gui_id, widget_ids=[gui_id], page=page
)
logger.info(f"Registered new page {unique_id} for {gui_id}")
return
if gui_id not in self._page_registry[unique_id].widget_ids:
self._page_registry[unique_id].widget_ids.append(gui_id)
def _unregister_page(self, unique_id: str, gui_id: str):
"""
Unregister a page from the registry.
Args:
unique_id (str): The unique identifier for the page.
gui_id (str): The GUI ID of the widget.
"""
if unique_id not in self._page_registry:
return
page_info = self._page_registry[unique_id]
if gui_id in page_info.widget_ids:
page_info.widget_ids.remove(gui_id)
if page_info.owner_gui_id == gui_id:
page_info.owner_gui_id = None
if not page_info.widget_ids:
if page_info.page:
page_info.page.deleteLater()
del self._page_registry[unique_id]
logger.info(f"Unregistered page {unique_id} for {gui_id}")
def get_page_info(self, unique_id: str) -> PageOwnerInfo | None:
"""
Get a page from the registry.
Args:
unique_id (str): The unique identifier for the page.
Returns:
PageOwnerInfo | None: The page info if found, None otherwise.
"""
if unique_id not in self._page_registry:
return None
return self._page_registry[unique_id]
def take_page_ownership(self, unique_id: str, new_owner_gui_id: str) -> QWebEnginePage | None:
"""
Transfer ownership of a page to a new owner.
Args:
unique_id (str): The unique identifier for the page.
new_owner_gui_id (str): The GUI ID of the new owner.
Returns:
QWebEnginePage | None: The page if ownership transfer was successful, None otherwise.
"""
if unique_id not in self._page_registry:
logger.warning(f"Page {unique_id} not found in registry")
return None
page_info = self._page_registry[unique_id]
old_owner_gui_id = page_info.owner_gui_id
if old_owner_gui_id:
old_owner_ref = self._instances.get(old_owner_gui_id)
if old_owner_ref:
old_owner_instance = old_owner_ref()
if old_owner_instance:
old_owner_instance.yield_ownership()
page_info.owner_gui_id = new_owner_gui_id
logger.info(f"Transferred ownership of page {unique_id} to {new_owner_gui_id}")
return page_info.page
def yield_ownership(self, gui_id: str) -> bool:
"""
Yield ownership of a page without destroying it. The page remains in the
registry with no owner, available for another widget to claim.
Args:
gui_id (str): The GUI ID of the widget yielding ownership.
Returns:
bool: True if ownership was yielded, False otherwise.
"""
if gui_id not in self._instances:
return False
instance = self._instances[gui_id]()
if instance is None:
return False
unique_id = instance._unique_id
if unique_id is None:
return False
if unique_id not in self._page_registry:
return False
page_owner_info = self._page_registry[unique_id]
if page_owner_info.owner_gui_id != gui_id:
return False
page_owner_info.owner_gui_id = None
return True
def owner_is_visible(self, unique_id: str) -> bool:
"""
Check if the owner of a page is currently visible.
Args:
unique_id (str): The unique identifier for the page.
Returns:
bool: True if the owner is visible, False otherwise.
"""
page_info = self.get_page_info(unique_id)
if page_info is None or page_info.owner_gui_id is None:
return False
owner_ref = self._instances.get(page_info.owner_gui_id)
if owner_ref is None:
return False
owner_instance = owner_ref()
if owner_instance is None:
return False
return owner_instance.isVisible()
_web_console_registry = WebConsoleRegistry()
def suppress_qt_messages(type_, context, msg):
if context.category in ["js", "default"]:
return
print(msg)
qInstallMessageHandler(suppress_qt_messages)
class BECWebEnginePage(QWebEnginePage):
def javaScriptConsoleMessage(self, level, message, lineNumber, sourceID):
logger.info(f"[JS Console] {level.name} at line {lineNumber} in {sourceID}: {message}")
class WebConsole(BECWidget, QWidget):
"""
A simple widget to display a website
"""
_js_callback = Signal(bool)
initialized = Signal()
PLUGIN = True
ICON_NAME = "terminal"
def __init__(
self,
parent=None,
config=None,
client=None,
gui_id=None,
startup_cmd: str | None = None,
is_bec_shell: bool = False,
unique_id: str | None = None,
**kwargs,
):
super().__init__(parent=parent, client=client, gui_id=gui_id, config=config, **kwargs)
self._mode = ConsoleMode.INACTIVE
self._is_bec_shell = is_bec_shell
self._startup_cmd = startup_cmd
self._is_initialized = False
self._unique_id = unique_id
self.page = None # Will be set in _set_up_page
self._startup_timer = QTimer()
self._startup_timer.setInterval(500)
self._startup_timer.timeout.connect(self._check_page_ready)
self._startup_timer.start()
self._js_callback.connect(self._on_js_callback)
self._set_up_page()
def _set_up_page(self):
"""
Set up the web page and UI elements.
"""
layout = QVBoxLayout()
layout.setContentsMargins(0, 0, 0, 0)
self.browser = QWebEngineView(self)
layout.addWidget(self.browser)
self.setLayout(layout)
# prepare overlay
self.overlay = QWidget(self)
layout = QVBoxLayout(self.overlay)
layout.setAlignment(Qt.AlignmentFlag.AlignCenter)
label = QLabel("Click to activate terminal", self.overlay)
layout.addWidget(label)
self.overlay.hide()
_web_console_registry.register(self)
self._token = _web_console_registry._token
# If no unique_id is provided, create a new page
if not self._unique_id:
self.page = BECWebEnginePage(self)
self.page.authenticationRequired.connect(self._authenticate)
self.page.setUrl(QUrl(f"http://localhost:{_web_console_registry._server_port}"))
self.browser.setPage(self.page)
self._set_mode(ConsoleMode.ACTIVE)
return
# Try to get the page from the registry
page_info = _web_console_registry.get_page_info(self._unique_id)
if page_info and page_info.page:
self.page = page_info.page
if not page_info.owner_gui_id or page_info.owner_gui_id == self.gui_id:
self.browser.setPage(self.page)
# Only set URL if this is a newly created page (no URL set yet)
if self.page.url().isEmpty():
self.page.setUrl(QUrl(f"http://localhost:{_web_console_registry._server_port}"))
else:
# We have an existing page, so we don't need the startup timer
self._startup_timer.stop()
if page_info.owner_gui_id != self.gui_id:
self._set_mode(ConsoleMode.INACTIVE)
else:
self._set_mode(ConsoleMode.ACTIVE)
def _set_mode(self, mode: ConsoleMode):
"""
Set the mode of the web console.
Args:
mode (ConsoleMode): The mode to set.
"""
if not self._unique_id:
# For non-unique_id consoles, always active
mode = ConsoleMode.ACTIVE
self._mode = mode
match mode:
case ConsoleMode.ACTIVE:
self.browser.setVisible(True)
self.overlay.hide()
case ConsoleMode.INACTIVE:
self.browser.setVisible(False)
self.overlay.show()
case ConsoleMode.HIDDEN:
self.browser.setVisible(False)
self.overlay.hide()
def _check_page_ready(self):
"""
Check if the page is ready and stop the timer if it is.
"""
if not self.page or self.page.isLoading():
return
self.page.runJavaScript("window.term !== undefined", self._js_callback.emit)
def _on_js_callback(self, ready: bool):
"""
Callback for when the JavaScript is ready.
"""
if not ready:
return
self._is_initialized = True
self._startup_timer.stop()
if self.startup_cmd:
if self._unique_id:
page_info = _web_console_registry.get_page_info(self._unique_id)
if page_info is None:
return
if not page_info.initialized:
page_info.initialized = True
self.write(self.startup_cmd)
else:
self.write(self.startup_cmd)
self.initialized.emit()
@property
def startup_cmd(self):
"""
Get the startup command for the web console.
"""
if self._is_bec_shell:
if self.bec_dispatcher.cli_server is None:
return "bec --nogui"
return f"bec --gui-id {self.bec_dispatcher.cli_server.gui_id}"
return self._startup_cmd
@startup_cmd.setter
def startup_cmd(self, cmd: str):
"""
Set the startup command for the web console.
"""
if not isinstance(cmd, str):
raise ValueError("Startup command must be a string.")
self._startup_cmd = cmd
def write(self, data: str, send_return: bool = True):
"""
Send data to the web page
Args:
data (str): The data to send.
send_return (bool): Whether to send a return after the data.
"""
cmd = f"window.term.paste({json.dumps(data)});"
if self.page is None:
logger.warning("Cannot write to web console: page is not initialized.")
return
self.page.runJavaScript(cmd)
if send_return:
self.send_return()
def take_page_ownership(self, unique_id: str | None = None):
"""
Take ownership of a web page from the registry. This will transfer the page
from its current owner (if any) to this widget.
Args:
unique_id (str): The unique identifier of the page to take ownership of.
If None, uses this widget's unique_id.
"""
if unique_id is None:
unique_id = self._unique_id
if not unique_id:
logger.warning("Cannot take page ownership without a unique_id")
return
# Get the page from registry
page = _web_console_registry.take_page_ownership(unique_id, self.gui_id)
if not page:
logger.warning(f"Page {unique_id} not found in registry")
return
self.page = page
self.browser.setPage(page)
self._set_mode(ConsoleMode.ACTIVE)
logger.info(f"Widget {self.gui_id} took ownership of page {unique_id}")
def _on_ownership_lost(self):
"""
Called when this widget loses ownership of its page.
Displays the overlay and hides the browser.
"""
self._set_mode(ConsoleMode.INACTIVE)
logger.info(f"Widget {self.gui_id} lost ownership of page {self._unique_id}")
def yield_ownership(self):
"""
Yield ownership of the page. The page remains in the registry with no owner,
available for another widget to claim. This is automatically called when the
widget becomes hidden.
"""
if not self._unique_id:
return
success = _web_console_registry.yield_ownership(self.gui_id)
if success:
self._on_ownership_lost()
logger.info(f"Widget {self.gui_id} yielded ownership of page {self._unique_id}")
def has_ownership(self) -> bool:
"""
Check if this widget currently has ownership of a page.
Returns:
bool: True if this widget owns a page, False otherwise.
"""
if not self._unique_id:
return False
page_info = _web_console_registry.get_page_info(self._unique_id)
if page_info is None:
return False
return page_info.owner_gui_id == self.gui_id
def hideEvent(self, event):
"""
Called when the widget is hidden. Automatically yields ownership.
"""
if self.has_ownership():
self.yield_ownership()
self._set_mode(ConsoleMode.HIDDEN)
super().hideEvent(event)
def showEvent(self, event):
"""
Called when the widget is shown. Updates UI state based on ownership.
"""
super().showEvent(event)
if self._unique_id and not self.has_ownership():
# Take ownership if the page does not have an owner or
# the owner is not visible
page_info = _web_console_registry.get_page_info(self._unique_id)
if page_info is None:
self._set_mode(ConsoleMode.INACTIVE)
return
if page_info.owner_gui_id is None or not _web_console_registry.owner_is_visible(
self._unique_id
):
self.take_page_ownership(self._unique_id)
return
if page_info.owner_gui_id != self.gui_id:
self._set_mode(ConsoleMode.INACTIVE)
return
def resizeEvent(self, event: QResizeEvent) -> None:
super().resizeEvent(event)
self.overlay.resize(event.size())
def mousePressEvent(self, event: QMouseEvent) -> None:
if event.button() == Qt.MouseButton.LeftButton and not self.has_ownership():
self.take_page_ownership(self._unique_id)
event.accept()
return
return super().mousePressEvent(event)
def _authenticate(self, _, auth):
"""
Authenticate the request with the provided username and password.
"""
auth.setUser("user")
auth.setPassword(self._token)
def send_return(self):
"""
Send return to the web page
"""
self.page.runJavaScript(
"document.querySelector('textarea.xterm-helper-textarea').dispatchEvent(new KeyboardEvent('keypress', {charCode: 13}))"
)
def send_ctrl_c(self):
"""
Send Ctrl+C to the web page
"""
self.page.runJavaScript(
"document.querySelector('textarea.xterm-helper-textarea').dispatchEvent(new KeyboardEvent('keypress', {charCode: 3}))"
)
def set_readonly(self, readonly: bool):
"""
Set the web console to read-only mode.
"""
if not isinstance(readonly, bool):
raise ValueError("Readonly must be a boolean.")
self.setEnabled(not readonly)
def cleanup(self):
"""
Clean up the registry by removing any instances that are no longer valid.
"""
self._startup_timer.stop()
_web_console_registry.unregister(self)
super().cleanup()
class BECShell(WebConsole):
"""
A WebConsole pre-configured to run the BEC shell.
We cannot simply expose the web console properties to Qt as we need to have a deterministic
startup behavior for sharing the same shell instance across multiple widgets.
"""
ICON_NAME = "hub"
def __init__(self, parent=None, config=None, client=None, gui_id=None, **kwargs):
super().__init__(
parent=parent,
config=config,
client=client,
gui_id=gui_id,
is_bec_shell=True,
unique_id="bec_shell",
**kwargs,
)
if __name__ == "__main__": # pragma: no cover
import sys
app = QApplication(sys.argv)
widget = QTabWidget()
# Create two consoles with different unique_ids
web_console1 = WebConsole(startup_cmd="bec --nogui", unique_id="console1")
web_console2 = WebConsole(startup_cmd="htop")
web_console3 = WebConsole(startup_cmd="bec --nogui", unique_id="console1")
widget.addTab(web_console1, "Console 1")
widget.addTab(web_console2, "Console 2")
widget.addTab(web_console3, "Console 3 -- mirror of Console 1")
widget.show()
# Demonstrate page sharing:
# After initialization, web_console2 can take ownership of console1's page:
# web_console2.take_page_ownership("console1")
widget.resize(800, 600)
def _close_cons1():
web_console2.close()
web_console2.deleteLater()
# QTimer.singleShot(3000, _close_cons1)
sys.exit(app.exec_())
bec_widgets/widgets/editors/web_console/web_console.pyproject
-1
View File
@@ -1 +0,0 @@
{'files': ['web_console.py']}
bec_widgets/widgets/utility/bec_term/__init__.py
+11
View File
@@ -0,0 +1,11 @@
if __name__ == "__main__": # pragma: no cover
import sys
from pyside6_qtermwidget import QTermWidget # pylint: disable=ungrouped-imports
from qtpy.QtWidgets import QApplication # pylint: disable=ungrouped-imports
app = QApplication(sys.argv)
widget = QTermWidget()
widget.show()
sys.exit(app.exec())
bec_widgets/widgets/utility/bec_term/protocol.py
+8
View File
@@ -0,0 +1,8 @@
from typing import Protocol, runtime_checkable
@runtime_checkable
class BecTerminal(Protocol):
"""Implementors of this protocol must also be subclasses of QWidget"""
def write(self, text: str, add_newline: bool = True): ...
bec_widgets/widgets/utility/bec_term/qtermwidget_wrapper.py
+241
View File
@@ -0,0 +1,241 @@
"""A wrapper for the optional external dependency pyside6_qtermwidget.
Simply displays a message in a QLabel if the dependency is not installed."""
import os
from functools import wraps
from typing import Sequence
from qtpy.QtCore import QIODevice, QPoint, QSize, QUrl, Signal # type: ignore
from qtpy.QtGui import QAction, QFont, QKeyEvent, QResizeEvent, Qt # type: ignore
from qtpy.QtWidgets import QLabel, QVBoxLayout, QWidget
try:
from pyside6_qtermwidget import QTermWidget
except ImportError:
QTermWidget = None
def _forward(func):
"""Apply to a private method to forward the call to the method on QTermWidget with the same name,
(with leading '_' removed) if it is defined, otherwise do nothing."""
@wraps(func)
def wrapper(self, *args, **kwargs):
target = getattr(self, "_main_widget")
method = getattr(target, func.__name__[1:])
if QTermWidget:
return method(*args, **kwargs)
else:
...
return wrapper
class BecQTerm(QWidget):
activity = Signal()
bell = Signal(str)
copy_available = Signal(bool)
current_directory_changed = Signal(str)
finished = Signal()
profile_changed = Signal(str)
received_data = Signal(str)
silence = Signal()
term_got_focus = Signal()
term_key_pressed = Signal(QKeyEvent)
term_lost_focus = Signal()
title_changed = Signal()
url_activated = Signal(QUrl, bool)
def __init__(self, /, parent: QWidget | None = None, **kwargs) -> None:
super().__init__(parent)
self._layout = QVBoxLayout()
self.setLayout(self._layout)
if QTermWidget:
self._main_widget = QTermWidget(parent=self)
self.activity.connect(self._main_widget.activity)
self.bell.connect(self._main_widget.bell)
self.copy_available.connect(self._main_widget.copyAvailable)
self.current_directory_changed.connect(self._main_widget.currentDirectoryChanged)
self.finished.connect(self._main_widget.finished)
self.profile_changed.connect(self._main_widget.profileChanged)
self.received_data.connect(self._main_widget.receivedData)
self.silence.connect(self._main_widget.silence)
self.term_got_focus.connect(self._main_widget.termGetFocus)
self.term_key_pressed.connect(self._main_widget.termKeyPressed)
self.term_lost_focus.connect(self._main_widget.termLostFocus)
self.title_changed.connect(self._main_widget.titleChanged)
self.url_activated.connect(self._main_widget.urlActivated)
self._setEnvironment([f"{k}={v}" for k, v in os.environ.items()])
self._setColorScheme("Solarized")
else:
self._layout.setAlignment(Qt.AlignmentFlag.AlignCenter)
self._main_widget = QLabel("pyside6_qterminal is not installed!")
self._layout.addWidget(self._main_widget)
def write(self, text: str, add_newline: bool = True):
if add_newline:
text += "\n"
self._sendText(text)
# automatically forwarded to the widget only if it exists
@_forward
def _addCustomColorSchemeDir(self, custom_dir: str, /) -> None: ...
@_forward
def _autoHideMouseAfter(self, delay: int, /) -> None: ...
@_forward
def _availableColorSchemes(self) -> list[str]: ...
@_forward
def _availableKeyBindings(self) -> list[str]: ...
@_forward
def _bracketText(self, text: str, /) -> None: ...
@_forward
def _bracketedPasteModeIsDisabled(self, /) -> bool: ...
@_forward
def _changeDir(self, dir: str, /) -> None: ...
@_forward
def _clear(self, /) -> None: ...
@_forward
def _clearCustomKeyBindingsDir(self, /) -> None: ...
@_forward
def _copyClipboard(self, /) -> None: ...
@_forward
def _disableBracketedPasteMode(self, disable: bool, /) -> None: ...
@_forward
def _filterActions(self, position: QPoint, /) -> list[QAction]: ...
@_forward
def _flowControlEnabled(self, /) -> bool: ...
@_forward
def _getAvailableColorSchemes(self, /) -> list[str]: ...
@_forward
def _getForegroundProcessId(self, /) -> int: ...
@_forward
def _getMargin(self, /) -> int: ...
@_forward
def _getPtySlaveFd(self, /) -> int: ...
@_forward
def _getSelectionEnd(self, row: int, column: int, /) -> None: ...
@_forward
def _getSelectionStart(self, row: int, column: int, /) -> None: ...
@_forward
def _getShellPID(self, /) -> int: ...
@_forward
def _getTerminalFont(self, /) -> QFont: ...
@_forward
def _historyLinesCount(self, /) -> int: ...
@_forward
def _historySize(self, /) -> int: ...
@_forward
def _icon(self, /) -> str: ...
@_forward
def _isBidiEnabled(self, /) -> bool: ...
@_forward
def _isTitleChanged(self, /) -> bool: ...
@_forward
def _keyBindings(self, /) -> str: ...
@_forward
def _pasteClipboard(self, /) -> None: ...
@_forward
def _pasteSelection(self, /) -> None: ...
@_forward
def _resizeEvent(self, arg__1: QResizeEvent, /) -> None: ...
@_forward
def _saveHistory(self, device: QIODevice, /) -> None: ...
@_forward
def _screenColumnsCount(self, /) -> int: ...
@_forward
def _screenLinesCount(self, /) -> int: ...
@_forward
def _scrollToEnd(self, /) -> None: ...
@_forward
def _selectedText(self, /, preserveLineBreaks: bool = ...) -> str: ...
@_forward
def _selectionChanged(self, textSelected: bool, /) -> None: ...
@_forward
def _sendKeyEvent(self, e: QKeyEvent, /) -> None: ...
@_forward
def _sendText(self, text: str, /) -> None: ...
@_forward
def _sessionFinished(self, /) -> None: ...
@_forward
def _setArgs(self, args: Sequence[str], /) -> None: ...
@_forward
def _setAutoClose(self, arg__1: bool, /) -> None: ...
@_forward
def _setBidiEnabled(self, enabled: bool, /) -> None: ...
@_forward
def _setBlinkingCursor(self, blink: bool, /) -> None: ...
@_forward
def _setBoldIntense(self, boldIntense: bool, /) -> None: ...
@_forward
def _setColorScheme(self, name: str, /) -> None: ...
@_forward
def _setConfirmMultilinePaste(self, confirmMultilinePaste: bool, /) -> None: ...
@_forward
def _setCustomKeyBindingsDir(self, custom_dir: str, /) -> None: ...
@_forward
def _setDrawLineChars(self, drawLineChars: bool, /) -> None: ...
@_forward
def _setEnvironment(self, environment: Sequence[str], /) -> None: ...
@_forward
def _setFlowControlEnabled(self, enabled: bool, /) -> None: ...
@_forward
def _setFlowControlWarningEnabled(self, enabled: bool, /) -> None: ...
@_forward
def _setHistorySize(self, lines: int, /) -> None: ...
@_forward
def _setKeyBindings(self, kb: str, /) -> None: ...
@_forward
def _setMargin(self, arg__1: int, /) -> None: ...
@_forward
def _setMonitorActivity(self, arg__1: bool, /) -> None: ...
@_forward
def _setMonitorSilence(self, arg__1: bool, /) -> None: ...
@_forward
def _setMotionAfterPasting(self, arg__1: int, /) -> None: ...
@_forward
def _setSelectionEnd(self, row: int, column: int, /) -> None: ...
@_forward
def _setSelectionStart(self, row: int, column: int, /) -> None: ...
@_forward
def _setShellProgram(self, program: str, /) -> None: ...
@_forward
def _setSilenceTimeout(self, seconds: int, /) -> None: ...
@_forward
def _setSize(self, arg__1: QSize, /) -> None: ...
@_forward
def _setTerminalBackgroundImage(self, backgroundImage: str, /) -> None: ...
@_forward
def _setTerminalBackgroundMode(self, mode: int, /) -> None: ...
@_forward
def _setTerminalFont(self, font: QFont | str | Sequence[str], /) -> None: ...
@_forward
def _setTerminalOpacity(self, level: float, /) -> None: ...
@_forward
def _setTerminalSizeHint(self, enabled: bool, /) -> None: ...
@_forward
def _setTrimPastedTrailingNewlines(self, trimPastedTrailingNewlines: bool, /) -> None: ...
@_forward
def _setWordCharacters(self, chars: str, /) -> None: ...
@_forward
def _setWorkingDirectory(self, dir: str, /) -> None: ...
@_forward
def _sizeHint(self, /) -> QSize: ...
@_forward
def _startShellProgram(self, /) -> None: ...
@_forward
def _startTerminalTeletype(self, /) -> None: ...
@_forward
def _terminalSizeHint(self, /) -> bool: ...
@_forward
def _title(self, /) -> str: ...
@_forward
def _toggleShowSearchBar(self, /) -> None: ...
@_forward
def _wordCharacters(self, /) -> str: ...
@_forward
def _workingDirectory(self, /) -> str: ...
@_forward
def _zoomIn(self, /) -> None: ...
@_forward
def _zoomOut(self, /) -> None: ...
bec_widgets/widgets/utility/bec_term/util.py
+6
View File
@@ -0,0 +1,6 @@
from bec_widgets.widgets.utility.bec_term.protocol import BecTerminal
from bec_widgets.widgets.utility.bec_term.qtermwidget_wrapper import BecQTerm
def get_current_bec_term_class() -> type[BecTerminal]:
return BecQTerm
pyproject.toml
+76 -71
View File
@@ -1,54 +1,34 @@
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "bec_widgets"
version = "3.4.4"
description = "BEC Widgets"
requires-python = ">=3.11"
classifiers = [
"Development Status :: 3 - Alpha",
"Programming Language :: Python :: 3",
"Topic :: Scientific/Engineering",
"Development Status :: 3 - Alpha",
"Programming Language :: Python :: 3",
"Topic :: Scientific/Engineering",
]
dependencies = [
"bec_ipython_client~=3.107,>=3.107.2", # needed for jupyter console
"bec_lib~=3.107,>=3.107.2",
"bec_qthemes~=1.0, >=1.3.4",
"black>=26,<27", # needed for bw-generate-cli
"isort>=5.13, <9.0", # needed for bw-generate-cli
"ophyd_devices~=1.29, >=1.29.1",
"pydantic~=2.0",
"pyqtgraph==0.13.7",
"PySide6==6.9.0",
"qtconsole~=5.5, >=5.5.1", # needed for jupyter console
"qtpy~=2.4",
"thefuzz~=0.22",
"qtmonaco~=0.8, >=0.8.1",
"darkdetect~=0.8",
"PySide6-QtAds==4.4.0",
"pylsp-bec~=1.2",
"copier~=9.7",
"typer~=0.15",
"markdown~=3.9",
"PyJWT~=2.9",
]
[project.optional-dependencies]
dev = [
"coverage~=7.0",
"fakeredis~=2.23, >=2.23.2",
"pytest-bec-e2e>=2.21.4, <=4.0",
"pytest-qt~=4.4",
"pytest-random-order~=1.1",
"pytest-timeout~=2.2",
"pytest-xvfb~=3.0",
"pytest~=8.0",
"pytest-cov~=6.1.1",
"watchdog~=6.0",
"pre_commit~=4.2",
"PyJWT~=2.9",
"PySide6==6.9.0",
"PySide6-QtAds==4.4.0",
"bec_ipython_client~=3.107,>=3.107.2", # needed for jupyter console
"bec_lib~=3.107,>=3.107.2",
"bec_qthemes~=1.0, >=1.3.4",
"black>=26,<27", # needed for bw-generate-cli
"copier~=9.7",
"darkdetect~=0.8",
"isort>=5.13, <9.0", # needed for bw-generate-cli
"markdown~=3.9",
"ophyd_devices~=1.29, >=1.29.1",
"pydantic~=2.0",
"pylsp-bec~=1.2",
"pyqtgraph==0.13.7",
"qtconsole~=5.5, >=5.5.1", # needed for jupyter console
"qtmonaco~=0.8, >=0.8.1",
"qtpy~=2.4",
"thefuzz~=0.22",
"typer~=0.15",
]
[project.urls]
@@ -56,10 +36,44 @@ dev = [
Homepage = "https://gitlab.psi.ch/bec/bec_widgets"
[project.scripts]
bw-generate-cli = "bec_widgets.cli.generate_cli:main"
bec-gui-server = "bec_widgets.cli.server:main"
bec-designer = "bec_widgets.utils.bec_designer:main"
bec-app = "bec_widgets.applications.main_app:main"
bec-designer = "bec_widgets.utils.bec_designer:main"
bec-gui-server = "bec_widgets.cli.server:main"
bw-generate-cli = "bec_widgets.cli.generate_cli:main"
[project.optional-dependencies]
dev = [
"coverage~=7.0",
"fakeredis~=2.23, >=2.23.2",
"pytest-bec-e2e>=2.21.4, <=4.0",
"pytest-qt~=4.4",
"pytest-random-order~=1.1",
"pytest-timeout~=2.2",
"pytest-xvfb~=3.0",
"pytest~=8.0",
"pytest-cov~=6.1.1",
"watchdog~=6.0",
"pre_commit~=4.2",
]
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.black]
line-length = 100
skip-magic-trailing-comma = true
[tool.coverage.report]
skip_empty = true # exclude empty *files*, e.g. __init__.py, from the report
exclude_lines = [
"pragma: no cover",
"if TYPE_CHECKING:",
"return NotImplemented",
"raise NotImplementedError",
"\\.\\.\\.",
'if __name__ == "__main__":',
]
[tool.hatch.build.targets.wheel]
include = ["*"]
@@ -69,10 +83,6 @@ exclude = ["docs/**", "tests/**"]
include = ["*"]
exclude = ["docs/**", "tests/**"]
[tool.black]
line-length = 100
skip-magic-trailing-comma = true
[tool.isort]
profile = "black"
line_length = 100
@@ -80,6 +90,12 @@ multi_line_output = 3
include_trailing_comma = true
known_first_party = ["bec_widgets"]
[tool.ruff]
line-length = 100
[tool.ruff.format]
skip-magic-trailing-comma = true
[tool.semantic_release]
build_command = "pip install build wheel && python -m build"
version_toml = ["pyproject.toml:project.version"]
@@ -90,16 +106,16 @@ default = "semantic-release <semantic-release>"
[tool.semantic_release.commit_parser_options]
allowed_tags = [
"build",
"chore",
"ci",
"docs",
"feat",
"fix",
"perf",
"style",
"refactor",
"test",
"build",
"chore",
"ci",
"docs",
"feat",
"fix",
"perf",
"style",
"refactor",
"test",
]
minor_tags = ["feat"]
patch_tags = ["fix", "perf"]
@@ -116,14 +132,3 @@ env = "GH_TOKEN"
[tool.semantic_release.publish]
dist_glob_patterns = ["dist/*"]
upload_to_vcs_release = true
[tool.coverage.report]
skip_empty = true # exclude empty *files*, e.g. __init__.py, from the report
exclude_lines = [
"pragma: no cover",
"if TYPE_CHECKING:",
"return NotImplemented",
"raise NotImplementedError",
"\\.\\.\\.",
'if __name__ == "__main__":',
]
tests/end-2-end/test_rpc_widgets_e2e.py
+2 -2
View File
@@ -93,8 +93,8 @@ def test_available_widgets(qtbot, connected_client_gui_obj):
if object_name == "BECShell":
continue
# Skip WebConsole as ttyd is not installed
if object_name == "WebConsole":
# Skip BecConsole as ttyd is not installed
if object_name == "BecConsole":
continue
#############################
tests/unit_tests/test_dock_area.py
+2 -2
View File
@@ -918,7 +918,7 @@ class TestToolbarFunctionality:
action.trigger()
if action_name == "terminal":
mock_new.assert_called_once_with(
widget="WebConsole", closable=True, startup_cmd=None
widget="BecConsole", closable=True, startup_cmd=None
)
else:
mock_new.assert_called_once_with(widget=widget_type)
@@ -2272,7 +2272,7 @@ class TestFlatToolbarActions:
"flat_queue": "BECQueue",
"flat_status": "BECStatusBox",
"flat_progress_bar": "RingProgressBar",
"flat_terminal": "WebConsole",
"flat_terminal": "BecConsole",
"flat_bec_shell": "BECShell",
"flat_sbb_monitor": "SBBMonitor",
}
tests/unit_tests/test_web_console.py
+65 -65
View File
@@ -5,11 +5,11 @@ from qtpy.QtCore import Qt
from qtpy.QtGui import QHideEvent
from qtpy.QtNetwork import QAuthenticator
from bec_widgets.widgets.editors.web_console.web_console import (
from bec_widgets.widgets.editors.bec_console.bec_console import (
BecConsole,
BECShell,
ConsoleMode,
WebConsole,
_web_console_registry,
_bec_console_registry,
)
from .client_mocks import mocked_client
@@ -19,19 +19,19 @@ from .client_mocks import mocked_client
def mocked_server_startup():
"""Mock the web console server startup process."""
with mock.patch(
"bec_widgets.widgets.editors.web_console.web_console.subprocess"
"bec_widgets.widgets.editors.bec_console.bec_console.subprocess"
) as mock_subprocess:
with mock.patch.object(_web_console_registry, "_wait_for_server_port"):
_web_console_registry._server_port = 12345
with mock.patch.object(_bec_console_registry, "_wait_for_server_port"):
_bec_console_registry._server_port = 12345
yield mock_subprocess
def static_console(qtbot, client, unique_id: str | None = None):
"""Fixture to provide a static unique_id for WebConsole tests."""
"""Fixture to provide a static unique_id for BecConsole tests."""
if unique_id is None:
widget = WebConsole(client=client)
widget = BecConsole(client=client)
else:
widget = WebConsole(client=client, unique_id=unique_id)
widget = BecConsole(client=client, terminal_id=unique_id)
qtbot.addWidget(widget)
qtbot.waitExposed(widget)
return widget
@@ -39,7 +39,7 @@ def static_console(qtbot, client, unique_id: str | None = None):
@pytest.fixture
def console_widget(qtbot, mocked_client, mocked_server_startup):
"""Create a WebConsole widget with mocked server startup."""
"""Create a BecConsole widget with mocked server startup."""
yield static_console(qtbot, mocked_client)
@@ -54,26 +54,26 @@ def bec_shell_widget(qtbot, mocked_client, mocked_server_startup):
@pytest.fixture
def console_widget_with_static_id(qtbot, mocked_client, mocked_server_startup):
"""Create a WebConsole widget with a static unique ID."""
"""Create a BecConsole widget with a static unique ID."""
yield static_console(qtbot, mocked_client, unique_id="test_console")
@pytest.fixture
def two_console_widgets_same_id(qtbot, mocked_client, mocked_server_startup):
"""Create two WebConsole widgets sharing the same unique ID."""
"""Create two BecConsole widgets sharing the same unique ID."""
widget1 = static_console(qtbot, mocked_client, unique_id="shared_console")
widget2 = static_console(qtbot, mocked_client, unique_id="shared_console")
yield widget1, widget2
def test_web_console_widget_initialization(console_widget):
def test_bec_console_widget_initialization(console_widget):
assert (
console_widget.page.url().toString()
== f"http://localhost:{_web_console_registry._server_port}"
== f"http://localhost:{_bec_console_registry._server_port}"
)
def test_web_console_write(console_widget):
def test_bec_console_write(console_widget):
# Test the write method
with mock.patch.object(console_widget.page, "runJavaScript") as mock_run_js:
console_widget.write("Hello, World!")
@@ -81,7 +81,7 @@ def test_web_console_write(console_widget):
assert mock.call('window.term.paste("Hello, World!");') in mock_run_js.mock_calls
def test_web_console_write_no_return(console_widget):
def test_bec_console_write_no_return(console_widget):
# Test the write method with send_return=False
with mock.patch.object(console_widget.page, "runJavaScript") as mock_run_js:
console_widget.write("Hello, World!", send_return=False)
@@ -90,7 +90,7 @@ def test_web_console_write_no_return(console_widget):
assert mock_run_js.call_count == 1
def test_web_console_send_return(console_widget):
def test_bec_console_send_return(console_widget):
# Test the send_return method
with mock.patch.object(console_widget.page, "runJavaScript") as mock_run_js:
console_widget.send_return()
@@ -100,7 +100,7 @@ def test_web_console_send_return(console_widget):
assert mock_run_js.call_count == 1
def test_web_console_send_ctrl_c(console_widget):
def test_bec_console_send_ctrl_c(console_widget):
# Test the send_ctrl_c method
with mock.patch.object(console_widget.page, "runJavaScript") as mock_run_js:
console_widget.send_ctrl_c()
@@ -110,31 +110,31 @@ def test_web_console_send_ctrl_c(console_widget):
assert mock_run_js.call_count == 1
def test_web_console_authenticate(console_widget):
def test_bec_console_authenticate(console_widget):
# Test the _authenticate method
token = _web_console_registry._token
token = _bec_console_registry._token
mock_auth = mock.MagicMock(spec=QAuthenticator)
console_widget._authenticate(None, mock_auth)
mock_auth.setUser.assert_called_once_with("user")
mock_auth.setPassword.assert_called_once_with(token)
def test_web_console_registry_wait_for_server_port():
def test_bec_console_registry_wait_for_server_port():
# Test the _wait_for_server_port method
with mock.patch.object(_web_console_registry, "_server_process") as mock_subprocess:
with mock.patch.object(_bec_console_registry, "_server_process") as mock_subprocess:
mock_subprocess.stderr.readline.side_effect = [b"Starting", b"Listening on port: 12345"]
_web_console_registry._wait_for_server_port()
assert _web_console_registry._server_port == 12345
_bec_console_registry._wait_for_server_port()
assert _bec_console_registry._server_port == 12345
def test_web_console_registry_wait_for_server_port_timeout():
def test_bec_console_registry_wait_for_server_port_timeout():
# Test the _wait_for_server_port method with timeout
with mock.patch.object(_web_console_registry, "_server_process") as mock_subprocess:
with mock.patch.object(_bec_console_registry, "_server_process") as mock_subprocess:
with pytest.raises(TimeoutError):
_web_console_registry._wait_for_server_port(timeout=0.1)
_bec_console_registry._wait_for_server_port(timeout=0.1)
def test_web_console_startup_command_execution(console_widget, qtbot):
def test_bec_console_startup_command_execution(console_widget, qtbot):
"""Test that the startup command is triggered after successful initialization."""
# Set a custom startup command
console_widget.startup_cmd = "test startup command"
@@ -196,7 +196,7 @@ def test_bec_shell_startup_contains_gui_id(bec_shell_widget):
assert bec_shell.startup_cmd == "bec --gui-id test_gui_id"
def test_web_console_set_readonly(console_widget):
def test_bec_console_set_readonly(console_widget):
# Test the set_readonly method
console_widget.set_readonly(True)
assert not console_widget.isEnabled()
@@ -205,31 +205,31 @@ def test_web_console_set_readonly(console_widget):
assert console_widget.isEnabled()
def test_web_console_with_unique_id(console_widget_with_static_id):
"""Test creating a WebConsole with a unique_id."""
def test_bec_console_with_unique_id(console_widget_with_static_id):
"""Test creating a BecConsole with a unique_id."""
widget = console_widget_with_static_id
assert widget._unique_id == "test_console"
assert widget._unique_id in _web_console_registry._page_registry
page_info = _web_console_registry.get_page_info("test_console")
assert widget._unique_id in _bec_console_registry._page_registry
page_info = _bec_console_registry.get_page_info("test_console")
assert page_info is not None
assert page_info.owner_gui_id == widget.gui_id
assert widget.gui_id in page_info.widget_ids
def test_web_console_page_sharing(two_console_widgets_same_id):
def test_bec_console_page_sharing(two_console_widgets_same_id):
"""Test that two widgets can share the same page using unique_id."""
widget1, widget2 = two_console_widgets_same_id
# Both should reference the same page in the registry
page_info = _web_console_registry.get_page_info("shared_console")
page_info = _bec_console_registry.get_page_info("shared_console")
assert page_info is not None
assert widget1.gui_id in page_info.widget_ids
assert widget2.gui_id in page_info.widget_ids
assert widget1.page == widget2.page
def test_web_console_has_ownership(console_widget_with_static_id):
def test_bec_console_has_ownership(console_widget_with_static_id):
"""Test the has_ownership method."""
widget = console_widget_with_static_id
@@ -237,7 +237,7 @@ def test_web_console_has_ownership(console_widget_with_static_id):
assert widget.has_ownership()
def test_web_console_yield_ownership(console_widget_with_static_id):
def test_bec_console_yield_ownership(console_widget_with_static_id):
"""Test yielding ownership of a page."""
widget = console_widget_with_static_id
@@ -248,13 +248,13 @@ def test_web_console_yield_ownership(console_widget_with_static_id):
# Widget should no longer have ownership
assert not widget.has_ownership()
page_info = _web_console_registry.get_page_info("test_console")
page_info = _bec_console_registry.get_page_info("test_console")
assert page_info.owner_gui_id is None
# Overlay should be shown
assert widget._mode == ConsoleMode.INACTIVE
def test_web_console_take_page_ownership(two_console_widgets_same_id):
def test_bec_console_take_page_ownership(two_console_widgets_same_id):
"""Test taking ownership of a page."""
widget1, widget2 = two_console_widgets_same_id
@@ -273,7 +273,7 @@ def test_web_console_take_page_ownership(two_console_widgets_same_id):
assert widget1._mode == ConsoleMode.INACTIVE
def test_web_console_hide_event_yields_ownership(qtbot, console_widget_with_static_id):
def test_bec_console_hide_event_yields_ownership(qtbot, console_widget_with_static_id):
"""Test that hideEvent yields ownership."""
widget = console_widget_with_static_id
@@ -287,11 +287,11 @@ def test_web_console_hide_event_yields_ownership(qtbot, console_widget_with_stat
# Widget should have yielded ownership
assert not widget.has_ownership()
page_info = _web_console_registry.get_page_info("test_console")
page_info = _bec_console_registry.get_page_info("test_console")
assert page_info.owner_gui_id is None
def test_web_console_show_event_takes_ownership(console_widget_with_static_id):
def test_bec_console_show_event_takes_ownership(console_widget_with_static_id):
"""Test that showEvent takes ownership when page has no owner."""
widget = console_widget_with_static_id
@@ -308,7 +308,7 @@ def test_web_console_show_event_takes_ownership(console_widget_with_static_id):
assert not widget.overlay.isVisible()
def test_web_console_mouse_press_takes_ownership(qtbot, two_console_widgets_same_id):
def test_bec_console_mouse_press_takes_ownership(qtbot, two_console_widgets_same_id):
"""Test that clicking on overlay takes ownership."""
widget1, widget2 = two_console_widgets_same_id
widget1.show()
@@ -328,20 +328,20 @@ def test_web_console_mouse_press_takes_ownership(qtbot, two_console_widgets_same
assert not widget1.has_ownership()
def test_web_console_registry_cleanup_removes_page(console_widget_with_static_id):
def test_bec_console_registry_cleanup_removes_page(console_widget_with_static_id):
"""Test that the registry cleans up pages when all widgets are removed."""
widget = console_widget_with_static_id
assert widget._unique_id in _web_console_registry._page_registry
assert widget._unique_id in _bec_console_registry._page_registry
# Cleanup the widget
widget.cleanup()
# Page should be removed from registry
assert widget._unique_id not in _web_console_registry._page_registry
assert widget._unique_id not in _bec_console_registry._page_registry
def test_web_console_without_unique_id_no_page_sharing(console_widget):
def test_bec_console_without_unique_id_no_page_sharing(console_widget):
"""Test that widgets without unique_id don't participate in page sharing."""
widget = console_widget
@@ -350,20 +350,20 @@ def test_web_console_without_unique_id_no_page_sharing(console_widget):
assert not widget.has_ownership() # Should return False for non-unique widgets
def test_web_console_registry_get_page_info_nonexistent(qtbot, mocked_client):
def test_bec_console_registry_get_page_info_nonexistent(qtbot, mocked_client):
"""Test getting page info for a non-existent page."""
page_info = _web_console_registry.get_page_info("nonexistent")
page_info = _bec_console_registry.get_page_info("nonexistent")
assert page_info is None
def test_web_console_take_ownership_without_unique_id(console_widget):
def test_bec_console_take_ownership_without_unique_id(console_widget):
"""Test that take_page_ownership fails gracefully without unique_id."""
widget = console_widget
# Should not crash when taking ownership without unique_id
widget.take_page_ownership()
def test_web_console_yield_ownership_without_unique_id(console_widget):
def test_bec_console_yield_ownership_without_unique_id(console_widget):
"""Test that yield_ownership fails gracefully without unique_id."""
widget = console_widget
# Should not crash when yielding ownership without unique_id
@@ -372,7 +372,7 @@ def test_web_console_yield_ownership_without_unique_id(console_widget):
def test_registry_yield_ownership_gui_id_not_in_instances():
"""Test registry yield_ownership returns False when gui_id not in instances."""
result = _web_console_registry.yield_ownership("nonexistent_gui_id")
result = _bec_console_registry.yield_ownership("nonexistent_gui_id")
assert result is False
@@ -382,9 +382,9 @@ def test_registry_yield_ownership_instance_is_none(console_widget_with_static_id
gui_id = widget.gui_id
# Store the gui_id and simulate the weakref being dead
_web_console_registry._instances[gui_id] = lambda: None
_bec_console_registry._consoles[gui_id] = lambda: None
result = _web_console_registry.yield_ownership(gui_id)
result = _bec_console_registry.yield_ownership(gui_id)
assert result is False
@@ -395,7 +395,7 @@ def test_registry_yield_ownership_unique_id_none(console_widget_with_static_id):
unique_id = widget._unique_id
widget._unique_id = None
result = _web_console_registry.yield_ownership(gui_id)
result = _bec_console_registry.yield_ownership(gui_id)
assert result is False
widget._unique_id = unique_id # Restore for cleanup
@@ -408,7 +408,7 @@ def test_registry_yield_ownership_unique_id_not_in_page_registry(console_widget_
unique_id = widget._unique_id
widget._unique_id = "nonexistent_unique_id"
result = _web_console_registry.yield_ownership(gui_id)
result = _bec_console_registry.yield_ownership(gui_id)
assert result is False
widget._unique_id = unique_id # Restore for cleanup
@@ -416,7 +416,7 @@ def test_registry_yield_ownership_unique_id_not_in_page_registry(console_widget_
def test_registry_owner_is_visible_page_info_none():
"""Test owner_is_visible returns False when page info doesn't exist."""
result = _web_console_registry.owner_is_visible("nonexistent_page")
result = _bec_console_registry.owner_is_visible("nonexistent_page")
assert result is False
@@ -426,10 +426,10 @@ def test_registry_owner_is_visible_no_owner(console_widget_with_static_id):
# Yield ownership so there's no owner
widget.yield_ownership()
page_info = _web_console_registry.get_page_info(widget._unique_id)
page_info = _bec_console_registry.get_page_info(widget._unique_id)
assert page_info.owner_gui_id is None
result = _web_console_registry.owner_is_visible(widget._unique_id)
result = _bec_console_registry.owner_is_visible(widget._unique_id)
assert result is False
@@ -439,9 +439,9 @@ def test_registry_owner_is_visible_owner_ref_none(console_widget_with_static_id)
unique_id = widget._unique_id
# Remove owner from instances dict
del _web_console_registry._instances[widget.gui_id]
del _bec_console_registry._consoles[widget.gui_id]
result = _web_console_registry.owner_is_visible(unique_id)
result = _bec_console_registry.owner_is_visible(unique_id)
assert result is False
@@ -452,9 +452,9 @@ def test_registry_owner_is_visible_owner_instance_none(console_widget_with_stati
gui_id = widget.gui_id
# Simulate dead weakref
_web_console_registry._instances[gui_id] = lambda: None
_bec_console_registry._consoles[gui_id] = lambda: None
result = _web_console_registry.owner_is_visible(unique_id)
result = _bec_console_registry.owner_is_visible(unique_id)
assert result is False
@@ -463,7 +463,7 @@ def test_registry_owner_is_visible_owner_visible(console_widget_with_static_id):
widget = console_widget_with_static_id
widget.show()
result = _web_console_registry.owner_is_visible(widget._unique_id)
result = _bec_console_registry.owner_is_visible(widget._unique_id)
assert result is True
@@ -472,5 +472,5 @@ def test_registry_owner_is_visible_owner_not_visible(console_widget_with_static_
widget = console_widget_with_static_id
widget.hide()
result = _web_console_registry.owner_is_visible(widget._unique_id)
result = _bec_console_registry.owner_is_visible(widget._unique_id)
assert result is False