mirror of
https://github.com/spalencsar/bearhub.git
synced 2026-07-07 01:14:15 +02:00
306 lines
11 KiB
Python
306 lines
11 KiB
Python
import datetime
|
|
import operator
|
|
import time
|
|
from functools import reduce
|
|
from typing import Tuple
|
|
|
|
from PyQt5.QtCore import QSize, Qt, QThread, pyqtSignal, QCoreApplication
|
|
from PyQt5.QtGui import QIcon
|
|
from PyQt5.QtWidgets import QWidget, QVBoxLayout, QLabel, QSizePolicy, QTableWidget, QHeaderView, QPushButton, QToolBar, \
|
|
QProgressBar, QApplication
|
|
|
|
from bauh.api.abstract.context import ApplicationContext
|
|
from bauh.api.abstract.controller import SoftwareManager
|
|
from bauh.api.abstract.handler import TaskManager
|
|
from bauh.view.qt import root
|
|
from bauh.view.qt.components import new_spacer
|
|
from bauh.view.qt.qt_utils import centralize
|
|
from bauh.view.qt.thread import AnimateProgress
|
|
from bauh.view.util.translation import I18n
|
|
|
|
|
|
class Prepare(QThread, TaskManager):
|
|
signal_register = pyqtSignal(str, str, object)
|
|
signal_update = pyqtSignal(str, float, str)
|
|
signal_finished = pyqtSignal(str)
|
|
signal_started = pyqtSignal()
|
|
signal_ask_password = pyqtSignal()
|
|
|
|
def __init__(self, context: ApplicationContext, manager: SoftwareManager, i18n: I18n):
|
|
super(Prepare, self).__init__()
|
|
self.manager = manager
|
|
self.i18n = i18n
|
|
self.context = context
|
|
self.waiting_password = False
|
|
self.password_response = None
|
|
|
|
def ask_password(self) -> Tuple[str, bool]:
|
|
self.waiting_password = True
|
|
self.signal_ask_password.emit()
|
|
|
|
while self.waiting_password:
|
|
time.sleep(0.1) # waiting for user input
|
|
|
|
return self.password_response
|
|
|
|
def set_password_reply(self, password: str, valid: bool):
|
|
self.password_response = password, valid
|
|
self.waiting_password = False
|
|
|
|
def run(self):
|
|
root_pwd = None
|
|
if self.manager.requires_root('prepare', None):
|
|
root_pwd, ok = self.ask_password()
|
|
|
|
if not ok:
|
|
QCoreApplication.exit(1)
|
|
|
|
self.signal_started.emit()
|
|
self.manager.prepare(self, root_pwd, None)
|
|
|
|
def update_progress(self, task_id: str, progress: float, substatus: str):
|
|
self.signal_update.emit(task_id, progress, substatus)
|
|
|
|
def register_task(self, id_: str, label: str, icon_path: str):
|
|
self.signal_register.emit(id_, label, icon_path)
|
|
|
|
def finish_task(self, task_id: str):
|
|
self.signal_finished.emit(task_id)
|
|
|
|
|
|
class CheckFinished(QThread):
|
|
signal_finished = pyqtSignal()
|
|
|
|
def __init__(self):
|
|
super(CheckFinished, self).__init__()
|
|
self.total = None
|
|
self.finished = None
|
|
|
|
def run(self):
|
|
while True:
|
|
if self.total is not None and self.finished is not None:
|
|
if self.total == self.finished:
|
|
break
|
|
|
|
time.sleep(0.01)
|
|
|
|
self.signal_finished.emit()
|
|
|
|
def update(self, total: int, finished: int):
|
|
if total is not None:
|
|
self.total = total
|
|
|
|
if finished is not None:
|
|
self.finished = finished
|
|
|
|
|
|
class EnableSkip(QThread):
|
|
|
|
signal_timeout = pyqtSignal()
|
|
|
|
def run(self):
|
|
ti = datetime.datetime.now()
|
|
|
|
while True:
|
|
if datetime.datetime.now() >= ti + datetime.timedelta(minutes=1.5):
|
|
self.signal_timeout.emit()
|
|
break
|
|
|
|
time.sleep(0.1)
|
|
|
|
|
|
class PreparePanel(QWidget, TaskManager):
|
|
|
|
signal_status = pyqtSignal(object, object)
|
|
signal_password_response = pyqtSignal(str, bool)
|
|
|
|
def __init__(self, context: ApplicationContext, manager: SoftwareManager, screen_size: QSize, i18n: I18n, manage_window: QWidget):
|
|
super(PreparePanel, self).__init__()
|
|
self.i18n = i18n
|
|
self.context = context
|
|
self.manage_window = manage_window
|
|
self.setWindowTitle(' ')
|
|
self.setMinimumWidth(screen_size.width() * 0.5)
|
|
self.setMinimumHeight(screen_size.height() * 0.35)
|
|
self.setMaximumHeight(screen_size.height() * 0.95)
|
|
self.setLayout(QVBoxLayout())
|
|
self.setSizePolicy(QSizePolicy.Minimum, QSizePolicy.Preferred)
|
|
self.manager = manager
|
|
self.tasks = {}
|
|
self.ntasks = 0
|
|
self.ftasks = 0
|
|
self.self_close = False
|
|
|
|
self.prepare_thread = Prepare(self.context, manager, self.i18n)
|
|
self.prepare_thread.signal_register.connect(self.register_task)
|
|
self.prepare_thread.signal_update.connect(self.update_progress)
|
|
self.prepare_thread.signal_finished.connect(self.finish_task)
|
|
self.prepare_thread.signal_started.connect(self.start)
|
|
self.prepare_thread.signal_ask_password.connect(self.ask_root_password)
|
|
self.signal_password_response.connect(self.prepare_thread.set_password_reply)
|
|
|
|
self.check_thread = CheckFinished()
|
|
self.signal_status.connect(self.check_thread.update)
|
|
self.check_thread.signal_finished.connect(self.finish)
|
|
|
|
self.skip_thread = EnableSkip()
|
|
self.skip_thread.signal_timeout.connect(self._enable_skip_button)
|
|
|
|
self.progress_thread = AnimateProgress()
|
|
self.progress_thread.signal_change.connect(self._change_progress)
|
|
|
|
self.label_top = QLabel()
|
|
self.label_top.setText("{}...".format(self.i18n['prepare_panel.title.start'].capitalize()))
|
|
self.label_top.setAlignment(Qt.AlignHCenter)
|
|
self.label_top.setStyleSheet("QLabel { font-size: 14px; font-weight: bold; }")
|
|
self.layout().addWidget(self.label_top)
|
|
self.layout().addWidget(QLabel())
|
|
|
|
self.table = QTableWidget()
|
|
self.table.setStyleSheet("QTableWidget { background-color: transparent; }")
|
|
self.table.setFocusPolicy(Qt.NoFocus)
|
|
self.table.setShowGrid(False)
|
|
self.table.verticalHeader().setVisible(False)
|
|
self.table.horizontalHeader().setVisible(False)
|
|
self.table.horizontalHeader().setSizePolicy(QSizePolicy.MinimumExpanding, QSizePolicy.Preferred)
|
|
self.table.setColumnCount(4)
|
|
self.table.setHorizontalHeaderLabels(['' for _ in range(4)])
|
|
self.layout().addWidget(self.table)
|
|
|
|
toolbar = QToolBar()
|
|
self.bt_close = QPushButton(self.i18n['close'].capitalize())
|
|
self.bt_close.clicked.connect(self.close)
|
|
self.bt_close.setVisible(False)
|
|
self.ref_bt_close = toolbar.addWidget(self.bt_close)
|
|
|
|
toolbar.addWidget(new_spacer())
|
|
self.progress_bar = QProgressBar()
|
|
self.progress_bar.setMaximumHeight(10 if QApplication.instance().style().objectName().lower() == 'windows' else 4)
|
|
self.progress_bar.setTextVisible(False)
|
|
self.progress_bar.setVisible(False)
|
|
self.ref_progress_bar = toolbar.addWidget(self.progress_bar)
|
|
toolbar.addWidget(new_spacer())
|
|
|
|
self.bt_skip = QPushButton(self.i18n['prepare_panel.bt_skip.label'].capitalize())
|
|
self.bt_skip.clicked.connect(self.finish)
|
|
self.bt_skip.setEnabled(False)
|
|
toolbar.addWidget(self.bt_skip)
|
|
|
|
self.layout().addWidget(toolbar)
|
|
|
|
def ask_root_password(self):
|
|
root_pwd, ok = root.ask_root_password(self.context, self.i18n)
|
|
self.signal_password_response.emit(root_pwd, ok)
|
|
|
|
def _enable_skip_button(self):
|
|
self.bt_skip.setEnabled(True)
|
|
|
|
def _change_progress(self, value: int):
|
|
self.progress_bar.setValue(value)
|
|
|
|
def get_table_width(self) -> int:
|
|
return reduce(operator.add, [self.table.columnWidth(i) for i in range(self.table.columnCount())])
|
|
|
|
def _resize_columns(self):
|
|
header_horizontal = self.table.horizontalHeader()
|
|
for i in range(self.table.columnCount()):
|
|
header_horizontal.setSectionResizeMode(i, QHeaderView.ResizeToContents)
|
|
|
|
self.resize(self.get_table_width() * 1.05, self.sizeHint().height())
|
|
|
|
def show(self):
|
|
super(PreparePanel, self).show()
|
|
self.prepare_thread.start()
|
|
centralize(self)
|
|
|
|
def start(self):
|
|
self.ref_bt_close.setVisible(True)
|
|
self.check_thread.start()
|
|
self.skip_thread.start()
|
|
|
|
self.ref_progress_bar.setVisible(True)
|
|
self.progress_thread.start()
|
|
|
|
def closeEvent(self, QCloseEvent):
|
|
if not self.self_close:
|
|
QCoreApplication.exit()
|
|
|
|
def register_task(self, id_: str, label: str, icon_path: str):
|
|
self.ntasks += 1
|
|
self.table.setRowCount(self.ntasks)
|
|
|
|
task_row = self.ntasks - 1
|
|
|
|
lb_icon = QLabel()
|
|
lb_icon.setContentsMargins(10, 0, 10, 0)
|
|
lb_icon.setSizePolicy(QSizePolicy.Minimum, QSizePolicy.Preferred)
|
|
|
|
if icon_path:
|
|
lb_icon.setPixmap(QIcon(icon_path).pixmap(14, 14))
|
|
|
|
self.table.setCellWidget(task_row, 0, lb_icon)
|
|
|
|
lb_status = QLabel(label)
|
|
lb_status.setMinimumWidth(50)
|
|
lb_status.setSizePolicy(QSizePolicy.Minimum, QSizePolicy.Preferred)
|
|
lb_status.setStyleSheet("QLabel { color: blue; font-weight: bold; }")
|
|
self.table.setCellWidget(task_row, 1, lb_status)
|
|
|
|
lb_sub = QLabel()
|
|
lb_sub.setContentsMargins(10, 0, 10, 0)
|
|
lb_sub.setSizePolicy(QSizePolicy.Minimum, QSizePolicy.Preferred)
|
|
lb_sub.setMinimumWidth(50)
|
|
self.table.setCellWidget(task_row, 2, lb_sub)
|
|
|
|
lb_progress = QLabel('{0:.2f}'.format(0) + '%')
|
|
lb_progress.setContentsMargins(10, 0, 10, 0)
|
|
lb_progress.setStyleSheet("QLabel { color: blue; font-weight: bold; }")
|
|
lb_progress.setSizePolicy(QSizePolicy.Minimum, QSizePolicy.Preferred)
|
|
|
|
self.table.setCellWidget(task_row, 3, lb_progress)
|
|
|
|
self.tasks[id_] = {'lb_status': lb_status,
|
|
'lb_prog': lb_progress,
|
|
'progress': 0,
|
|
'lb_sub': lb_sub,
|
|
'finished': False}
|
|
|
|
self.signal_status.emit(self.ntasks, self.ftasks)
|
|
|
|
def update_progress(self, task_id: str, progress: float, substatus: str):
|
|
task = self.tasks[task_id]
|
|
|
|
if progress != task['progress']:
|
|
task['progress'] = progress
|
|
task['lb_prog'].setText('{0:.2f}'.format(progress) + '%')
|
|
|
|
if substatus:
|
|
task['lb_sub'].setText('( {} )'.format(substatus))
|
|
else:
|
|
task['lb_sub'].setText('')
|
|
|
|
self._resize_columns()
|
|
|
|
def finish_task(self, task_id: str):
|
|
task = self.tasks[task_id]
|
|
task['lb_sub'].setText('')
|
|
|
|
for key in ('lb_prog', 'lb_status'):
|
|
task[key].setStyleSheet('QLabel { color: green; text-decoration: line-through; }')
|
|
|
|
task['finished'] = True
|
|
self._resize_columns()
|
|
|
|
self.ftasks += 1
|
|
self.signal_status.emit(self.ntasks, self.ftasks)
|
|
|
|
if self.ntasks == self.ftasks:
|
|
self.label_top.setText("... {} ...".format(self.i18n['ready'].capitalize()))
|
|
|
|
def finish(self):
|
|
if self.isVisible():
|
|
self.manage_window.refresh_packages()
|
|
self.manage_window.show()
|
|
self.self_close = True
|
|
self.close()
|