import os from multiprocessing import Queue as MPQueue from PyQt5.QtCore import Qt, QObject, QRunnable, QThreadPool, pyqtSignal, pyqtSlot from PyQt5.QtGui import QCloseEvent from PyQt5.QtWidgets import QDialog, QFileDialog, QCheckBox, QMessageBox from custom_legendary.core import LegendaryCore from custom_legendary.utils.selective_dl import games from rare.ui.components.dialogs.install_dialog import Ui_InstallDialog from rare.utils.extra_widgets import PathEdit from rare.utils.models import InstallDownloadModel, InstallQueueItemModel from rare.utils.utils import get_size class InstallDialog(QDialog, Ui_InstallDialog): result_ready = pyqtSignal(InstallQueueItemModel) def __init__(self, core: LegendaryCore, dl_item: InstallQueueItemModel, update=False, silent=False, parent=None): super(InstallDialog, self).__init__(parent) self.setupUi(self) self.setAttribute(Qt.WA_DeleteOnClose, True) self.setWindowFlags(Qt.Dialog | Qt.CustomizeWindowHint | Qt.WindowTitleHint) self.core = core self.dl_item = dl_item self.dl_item.status_q = MPQueue() self.app_name = self.dl_item.options.app_name self.game = self.core.get_game(self.app_name) self.update = update self.silent = silent self.threadpool = QThreadPool(self) self.threadpool.setMaxThreadCount(1) header = self.tr("Update") if update else self.tr("Install") self.install_dialog_label.setText(f"

{header} \"{self.game.app_title}\"

") self.setWindowTitle(f"{self.windowTitle()} - {header} \"{self.game.app_title}\"") if self.core.lgd.config.has_option("Legendary", "install_dir"): default_path = self.core.lgd.config.get("Legendary", "install_dir") else: default_path = os.path.expanduser("~/legendary") if not default_path: default_path = os.path.expanduser("~/legendary") self.install_dir_edit = PathEdit(text=default_path, file_type=QFileDialog.DirectoryOnly, edit_func=self.option_changed) self.install_dir_layout.addWidget(self.install_dir_edit) if self.update: self.install_dir_label.setVisible(False) self.install_dir_edit.setVisible(False) if self.core.lgd.config.has_option("Legendary", "max_workers"): max_workers = self.core.lgd.config.get("Legendary", "max_workers") else: max_workers = 0 self.max_workers_spin.setValue(int(max_workers)) self.max_workers_spin.valueChanged.connect(self.option_changed) self.force_download_check.stateChanged.connect(self.option_changed) self.ignore_space_check.stateChanged.connect(self.option_changed) self.download_only_check.stateChanged.connect(self.option_changed) self.sdl_list_checks = list() try: for key, info in games[self.app_name].items(): cb = QDataCheckBox(info['name'], info['tags']) if key == '__required': self.dl_item.options.sdl_list.extend(info['tags']) cb.setChecked(True) cb.setDisabled(True) self.sdl_list_layout.addWidget(cb) self.sdl_list_checks.append(cb) self.sdl_list_frame.resize(self.sdl_list_frame.minimumSize()) for cb in self.sdl_list_checks: cb.stateChanged.connect(self.option_changed) except KeyError: self.sdl_list_frame.setVisible(False) self.sdl_list_label.setVisible(False) self.install_button.setEnabled(False) self.cancel_button.clicked.connect(self.cancel_clicked) self.verify_button.clicked.connect(self.verify_clicked) self.install_button.clicked.connect(self.install_clicked) self.options_changed = False self.worker_running = False self.reject_close = True self.resize(self.minimumSize()) self.setFixedSize(self.size()) if self.silent: self.reject_close = False self.get_download_info() else: self.show() def get_options(self): self.dl_item.options.base_path = self.install_dir_edit.text() if not self.update else None self.dl_item.options.max_workers = self.max_workers_spin.value() self.dl_item.options.force = self.force_download_check.isChecked() self.dl_item.options.ignore_space_req = self.ignore_space_check.isChecked() self.dl_item.options.no_install = self.download_only_check.isChecked() self.dl_item.options.sdl_list = [''] for cb in self.sdl_list_checks: if data := cb.isChecked(): self.dl_item.options.sdl_list.extend(data) def get_download_info(self): self.dl_item.download = None info_worker = InstallInfoWorker(self.core, self.dl_item) info_worker.setAutoDelete(True) info_worker.signals.result.connect(self.on_worker_result) info_worker.signals.failed.connect(self.on_worker_failed) info_worker.signals.finished.connect(self.on_worker_finished) self.worker_running = True self.threadpool.start(info_worker) def verify_clicked(self): message = self.tr("Updating...") self.download_size_info_label.setText(message) self.download_size_info_label.setStyleSheet("font-style: italic; font-weight: normal") self.install_size_info_label.setText(message) self.install_size_info_label.setStyleSheet("font-style: italic; font-weight: normal") self.cancel_button.setEnabled(False) self.verify_button.setEnabled(False) self.install_button.setEnabled(False) self.options_changed = False self.get_options() self.get_download_info() def option_changed(self): self.options_changed = True self.install_button.setEnabled(False) self.verify_button.setEnabled(not self.worker_running) def cancel_clicked(self): self.dl_item.download = None self.reject_close = False self.close() def install_clicked(self): self.reject_close = False self.close() def on_worker_result(self, dl_item: InstallDownloadModel): self.dl_item.download = dl_item # TODO: Check available size and act accordingly # TODO: (show message in label | color it | disable install unless ignore) # TODO: Find a way to get the installation size delta and show it download_size = self.dl_item.download.analysis.dl_size install_size = self.dl_item.download.analysis.install_size if download_size: self.download_size_info_label.setText("{}".format(get_size(download_size))) self.download_size_info_label.setStyleSheet("font-style: normal; font-weight: bold") self.install_button.setEnabled(not self.options_changed) else: self.install_size_info_label.setText(self.tr("Game already installed")) self.install_size_info_label.setStyleSheet("font-style: italics; font-weight: normal") self.install_size_info_label.setText("{}".format(get_size(install_size))) self.install_size_info_label.setStyleSheet("font-style: normal; font-weight: bold") self.verify_button.setEnabled(self.options_changed) self.cancel_button.setEnabled(True) if self.silent: self.close() def on_worker_failed(self, message: str): error_text = self.tr("Error") self.download_size_info_label.setText(error_text) self.install_size_info_label.setText(error_text) QMessageBox.critical(self, self.windowTitle(), message) self.verify_button.setEnabled(self.options_changed) self.cancel_button.setEnabled(True) if self.silent: self.close() def on_worker_finished(self): self.worker_running = False # lk: happens when close() is called, also when top right 'X' is pressed. # lk: reject any events not coming from the buttons in case the WM # lk: doesn't honor the window hints def closeEvent(self, a0: QCloseEvent) -> None: if self.reject_close: a0.ignore() else: self.threadpool.clear() self.threadpool.waitForDone() self.result_ready.emit(self.dl_item) a0.accept() class InstallInfoWorkerSignals(QObject): result = pyqtSignal(InstallDownloadModel) failed = pyqtSignal(str) finished = pyqtSignal() class InstallInfoWorker(QRunnable): def __init__(self, core: LegendaryCore, dl_item: InstallQueueItemModel): super(InstallInfoWorker, self).__init__() self.signals = InstallInfoWorkerSignals() self.core = core self.dl_item = dl_item @pyqtSlot() def run(self): try: download = InstallDownloadModel(*self.core.prepare_download( app_name=self.dl_item.options.app_name, base_path=self.dl_item.options.base_path, force=self.dl_item.options.force, no_install=self.dl_item.options.no_install, status_q=self.dl_item.status_q, # max_shm=, max_workers=self.dl_item.options.max_workers, # game_folder=, # disable_patching=, # override_manifest=, # override_old_manifest=, # override_base_url=, # platform_override=, # file_prefix_filter=, # file_exclude_filter=, # file_install_tag=, # dl_optimizations=, # dl_timeout=, repair=self.dl_item.options.repair, # repair_use_latest=, ignore_space_req=self.dl_item.options.ignore_space_req, # disable_delta=, # override_delta_manifest=, # reset_sdl=, sdl_prompt=lambda app_name, title: self.dl_item.options.sdl_list )) self.signals.result.emit(download) except RuntimeError as e: self.signals.failed.emit(str(e)) self.signals.finished.emit() class QDataCheckBox(QCheckBox): def __init__(self, text, data=None, parent=None): super(QDataCheckBox, self).__init__(parent) self.setText(text) self.data = data def isChecked(self): if super(QDataCheckBox, self).isChecked(): return self.data else: return False