281 lines
10 KiB
Python
Executable file
281 lines
10 KiB
Python
Executable file
from gui.v2.infrastructure.ThreadSafetyTool import ThreadSafetyTool
|
|
|
|
from core.observers.ConnectionObserver import ConnectionObserver
|
|
from core.observers.ClientObserver import ClientObserver
|
|
from core.services.networking.api_requests.ApiResponseModel import ApiResponse, ErrorType
|
|
from core.controllers.SyncController import new_sync
|
|
from core.models.Result import Result, ResultError
|
|
|
|
from PyQt6.QtWidgets import (QDialog, QVBoxLayout, QHBoxLayout, QTextEdit,
|
|
QLabel, QPushButton, QApplication)
|
|
from PyQt6.QtGui import QFont, QColor, QTextCursor
|
|
from PyQt6.QtCore import Qt, QTimer, QThread, pyqtSignal, pyqtSlot
|
|
from PyQt6.QtWidgets import QGraphicsDropShadowEffect
|
|
import sys
|
|
import traceback
|
|
from typing import Callable, Any
|
|
from PyQt6.QtGui import QTextCharFormat
|
|
|
|
import threading
|
|
from threading import Thread
|
|
|
|
|
|
class NotificationTerminalUI(QDialog):
|
|
def __init__(self, connection_observer: ConnectionObserver, client_observer: ClientObserver):
|
|
super().__init__()
|
|
self.mainthread = ThreadSafetyTool()
|
|
|
|
self.connection_observer = connection_observer
|
|
self.client_observer = client_observer
|
|
self.synchronized = False
|
|
self.worker_thread = None
|
|
self.task_success = False
|
|
|
|
self.connection_observer.subscribe("message",
|
|
lambda msg: self.add_notification("UPDATE", msg, "#00ff88"))
|
|
|
|
self.connection_observer.subscribe("connecting",
|
|
lambda msg: self.add_notification("CONNECT", msg, "#00ff88"))
|
|
|
|
self.connection_observer.subscribe("tor_bootstrapping",
|
|
lambda msg: self.add_notification("BOOTSTRAP", msg, "#00d4ff"))
|
|
|
|
self.connection_observer.subscribe("tor_bootstrap_progressing",
|
|
lambda msg: self.add_notification("PROGRESS", msg, "#ff0088"))
|
|
|
|
self.connection_observer.subscribe("tor_bootstrapped",
|
|
lambda msg: self.add_notification("SUCCESS", msg, "#00ff88"))
|
|
|
|
self.connection_observer.subscribe("custom_message",
|
|
lambda msg: self.add_notification("INFO", msg, "#ffff00"))
|
|
|
|
# Subscribe to client_observer events
|
|
self.client_observer.subscribe("synchronizing",
|
|
lambda event: self.add_notification("SYNC",
|
|
event.subject if event.subject else "Sync in progress...", "#00d4ff"))
|
|
|
|
self.client_observer.subscribe("synchronized",
|
|
lambda event: self.on_synchronized())
|
|
|
|
self.client_observer.subscribe("updating",
|
|
lambda event: self.add_notification("UPDATE", "Updating client...", "#ff0088"))
|
|
|
|
self.client_observer.subscribe("update_progressing",
|
|
lambda event: self.add_notification("PROGRESS",
|
|
f'Current progress: {event.meta.get("progress", 0):.2f}%', "#ffff00"))
|
|
|
|
self.client_observer.subscribe("updated",
|
|
lambda event: self.add_notification("COMPLETE",
|
|
"Restart client to apply update.", "#00ff88"))
|
|
|
|
self.client_observer.subscribe("custom_message",
|
|
lambda event: self.add_notification("ERROR",
|
|
event.subject if event.subject else "Error, check logs", "#ff0088"))
|
|
|
|
self.init_ui()
|
|
|
|
def init_ui(self):
|
|
self.setWindowTitle("SYNC PROCESS")
|
|
self.setGeometry(100, 100, 700, 600)
|
|
self.setModal(True)
|
|
|
|
layout = QVBoxLayout(self)
|
|
layout.setSpacing(15)
|
|
layout.setContentsMargins(30, 30, 30, 30)
|
|
|
|
self.setStyleSheet("""
|
|
QDialog {
|
|
background-color: #0a0e27;
|
|
color: #00ff88;
|
|
}
|
|
""")
|
|
|
|
# Title Label
|
|
title_label = QLabel("▓░ NOTIFICATION TERMINAL ░▓")
|
|
title_label.setFont(QFont("Courier New", 14, QFont.Weight.Bold))
|
|
title_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
|
title_label.setStyleSheet("""
|
|
QLabel {
|
|
color: #00ff88;
|
|
background-color: transparent;
|
|
padding: 15px;
|
|
border: 2px solid #00ff88;
|
|
border-radius: 5px;
|
|
letter-spacing: 2px;
|
|
}
|
|
""")
|
|
title_shadow = QGraphicsDropShadowEffect()
|
|
title_shadow.setBlurRadius(15)
|
|
title_shadow.setColor(QColor(0, 255, 136, 200))
|
|
title_shadow.setOffset(0, 0)
|
|
title_label.setGraphicsEffect(title_shadow)
|
|
layout.addWidget(title_label)
|
|
|
|
# Scrollable text display
|
|
self.output_text = QTextEdit()
|
|
self.output_text.setReadOnly(True)
|
|
self.output_text.setFont(QFont("Courier New", 10))
|
|
self.output_text.setMinimumHeight(400)
|
|
self.output_text.setStyleSheet("""
|
|
QTextEdit {
|
|
background-color: #1a1f3a;
|
|
color: #00ff88;
|
|
border: 2px solid #00ff88;
|
|
border-radius: 5px;
|
|
padding: 15px;
|
|
font-size: 16pt;
|
|
font-family: 'Courier New', monospace;
|
|
selection-background-color: #ff0088;
|
|
}
|
|
QScrollBar:vertical {
|
|
background-color: #0a0e27;
|
|
width: 12px;
|
|
border-radius: 6px;
|
|
}
|
|
QScrollBar::handle:vertical {
|
|
background-color: #00ff88;
|
|
border-radius: 6px;
|
|
min-height: 20px;
|
|
}
|
|
QScrollBar::handle:vertical:hover {
|
|
background-color: #00d4ff;
|
|
}
|
|
""")
|
|
layout.addWidget(self.output_text)
|
|
|
|
# Status label
|
|
self.status_label = QLabel("◄ WAITING FOR TASK START ►")
|
|
self.status_label.setFont(QFont("Courier New", 9))
|
|
self.status_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
|
self.status_label.setStyleSheet("""
|
|
QLabel {
|
|
color: #ff0088;
|
|
background-color: transparent;
|
|
padding: 8px;
|
|
letter-spacing: 1px;
|
|
}
|
|
""")
|
|
layout.addWidget(self.status_label)
|
|
|
|
# Button layout
|
|
button_layout = QHBoxLayout()
|
|
clear_btn = QPushButton("CLEAR")
|
|
clear_btn.setFont(QFont("Courier New", 10, QFont.Weight.Bold))
|
|
clear_btn.setMinimumHeight(40)
|
|
clear_btn.setCursor(Qt.CursorShape.PointingHandCursor)
|
|
clear_btn.setStyleSheet("""
|
|
QPushButton {
|
|
background-color: #1a1f3a;
|
|
color: #00d4ff;
|
|
border: 2px solid #00d4ff;
|
|
border-radius: 5px;
|
|
padding: 8px 20px;
|
|
font-weight: bold;
|
|
}
|
|
QPushButton:hover {
|
|
background-color: #00d4ff;
|
|
color: #0a0e27;
|
|
}
|
|
QPushButton:pressed {
|
|
background-color: #0a0e27;
|
|
border: 3px solid #00d4ff;
|
|
}
|
|
""")
|
|
clear_btn.clicked.connect(self.clear_output)
|
|
shadow = QGraphicsDropShadowEffect()
|
|
shadow.setBlurRadius(15)
|
|
shadow.setColor(QColor(0, 212, 255))
|
|
shadow.setOffset(0, 0)
|
|
clear_btn.setGraphicsEffect(shadow)
|
|
button_layout.addWidget(clear_btn)
|
|
button_layout.addStretch()
|
|
layout.addLayout(button_layout)
|
|
|
|
def add_notification(self, prefix: str, message: str, color: str = "#00ff88"):
|
|
"""Queue to main thread safely."""
|
|
self.mainthread(lambda: self._add_notification_impl(prefix, message, color))()
|
|
|
|
def _add_notification_impl(self, prefix: str, message: str, color: str):
|
|
"""Actual implementation on main thread."""
|
|
cursor = self.output_text.textCursor()
|
|
cursor.movePosition(cursor.MoveOperation.End)
|
|
fmt = QTextCharFormat()
|
|
fmt.setForeground(QColor(color))
|
|
cursor.insertText(f"[{prefix}] {message}\n", fmt)
|
|
self.output_text.setTextCursor(cursor)
|
|
self.output_text.ensureCursorVisible()
|
|
|
|
def set_status(self, text: str):
|
|
"""This runs on the main thread, safely."""
|
|
self.status_label.setText(text)
|
|
|
|
def on_synchronized(self):
|
|
"""Handle synchronized notification and exit."""
|
|
self.add_notification("SYNC", "Sync complete ✓", "#00ff88")
|
|
self.set_status("◄ SYNCHRONIZED - CLOSING ►")
|
|
self.synchronized = True
|
|
self.task_success = True
|
|
QTimer.singleShot(500, self.accept)
|
|
|
|
def run_task(self, task_func: Callable):
|
|
"""
|
|
Purpose:
|
|
Insert any function for a pure python background thread.
|
|
Which then pushes to the main UI via ThreadSafety
|
|
"""
|
|
if self.worker_thread is not None and self.worker_thread.is_alive():
|
|
self.add_notification("SYSTEM", "Task already running!", "#ff0088")
|
|
return
|
|
|
|
self.set_status("◄ TASK RUNNING ►")
|
|
self.add_notification("SYSTEM", "Background task started", "#00d4ff")
|
|
|
|
# Start in pure Python thread (daemon=True so it doesn't block exit)
|
|
self.worker_thread = Thread(target=self._run_task_wrapper, args=(task_func,), daemon=True)
|
|
self.worker_thread.start()
|
|
|
|
def _run_task_wrapper(self, task_func: Callable):
|
|
"""Wrapper that runs task and handles errors."""
|
|
try:
|
|
task_func()
|
|
self.task_success = True
|
|
except Exception as e:
|
|
self.add_notification("ERROR", f"Task failed: {str(e)}", "#ff0088")
|
|
self.set_status("◄ TASK FAILED ►")
|
|
self.task_success = False
|
|
|
|
def clear_output(self):
|
|
"""Clear the terminal output."""
|
|
self.output_text.clear()
|
|
self.add_notification("SYSTEM", "Terminal cleared", "#00d4ff")
|
|
|
|
def get_result(self) -> bool:
|
|
"""Return True if synchronized, False otherwise."""
|
|
return self.synchronized and self.task_success
|
|
|
|
def closeEvent(self, event):
|
|
"""Ensure thread is properly cleaned up on close."""
|
|
if self.worker_thread is not None and self.worker_thread.is_alive():
|
|
self.worker_thread.join(timeout=1)
|
|
event.accept()
|
|
|
|
|
|
def show_terminal(
|
|
do_this_function: Callable[[Any, Any], None],
|
|
connection_observer: ConnectionObserver,
|
|
client_observer: ClientObserver
|
|
) -> bool:
|
|
|
|
app = QApplication.instance() or QApplication(sys.argv)
|
|
|
|
# Create the UI
|
|
dialog = NotificationTerminalUI(connection_observer=connection_observer, client_observer=client_observer)
|
|
|
|
# Start the background task
|
|
dialog.run_task(do_this_function)
|
|
|
|
# Show dialog and wait
|
|
dialog.exec()
|
|
|
|
# Return success
|
|
return dialog.get_result()
|