Dom/rpa_vision_v3
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
ad7ff3bce4b7bfcde862ce273d3ce6b9181dafdb
rpa_vision_v3/agent_v0/lea_ui/main_window.py
Dom ae65be2555 chore: ajouter agent_v0/ au tracking git (était un repo embarqué) 
Suppression du .git embarqué dans agent_v0/ — le code est maintenant
tracké normalement dans le repo principal.
Inclut : agent_v1 (client), server_v1 (streaming), lea_ui (chat client)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-18 11:12:23 +01:00

773 lines
27 KiB
Python
Raw Blame History

# agent_v0/lea_ui/main_window.py
"""
Fenetre principale du panneau Lea.
Panneau semi-transparent, ancre a droite de l'ecran, toujours visible.
Peut etre reduit en mini-barre flottante (avatar + indicateur status).
Sections :
- Header : avatar "L" + status connexion
- Zone de chat : messages entrants/sortants (natif PyQt5)
- Zone de status : progression du replay
- Boutons rapides : "Apprends-moi", "Que sais-tu faire ?"
"""
from __future__ import annotations
import logging
from typing import Dict, Any, Optional
from PyQt5.QtCore import (
QPoint,
QPropertyAnimation,
QRect,
QSize,
Qt,
QTimer,
pyqtSignal,
pyqtSlot,
)
from PyQt5.QtGui import (
QColor,
QFont,
QIcon,
QKeySequence,
QPainter,
QPainterPath,
QPen,
)
from PyQt5.QtWidgets import (
QAction,
QApplication,
QDesktopWidget,
QFrame,
QGraphicsDropShadowEffect,
QHBoxLayout,
QLabel,
QProgressBar,
QPushButton,
QShortcut,
QSizePolicy,
QVBoxLayout,
QWidget,
)
from . import styles
from .chat_widget import ChatWidget
from .overlay import OverlayWidget
from .server_client import LeaServerClient
logger = logging.getLogger("lea_ui.main_window")
class LeaAvatar(QWidget):
"""Avatar rond avec l'initiale 'L'."""
def __init__(self, size: int = 40, parent: Optional[QWidget] = None) -> None:
super().__init__(parent)
self._size = size
self._connected = False
self.setFixedSize(size, size)
def set_connected(self, connected: bool) -> None:
self._connected = connected
self.update()
def paintEvent(self, event) -> None: # noqa: N802
painter = QPainter(self)
painter.setRenderHint(QPainter.Antialiasing, True)
# Cercle de fond
painter.setBrush(QColor(styles.COLOR_ACCENT))
painter.setPen(Qt.NoPen)
painter.drawEllipse(2, 2, self._size - 4, self._size - 4)
# Initiale "L"
painter.setPen(QColor(styles.COLOR_TEXT_ON_ACCENT))
font = QFont(styles.FONT_FAMILY, self._size // 3, QFont.Bold)
painter.setFont(font)
painter.drawText(
QRect(0, 0, self._size, self._size),
Qt.AlignCenter,
"L",
)
# Indicateur de connexion (petit cercle en bas a droite)
indicator_size = 12
ix = self._size - indicator_size - 1
iy = self._size - indicator_size - 1
indicator_color = (
QColor(styles.COLOR_SUCCESS) if self._connected
else QColor(styles.COLOR_ERROR)
)
painter.setBrush(indicator_color)
painter.setPen(QPen(QColor(styles.COLOR_BG), 2))
painter.drawEllipse(ix, iy, indicator_size, indicator_size)
painter.end()
class LeaMainWindow(QWidget):
"""Panneau principal de l'interface Lea.
Fenetre semi-transparente, ancree a droite de l'ecran.
Peut basculer en mode mini-barre.
"""
# Signal pour les actions de replay a afficher sur l'overlay
replay_action_received = pyqtSignal(dict)
def __init__(
self,
server_client: Optional[LeaServerClient] = None,
parent: Optional[QWidget] = None,
) -> None:
super().__init__(parent)
# Client serveur
self._client = server_client or LeaServerClient()
# Overlay de feedback
self._overlay = OverlayWidget()
# Mode courant
self._minimized = False
# Setup
self._setup_window()
self._setup_ui()
self._setup_shortcuts()
self._connect_signals()
self._start_connection_check()
# Message d'accueil
QTimer.singleShot(500, self._show_welcome)
# ---------------------------------------------------------------------------
# Setup
# ---------------------------------------------------------------------------
def _setup_window(self) -> None:
"""Configurer les proprietes de la fenetre."""
self.setWindowFlags(
Qt.WindowStaysOnTopHint
| Qt.FramelessWindowHint
| Qt.Tool
)
self.setAttribute(Qt.WA_TranslucentBackground, True)
self.setObjectName("LeaMainWindow")
# Dimensions et position (ancre a droite)
self.setFixedWidth(styles.PANEL_WIDTH)
self.setMinimumHeight(styles.PANEL_MIN_HEIGHT)
self._anchor_to_right()
# Ombre portee
shadow = QGraphicsDropShadowEffect()
shadow.setBlurRadius(20)
shadow.setColor(QColor(0, 0, 0, 60))
shadow.setOffset(0, 4)
self.setGraphicsEffect(shadow)
def _anchor_to_right(self) -> None:
"""Positionner le panneau ancre a droite de l'ecran."""
desktop = QApplication.desktop()
if desktop:
screen_rect = desktop.availableGeometry(desktop.primaryScreen())
x = screen_rect.right() - styles.PANEL_WIDTH - 10
y = screen_rect.top() + 40
height = screen_rect.height() - 80
self.setGeometry(x, y, styles.PANEL_WIDTH, height)
def _setup_ui(self) -> None:
"""Construire l'interface du panneau."""
# Conteneur principal avec fond et coins arrondis
self._main_layout = QVBoxLayout(self)
self._main_layout.setContentsMargins(0, 0, 0, 0)
self._main_layout.setSpacing(0)
# Widget de fond (pour appliquer le style)
self._bg_widget = QWidget()
self._bg_widget.setObjectName("LeaPanelBg")
self._bg_widget.setStyleSheet(f"""
QWidget#LeaPanelBg {{
background-color: {styles.COLOR_BG};
border-radius: {styles.BORDER_RADIUS}px;
border: 1px solid {styles.COLOR_BORDER};
}}
""")
bg_layout = QVBoxLayout(self._bg_widget)
bg_layout.setContentsMargins(0, 0, 0, 0)
bg_layout.setSpacing(0)
# --- Header ---
self._header = self._create_header()
bg_layout.addWidget(self._header)
# --- Chat ---
self._chat = ChatWidget()
bg_layout.addWidget(self._chat, stretch=1)
# --- Zone de status replay ---
self._status_bar = self._create_status_bar()
bg_layout.addWidget(self._status_bar)
# --- Boutons rapides ---
self._quick_buttons = self._create_quick_buttons()
bg_layout.addWidget(self._quick_buttons)
self._main_layout.addWidget(self._bg_widget)
# --- Mini-barre (cachee par defaut) ---
self._mini_bar = self._create_mini_bar()
self._mini_bar.hide()
self._main_layout.addWidget(self._mini_bar)
def _create_header(self) -> QWidget:
"""Creer le header avec avatar et status."""
header = QWidget()
header.setObjectName("LeaHeader")
header.setStyleSheet(styles.HEADER_STYLE)
header.setFixedHeight(60)
layout = QHBoxLayout(header)
layout.setContentsMargins(
styles.PADDING, styles.SPACING,
styles.PADDING, styles.SPACING,
)
# Avatar
self._avatar = LeaAvatar(styles.AVATAR_SIZE)
layout.addWidget(self._avatar)
# Titre + status
text_layout = QVBoxLayout()
text_layout.setSpacing(2)
title = QLabel("Lea")
title.setObjectName("LeaTitle")
title.setStyleSheet(styles.HEADER_STYLE)
text_layout.addWidget(title)
self._status_label = QLabel("Connexion...")
self._status_label.setObjectName("LeaStatus")
self._status_label.setStyleSheet(styles.HEADER_STYLE)
text_layout.addWidget(self._status_label)
layout.addLayout(text_layout, stretch=1)
# Bouton reduire
minimize_btn = QPushButton("_")
minimize_btn.setFixedSize(30, 30)
minimize_btn.setCursor(Qt.PointingHandCursor)
minimize_btn.setStyleSheet(f"""
QPushButton {{
background: transparent;
color: {styles.COLOR_TEXT_SECONDARY};
border: none;
border-radius: 15px;
font-size: 16px;
font-weight: bold;
}}
QPushButton:hover {{
background-color: {styles.COLOR_BORDER};
}}
""")
minimize_btn.clicked.connect(self.toggle_minimize)
layout.addWidget(minimize_btn)
return header
def _create_status_bar(self) -> QWidget:
"""Creer la barre de status du replay."""
container = QWidget()
container.setFixedHeight(50)
layout = QVBoxLayout(container)
layout.setContentsMargins(
styles.PADDING, styles.SPACING,
styles.PADDING, styles.SPACING,
)
layout.setSpacing(4)
self._replay_label = QLabel("")
self._replay_label.setObjectName("StatusLabel")
self._replay_label.setStyleSheet(styles.STATUS_LABEL_STYLE)
self._replay_label.hide()
layout.addWidget(self._replay_label)
self._progress_bar = QProgressBar()
self._progress_bar.setStyleSheet(styles.PROGRESS_STYLE)
self._progress_bar.setTextVisible(False)
self._progress_bar.hide()
layout.addWidget(self._progress_bar)
container.hide()
self._status_container = container
return container
def _create_quick_buttons(self) -> QWidget:
"""Creer les boutons d'action rapide."""
container = QWidget()
layout = QHBoxLayout(container)
layout.setContentsMargins(
styles.PADDING, styles.SPACING,
styles.PADDING, styles.PADDING,
)
layout.setSpacing(styles.SPACING)
btn_learn = QPushButton("Apprends-moi")
btn_learn.setObjectName("QuickButton")
btn_learn.setStyleSheet(styles.QUICK_BUTTON_STYLE)
btn_learn.setCursor(Qt.PointingHandCursor)
btn_learn.clicked.connect(self._on_learn_clicked)
layout.addWidget(btn_learn)
btn_list = QPushButton("Que sais-tu faire ?")
btn_list.setObjectName("QuickButton")
btn_list.setStyleSheet(styles.QUICK_BUTTON_STYLE)
btn_list.setCursor(Qt.PointingHandCursor)
btn_list.clicked.connect(self._on_list_clicked)
layout.addWidget(btn_list)
return container
def _create_mini_bar(self) -> QWidget:
"""Creer la mini-barre flottante (mode reduit)."""
bar = QWidget()
bar.setObjectName("MiniBar")
bar.setStyleSheet(styles.MINI_BAR_STYLE)
bar.setFixedSize(80, 50)
layout = QHBoxLayout(bar)
layout.setContentsMargins(8, 4, 8, 4)
mini_avatar = LeaAvatar(32)
self._mini_avatar = mini_avatar
layout.addWidget(mini_avatar)
expand_btn = QPushButton(">")
expand_btn.setFixedSize(24, 24)
expand_btn.setCursor(Qt.PointingHandCursor)
expand_btn.setStyleSheet(f"""
QPushButton {{
background: transparent;
color: {styles.COLOR_TEXT_SECONDARY};
border: none;
font-size: 14px;
font-weight: bold;
}}
QPushButton:hover {{
color: {styles.COLOR_ACCENT};
}}
""")
expand_btn.clicked.connect(self.toggle_minimize)
layout.addWidget(expand_btn)
return bar
def _setup_shortcuts(self) -> None:
"""Configurer les raccourcis globaux."""
# Ctrl+Shift+L pour afficher/cacher
# Note : Sur Windows, les raccourcis globaux necessitent
# un mecanisme supplementaire (keyboard hook). Ici on utilise
# le raccourci local qui fonctionne quand le panneau a le focus.
# Un hook global sera ajoute dans le launcher.
shortcut = QShortcut(QKeySequence("Ctrl+Shift+L"), self)
shortcut.activated.connect(self.toggle_visibility)
def _connect_signals(self) -> None:
"""Connecter les signaux internes."""
# Chat
self._chat.message_sent.connect(self._on_message_sent)
# Client serveur
self._client.set_on_connection_change(self._on_connection_changed)
self._client.set_on_replay_action(self._on_replay_action)
# Overlay
self._overlay.action_display_finished.connect(self._on_overlay_finished)
# Replay via signal (thread-safe)
self.replay_action_received.connect(self._handle_replay_action)
def _start_connection_check(self) -> None:
"""Demarrer le timer de verification de connexion."""
self._conn_timer = QTimer(self)
self._conn_timer.timeout.connect(self._check_connection)
self._conn_timer.start(10000) # Toutes les 10 secondes
# Premiere verification immediatement
QTimer.singleShot(1000, self._check_connection)
# ---------------------------------------------------------------------------
# Actions
# ---------------------------------------------------------------------------
def _show_welcome(self) -> None:
"""Afficher le message d'accueil."""
self._chat.add_lea_message(
"Bonjour ! Je suis <b>Lea</b>, votre assistante RPA.<br>"
"Je peux apprendre vos taches, les rejouer, "
"et vous montrer ce que je fais.<br><br>"
"Que souhaitez-vous faire ?"
)
@pyqtSlot(str)
def _on_message_sent(self, message: str) -> None:
"""Traiter un message envoye par l'utilisateur."""
self._chat.set_input_enabled(False)
# Envoyer au serveur dans un timer pour ne pas bloquer
QTimer.singleShot(100, lambda: self._send_to_server(message))
def _send_to_server(self, message: str) -> None:
"""Envoyer le message au serveur et afficher la reponse."""
response = self._client.send_chat_message(message)
if response is None:
self._chat.add_lea_message(
"Je n'arrive pas a joindre le serveur. "
"Verifiez que le serveur Linux est demarre."
)
elif "error" in response:
self._chat.add_lea_message(
f"Erreur : {response['error']}"
)
else:
# Extraire la reponse textuelle
reply_text = response.get("response", "")
if not reply_text:
# Construire une reponse a partir des donnees structurees
reply_text = self._format_response(response)
self._chat.add_lea_message(reply_text)
# Si un workflow a ete lance, mettre a jour la status bar
if response.get("success") and response.get("workflow"):
self._show_replay_status(
f"Execution : {response['workflow']}",
0, 1,
)
self._chat.set_input_enabled(True)
def _format_response(self, data: Dict[str, Any]) -> str:
"""Formater une reponse structuree du serveur en texte lisible."""
# Reponse de confirmation
if data.get("needs_confirmation"):
conf = data.get("confirmation", {})
return (
f"Voulez-vous que j'execute <b>{conf.get('workflow_name', '?')}</b> ?<br>"
f"Risque : {conf.get('risk_level', 'normal')}<br>"
"Repondez <b>oui</b> ou <b>non</b>."
)
# Liste de workflows
if "workflows" in data:
workflows = data["workflows"]
if not workflows:
return "Je ne connais aucun workflow pour le moment."
items = []
for wf in workflows[:10]:
name = wf.get("name", wf.get("id", "?"))
desc = wf.get("description", "")
items.append(f"- <b>{name}</b>{': ' + desc if desc else ''}")
result = "Voici ce que je sais faire :<br>" + "<br>".join(items)
if len(workflows) > 10:
result += f"<br><i>... et {len(workflows) - 10} autres</i>"
return result
# Workflow non trouve
if data.get("not_found"):
return (
f"Je ne trouve pas de workflow correspondant a "
f"'{data.get('query', '?')}'.<br>"
"Essayez 'Que sais-tu faire ?' pour voir la liste."
)
# Execution reussie
if data.get("success"):
return (
f"C'est parti ! J'execute <b>{data.get('workflow', '?')}</b>.<br>"
"Regardez l'ecran, je vais vous montrer ce que je fais."
)
# Confirmation/refus
if data.get("confirmed"):
return f"D'accord, je lance <b>{data.get('workflow', '?')}</b> !"
if data.get("denied"):
return "Pas de probleme, j'annule."
# Fallback
return str(data)
def _on_learn_clicked(self) -> None:
"""Action du bouton 'Apprends-moi'."""
self._chat.add_user_message("Apprends-moi une nouvelle tache")
self._chat.add_lea_message(
"D'accord ! Pour m'apprendre une tache :<br>"
"1. Cliquez sur <b>Demarrer</b> dans le tray Agent V1<br>"
"2. Effectuez votre tache normalement<br>"
"3. Cliquez sur <b>Terminer</b> quand c'est fini<br><br>"
"Je vais observer et apprendre automatiquement."
)
def _on_list_clicked(self) -> None:
"""Action du bouton 'Que sais-tu faire ?'."""
self._chat.add_user_message("Que sais-tu faire ?")
self._chat.set_input_enabled(False)
QTimer.singleShot(100, self._fetch_workflows)
def _fetch_workflows(self) -> None:
"""Recuperer et afficher la liste des workflows."""
workflows = self._client.list_workflows()
if workflows:
items = []
for wf in workflows[:15]:
name = wf.get("name", wf.get("id", "?"))
desc = wf.get("description", "")
items.append(f"- <b>{name}</b>{': ' + desc if desc else ''}")
text = "Voici les workflows que je connais :<br>" + "<br>".join(items)
if len(workflows) > 15:
text += f"<br><i>... et {len(workflows) - 15} autres</i>"
else:
text = (
"Je ne connais aucun workflow pour le moment.<br>"
"Apprenez-moi une tache avec le bouton 'Apprends-moi' !"
)
self._chat.add_lea_message(text)
self._chat.set_input_enabled(True)
# ---------------------------------------------------------------------------
# Connexion
# ---------------------------------------------------------------------------
def _check_connection(self) -> None:
"""Verifier la connexion au serveur (dans un timer)."""
connected = self._client.check_connection()
self._update_connection_ui(connected)
def _on_connection_changed(self, connected: bool) -> None:
"""Callback quand l'etat de connexion change."""
# Appeler dans le thread principal via QTimer
QTimer.singleShot(0, lambda: self._update_connection_ui(connected))
def _update_connection_ui(self, connected: bool) -> None:
"""Mettre a jour l'UI selon l'etat de connexion."""
self._avatar.set_connected(connected)
if hasattr(self, '_mini_avatar'):
self._mini_avatar.set_connected(connected)
if connected:
self._status_label.setText(
f"Connecte a {self._client.server_host}"
)
self._status_label.setStyleSheet(
f"color: {styles.COLOR_SUCCESS}; "
f"font-family: '{styles.FONT_FAMILY}'; "
f"font-size: {styles.FONT_SIZE_SMALL}px; "
f"background: transparent; border: none;"
)
else:
error = self._client.last_error or "Serveur injoignable"
self._status_label.setText(f"Deconnecte ({error[:30]})")
self._status_label.setStyleSheet(
f"color: {styles.COLOR_ERROR}; "
f"font-family: '{styles.FONT_FAMILY}'; "
f"font-size: {styles.FONT_SIZE_SMALL}px; "
f"background: transparent; border: none;"
)
# ---------------------------------------------------------------------------
# Replay & Overlay
# ---------------------------------------------------------------------------
def _on_replay_action(self, action: Dict[str, Any]) -> None:
"""Callback appelee depuis le thread de polling (pas thread-safe).
Emettre un signal pour traiter dans le thread Qt.
"""
self.replay_action_received.emit(action)
@pyqtSlot(dict)
def _handle_replay_action(self, action: Dict[str, Any]) -> None:
"""Traiter une action de replay dans le thread Qt.
Afficher l'overlay AVANT l'execution pour que l'utilisateur
voie ce qui va se passer.
"""
action_type = action.get("type", "?")
action_text = self._describe_action(action)
# Calculer les coordonnees ecran
desktop = QApplication.desktop()
screen = desktop.screenGeometry(desktop.primaryScreen()) if desktop else None
if screen:
sw, sh = screen.width(), screen.height()
else:
sw, sh = 1920, 1080
target_x = int(action.get("x_pct", 0.5) * sw)
target_y = int(action.get("y_pct", 0.5) * sh)
# Recuperer la progression depuis le replay status
replay = self._client.get_replay_status()
step_current = 0
step_total = 0
if replay:
step_total = replay.get("total_actions", 0)
step_current = replay.get("completed_actions", 0) + 1
# Mettre a jour la status bar
self._show_replay_status(action_text, step_current, step_total)
# Afficher l'overlay
self._overlay.show_action(
target_x, target_y,
action_text,
step_current, step_total,
duration_ms=1500,
)
# Ajouter dans le chat
self._chat.add_system_message(
f"Etape {step_current}/{step_total} : {action_text}"
)
def _describe_action(self, action: Dict[str, Any]) -> str:
"""Generer une description lisible d'une action de replay."""
action_type = action.get("type", "?")
target_text = action.get("target_text", "")
target_role = action.get("target_role", "")
if action_type == "click":
target = target_text or target_role or "cet element"
return f"Je clique sur [{target}]"
elif action_type == "type":
text = action.get("text", "")
preview = text[:30] + "..." if len(text) > 30 else text
return f"Je tape : {preview}"
elif action_type == "key_combo":
keys = action.get("keys", [])
return f"Je tape : {'+'.join(keys)}"
elif action_type == "scroll":
return "Je fais defiler la page"
elif action_type == "wait":
ms = action.get("duration_ms", 500)
return f"J'attends {ms}ms"
else:
return f"Action : {action_type}"
def _on_overlay_finished(self) -> None:
"""Callback quand l'overlay a fini d'afficher une action."""
pass # L'executor continue de son cote
def _show_replay_status(
self, text: str, current: int, total: int,
) -> None:
"""Afficher la barre de progression du replay."""
self._status_container.show()
self._replay_label.show()
self._replay_label.setText(text)
if total > 0:
self._progress_bar.show()
self._progress_bar.setMaximum(total)
self._progress_bar.setValue(current)
else:
self._progress_bar.hide()
def hide_replay_status(self) -> None:
"""Masquer la barre de progression du replay."""
self._status_container.hide()
# ---------------------------------------------------------------------------
# Visibilite
# ---------------------------------------------------------------------------
def toggle_visibility(self) -> None:
"""Afficher/cacher le panneau (raccourci Ctrl+Shift+L)."""
if self.isVisible():
self.hide()
else:
self.show()
self.raise_()
self.activateWindow()
def toggle_minimize(self) -> None:
"""Basculer entre panneau complet et mini-barre."""
if self._minimized:
# Restaurer
self._mini_bar.hide()
self._bg_widget.show()
self._minimized = False
self._anchor_to_right()
else:
# Reduire
self._bg_widget.hide()
self._mini_bar.show()
self._minimized = True
# Positionner la mini-barre en haut a droite
desktop = QApplication.desktop()
if desktop:
screen = desktop.availableGeometry(desktop.primaryScreen())
x = screen.right() - 90
y = screen.top() + 10
self.setGeometry(x, y, 80, 50)
# ---------------------------------------------------------------------------
# Drag (deplacer la fenetre sans barre de titre)
# ---------------------------------------------------------------------------
def mousePressEvent(self, event) -> None: # noqa: N802
if event.button() == Qt.LeftButton:
self._drag_pos = event.globalPos() - self.frameGeometry().topLeft()
event.accept()
def mouseMoveEvent(self, event) -> None: # noqa: N802
if event.buttons() == Qt.LeftButton and hasattr(self, '_drag_pos'):
self.move(event.globalPos() - self._drag_pos)
event.accept()
# ---------------------------------------------------------------------------
# Painting (fond arrondi semi-transparent)
# ---------------------------------------------------------------------------
def paintEvent(self, event) -> None: # noqa: N802
"""Peindre le fond semi-transparent avec coins arrondis."""
painter = QPainter(self)
painter.setRenderHint(QPainter.Antialiasing, True)
path = QPainterPath()
path.addRoundedRect(
0, 0, self.width(), self.height(),
styles.BORDER_RADIUS, styles.BORDER_RADIUS,
)
# Fond semi-transparent
bg = QColor(styles.COLOR_BG)
bg.setAlpha(245) # Legerement transparent
painter.fillPath(path, bg)
# Bordure
painter.setPen(QPen(QColor(styles.COLOR_BORDER), 1))
painter.drawPath(path)
painter.end()
# ---------------------------------------------------------------------------
# Lifecycle
# ---------------------------------------------------------------------------
def closeEvent(self, event) -> None: # noqa: N802
"""Ne pas fermer, juste cacher."""
event.ignore()
self.hide()
def shutdown(self) -> None:
"""Arret propre."""
self._conn_timer.stop()
self._overlay.hide_overlay()
self._client.shutdown()
logger.info("LeaMainWindow arretee")
Reference in New Issue View Git Blame Copy Permalink