''' app/ui/main_window.py ''' import ast import datetime import io import json import logging import os import sys import threading import time import PIL from PIL import Image from PyQt6.QtCore import Qt, QByteArray, QBuffer, QIODevice, QThread from PyQt6.QtGui import QPixmap, QImage, QFont, QIcon from PyQt6.QtWidgets import QMainWindow, QWidget, QHBoxLayout, QListWidget, QTableWidget, QAbstractItemView, \ QTableWidgetItem, QLabel, QVBoxLayout, QHeaderView, QMessageBox, QGroupBox, QPushButton, QListWidgetItem, \ QFileDialog from src.api.api_server import API_VERSION from src.render_queue import RenderQueue from src.utilities.misc_helper import get_time_elapsed, resources_dir, is_localhost from src.utilities.status_utils import RenderStatus from src.utilities.zeroconf_server import ZeroconfServer from src.ui.add_job_window import NewRenderJobForm from src.ui.console_window import ConsoleWindow from src.ui.engine_browser import EngineBrowserWindow from src.ui.log_window import LogViewer from src.ui.widgets.menubar import MenuBar from src.ui.widgets.proportional_image_label import ProportionalImageLabel from src.ui.widgets.statusbar import StatusBar from src.ui.widgets.toolbar import ToolBar from src.api.serverproxy_manager import ServerProxyManager from src.utilities.misc_helper import launch_url, iso_datestring_to_formatted_datestring from src.version import APP_NAME logger = logging.getLogger() class MainWindow(QMainWindow): """ MainWindow Args: QMainWindow (QMainWindow): Inheritance """ def __init__(self) -> None: """ Initialize the Main-Window. """ super().__init__() # Load the queue self.job_list_view = None self.server_info_ram = None self.server_info_cpu = None self.server_info_os = None self.server_info_gpu = None self.server_info_hostname = None self.engine_browser_window = None self.server_info_group = None self.current_hostname = None self.subprocess_runner = None # To pass to console self.buffer_handler = None # Window-Settings self.setWindowTitle(APP_NAME) self.setGeometry(100, 100, 900, 800) central_widget = QWidget(self) self.setCentralWidget(central_widget) main_layout = QHBoxLayout(central_widget) # Create a QLabel widget to display the image self.image_label = ProportionalImageLabel() self.image_label.setMaximumSize(700, 500) self.image_label.setFixedHeight(300) self.image_label.setAlignment(Qt.AlignmentFlag.AlignTop | Qt.AlignmentFlag.AlignHCenter) self.load_image_path(os.path.join(resources_dir(), 'Rectangle.png')) # Server list self.server_list_view = QListWidget() self.server_list_view.itemClicked.connect(self.server_picked) list_font = QFont() list_font.setPointSize(16) self.server_list_view.setFont(list_font) self.added_hostnames = [] self.setup_ui(main_layout) # Add Widgets to Window # self.custom_menu_bar = self.setMenuBar(MenuBar(self)) self.setStatusBar(StatusBar(self)) self.create_toolbars() # start background update self.bg_update_thread = QThread() self.bg_update_thread.run = self.__background_update self.bg_update_thread.start() # Setup other windows self.new_job_window = None self.console_window = None self.log_viewer_window = None # Pick default job self.job_picked() def setup_ui(self, main_layout): # Servers server_list_group = QGroupBox("Available Servers") list_layout = QVBoxLayout() list_layout.addWidget(self.server_list_view) list_layout.setContentsMargins(0, 0, 0, 0) server_list_group.setLayout(list_layout) server_info_group = QGroupBox("Server Info") # Server Info Group self.server_info_hostname = QLabel() self.server_info_os = QLabel() self.server_info_cpu = QLabel() self.server_info_ram = QLabel() self.server_info_gpu = QLabel() server_info_engines_button = QPushButton("Render Engines") server_info_engines_button.clicked.connect(self.engine_browser) server_info_layout = QVBoxLayout() server_info_layout.addWidget(self.server_info_hostname) server_info_layout.addWidget(self.server_info_os) server_info_layout.addWidget(self.server_info_cpu) server_info_layout.addWidget(self.server_info_ram) server_info_layout.addWidget(self.server_info_gpu) server_info_layout.addWidget(server_info_engines_button) server_info_group.setLayout(server_info_layout) # Server Button Layout server_button_layout = QHBoxLayout() add_server_button = QPushButton(text="+") remove_server_button = QPushButton(text="-") server_button_layout.addWidget(add_server_button) server_button_layout.addWidget(remove_server_button) # Layouts info_layout = QVBoxLayout() info_layout.addWidget(server_list_group, stretch=True) info_layout.addWidget(server_info_group) info_layout.setContentsMargins(0, 0, 0, 0) server_list_group.setFixedWidth(260) self.server_picked() # Job list self.job_list_view = QTableWidget() self.job_list_view.setRowCount(0) self.job_list_view.setColumnCount(8) self.job_list_view.setSelectionBehavior(QAbstractItemView.SelectionBehavior.SelectRows) self.job_list_view.verticalHeader().setVisible(False) self.job_list_view.itemSelectionChanged.connect(self.job_picked) self.job_list_view.setEditTriggers(QAbstractItemView.EditTrigger.NoEditTriggers) self.refresh_job_headers() # Image Layout image_group = QGroupBox("Job Preview") image_layout = QVBoxLayout(image_group) image_layout.setContentsMargins(0, 0, 0, 0) image_center_layout = QHBoxLayout() image_center_layout.addWidget(self.image_label) image_layout.addWidget(self.image_label) # image_layout.addLayout(image_center_layout) # Job Layout job_list_group = QGroupBox("Render Jobs") job_list_layout = QVBoxLayout(job_list_group) job_list_layout.setContentsMargins(0, 0, 0, 0) image_layout.addWidget(self.job_list_view, stretch=True) image_layout.addLayout(job_list_layout) # Add them all to the window main_layout.addLayout(info_layout) right_layout = QVBoxLayout() right_layout.setContentsMargins(0, 0, 0, 0) right_layout.addWidget(image_group) # right_layout.addWidget(job_list_group) main_layout.addLayout(right_layout) def __background_update(self): while True: try: self.update_servers() self.fetch_jobs() except RuntimeError: pass except Exception as e: logger.error(f"Uncaught exception in background update: {e}") time.sleep(0.5) def closeEvent(self, event): running_jobs = len(RenderQueue.running_jobs()) if running_jobs: reply = QMessageBox.question(self, "Running Jobs", f"You have {running_jobs} jobs running.\n" f"Quitting will cancel these renders. Continue?", QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.Cancel, QMessageBox.StandardButton.Cancel) if reply == QMessageBox.StandardButton.Yes: event.accept() else: event.ignore() # -- Server Code -- # @property def current_server_proxy(self): return ServerProxyManager.get_proxy_for_hostname(self.current_hostname) def server_picked(self): """Update the UI elements relevant to the server selection.""" try: # Retrieve the new hostname selected by the user new_hostname = self.server_list_view.currentItem().text() # Check if the hostname has changed to avoid unnecessary updates if new_hostname != self.current_hostname: # Update the current hostname and clear the job list self.current_hostname = new_hostname self.job_list_view.setRowCount(0) self.fetch_jobs(clear_table=True) # Select the first row if there are jobs listed if self.job_list_view.rowCount(): self.job_list_view.selectRow(0) # Update server information display self.update_server_info_display(new_hostname) except AttributeError: # Handle cases where the server list view might not be properly initialized pass def update_server_info_display(self, hostname): """Updates the server information section of the UI.""" self.server_info_hostname.setText(f"Name: {hostname}") server_info = ZeroconfServer.get_hostname_properties(hostname) # Use the get method with defaults to avoid KeyError os_info = f"OS: {server_info.get('system_os', 'Unknown')} {server_info.get('system_os_version', '')}" cleaned_cpu_name = server_info.get('system_cpu_brand', 'Unknown').replace(' CPU','').replace('(TM)','').replace('(R)', '') cpu_info = f"CPU: {cleaned_cpu_name} ({server_info.get('system_cpu_cores', 'Unknown')} cores)" memory_info = f"RAM: {server_info.get('system_memory', 'Unknown')} GB" # Get and format GPU info try: gpu_list = ast.literal_eval(server_info.get('gpu_info', [])) # Format all GPUs gpu_info_parts = [] for gpu in gpu_list: gpu_name = gpu.get('name', 'Unknown').replace('(TM)','').replace('(R)', '') gpu_memory = gpu.get('memory', 'Unknown') # Add " GB" suffix if memory is a number if isinstance(gpu_memory, (int, float)) or (isinstance(gpu_memory, str) and gpu_memory.isdigit()): gpu_memory_str = f"{gpu_memory} GB" else: gpu_memory_str = str(gpu_memory) gpu_info_parts.append(f"{gpu_name} ({gpu_memory_str})") gpu_info = f"GPU: {', '.join(gpu_info_parts)}" if gpu_info_parts else "GPU: Unknown" except Exception as e: logger.error(f"Error parsing GPU info: {e}") gpu_info = "GPU: Unknown" self.server_info_os.setText(os_info.strip()) self.server_info_cpu.setText(cpu_info) self.server_info_ram.setText(memory_info) self.server_info_gpu.setText(gpu_info) def fetch_jobs(self, clear_table=False): if not self.current_server_proxy: return if clear_table: self.job_list_view.clear() self.refresh_job_headers() job_fetch = self.current_server_proxy.get_all_jobs(ignore_token=False) if job_fetch: num_jobs = len(job_fetch) self.job_list_view.setRowCount(num_jobs) for row, job in enumerate(job_fetch): display_status = job['status'] if job['status'] != RenderStatus.RUNNING.value else \ ('%.0f%%' % (job['percent_complete'] * 100)) # if running, show percent, otherwise just show status tags = (job['status'],) start_time = datetime.datetime.fromisoformat(job['start_time']) if job['start_time'] else None end_time = datetime.datetime.fromisoformat(job['end_time']) if job['end_time'] else None time_elapsed = "" if (job['status'] != RenderStatus.RUNNING.value and not end_time) else \ get_time_elapsed(start_time, end_time) name = job.get('name') or os.path.basename(job.get('input_path', '')) engine_name = f"{job.get('engine', '')}-{job.get('engine_version')}" priority = str(job.get('priority', '')) total_frames = str(job.get('total_frames', '')) date_created_string = iso_datestring_to_formatted_datestring(job['date_created']) items = [QTableWidgetItem(job['id']), QTableWidgetItem(name), QTableWidgetItem(engine_name), QTableWidgetItem(priority), QTableWidgetItem(display_status), QTableWidgetItem(time_elapsed), QTableWidgetItem(total_frames), QTableWidgetItem(date_created_string)] for col, item in enumerate(items): self.job_list_view.setItem(row, col, item) # -- Job Code -- # def job_picked(self): def fetch_preview(job_id): try: default_image_path = "error.png" before_fetch_hostname = self.current_server_proxy.hostname response = self.current_server_proxy.request(f'job/{job_id}/thumbnail?size=big') if response.ok: try: with io.BytesIO(response.content) as image_data_stream: image = Image.open(image_data_stream) if self.current_server_proxy.hostname == before_fetch_hostname and job_id == \ self.selected_job_ids()[0]: self.load_image_data(image) return except PIL.UnidentifiedImageError: default_image_path = response.text else: default_image_path = default_image_path or response.text self.load_image_path(os.path.join(resources_dir(), default_image_path)) except ConnectionError as e: logger.error(f"Connection error fetching image: {e}") except Exception as e: logger.error(f"Error fetching image: {e}") job_id = self.selected_job_ids()[0] if self.selected_job_ids() else None local_server = is_localhost(self.current_hostname) if job_id: fetch_thread = threading.Thread(target=fetch_preview, args=(job_id,)) fetch_thread.daemon = True fetch_thread.start() selected_row = self.job_list_view.selectionModel().selectedRows()[0] current_status = self.job_list_view.item(selected_row.row(), 4).text() # show / hide the stop button show_stop_button = "%" in current_status self.topbar.actions_call['Stop Job'].setEnabled(show_stop_button) self.topbar.actions_call['Stop Job'].setVisible(show_stop_button) self.topbar.actions_call['Delete Job'].setEnabled(not show_stop_button) self.topbar.actions_call['Delete Job'].setVisible(not show_stop_button) self.topbar.actions_call['Render Log'].setEnabled(True) self.topbar.actions_call['Download'].setEnabled(not local_server) self.topbar.actions_call['Download'].setVisible(not local_server) self.topbar.actions_call['Open Files'].setEnabled(local_server) self.topbar.actions_call['Open Files'].setVisible(local_server) else: # load default default_image_path = os.path.join(resources_dir(), 'Rectangle.png') self.load_image_path(default_image_path) self.topbar.actions_call['Stop Job'].setVisible(False) self.topbar.actions_call['Stop Job'].setEnabled(False) self.topbar.actions_call['Delete Job'].setEnabled(False) self.topbar.actions_call['Render Log'].setEnabled(False) self.topbar.actions_call['Download'].setEnabled(False) self.topbar.actions_call['Download'].setVisible(True) self.topbar.actions_call['Open Files'].setEnabled(False) self.topbar.actions_call['Open Files'].setVisible(False) def selected_job_ids(self): try: selected_rows = self.job_list_view.selectionModel().selectedRows() job_ids = [] for selected_row in selected_rows: id_item = self.job_list_view.item(selected_row.row(), 0) job_ids.append(id_item.text()) return job_ids except AttributeError: return [] def refresh_job_headers(self): self.job_list_view.setHorizontalHeaderLabels(["ID", "Name", "Engine", "Priority", "Status", "Time Elapsed", "Frames", "Date Created"]) self.job_list_view.setColumnHidden(0, True) self.job_list_view.horizontalHeader().setSectionResizeMode(1, QHeaderView.ResizeMode.Stretch) self.job_list_view.horizontalHeader().setSectionResizeMode(2, QHeaderView.ResizeMode.ResizeToContents) self.job_list_view.horizontalHeader().setSectionResizeMode(3, QHeaderView.ResizeMode.ResizeToContents) self.job_list_view.horizontalHeader().setSectionResizeMode(4, QHeaderView.ResizeMode.ResizeToContents) self.job_list_view.horizontalHeader().setSectionResizeMode(5, QHeaderView.ResizeMode.ResizeToContents) self.job_list_view.horizontalHeader().setSectionResizeMode(6, QHeaderView.ResizeMode.ResizeToContents) self.job_list_view.horizontalHeader().setSectionResizeMode(7, QHeaderView.ResizeMode.ResizeToContents) # -- Image Code -- # def load_image_path(self, image_path): # Load and set the image using QPixmap try: pixmap = QPixmap(image_path) if not pixmap: logger.error("Error loading image") return self.image_label.setPixmap(pixmap) except Exception as e: logger.error(f"Error loading image path: {e}") def load_image_data(self, pillow_image): try: # Convert the Pillow Image to a QByteArray (byte buffer) byte_array = QByteArray() buffer = QBuffer(byte_array) buffer.open(QIODevice.OpenModeFlag.WriteOnly) pillow_image.save(buffer, "PNG") buffer.close() # Create a QImage from the QByteArray image = QImage.fromData(byte_array) # Create a QPixmap from the QImage pixmap = QPixmap.fromImage(image) if not pixmap: logger.error("Error loading image") return self.image_label.setPixmap(pixmap) except Exception as e: logger.error(f"Error loading image data: {e}") def update_servers(self): found_servers = list(set(ZeroconfServer.found_hostnames() + self.added_hostnames)) found_servers = [x for x in found_servers if ZeroconfServer.get_hostname_properties(x)['api_version'] == API_VERSION] # Always make sure local hostname is first if found_servers and not is_localhost(found_servers[0]): for hostname in found_servers: if is_localhost(hostname): found_servers.remove(hostname) found_servers.insert(0, hostname) break old_count = self.server_list_view.count() # Update proxys for hostname in found_servers: ServerProxyManager.get_proxy_for_hostname(hostname) # setup background updates # Add in all the missing servers current_server_list = [] for i in range(self.server_list_view.count()): current_server_list.append(self.server_list_view.item(i).text()) for hostname in found_servers: if hostname not in current_server_list: properties = ZeroconfServer.get_hostname_properties(hostname) image_path = os.path.join(resources_dir(), f"{properties.get('system_os', 'Monitor')}.png") list_widget = QListWidgetItem(QIcon(image_path), hostname) self.server_list_view.addItem(list_widget) # find any servers that shouldn't be shown any longer servers_to_remove = [] for i in range(self.server_list_view.count()): name = self.server_list_view.item(i).text() if name not in found_servers: servers_to_remove.append(name) # remove any servers that shouldn't be shown any longer for server in servers_to_remove: # Find and remove the item with the specified text for i in range(self.server_list_view.count()): item = self.server_list_view.item(i) if item is not None and item.text() == server: self.server_list_view.takeItem(i) break # Stop searching after the first match is found if not old_count and self.server_list_view.count(): self.server_list_view.setCurrentRow(0) self.server_picked() def create_toolbars(self) -> None: """ Creates and adds the top and right toolbars to the main window. """ # Top Toolbar [PyQt6.QtWidgets.QToolBar] self.topbar = ToolBar(self, orientation=Qt.Orientation.Horizontal, style=Qt.ToolButtonStyle.ToolButtonTextUnderIcon, icon_size=(24, 24)) self.topbar.setMovable(False) resources_directory = resources_dir() # Top Toolbar Buttons self.topbar.add_button( "Settings", f"{resources_directory}/Gear.png", self.menuBar().show_settings) self.topbar.add_button( "Console", f"{resources_directory}/Console.png", self.open_console_window) self.topbar.add_separator() self.topbar.add_button( "Stop Job", f"{resources_directory}/StopSign.png", self.stop_job) self.topbar.add_button( "Delete Job", f"{resources_directory}/Trash.png", self.delete_job) self.topbar.add_button( "Render Log", f"{resources_directory}/Document.png", self.job_logs) self.topbar.add_button( "Download", f"{resources_directory}/Download.png", self.download_files) self.topbar.add_button( "Open Files", f"{resources_directory}/SearchFolder.png", self.open_files) self.topbar.add_button( "New Job", f"{resources_directory}/AddProduct.png", self.new_job) self.addToolBar(Qt.ToolBarArea.TopToolBarArea, self.topbar) # -- Toolbar Buttons -- # def open_console_window(self) -> None: """ Event handler for the "Open Console" button """ self.console_window = ConsoleWindow(self.buffer_handler) self.console_window.buffer_handler = self.buffer_handler self.console_window.show() def engine_browser(self): self.engine_browser_window = EngineBrowserWindow(hostname=self.current_hostname) self.engine_browser_window.show() def job_logs(self) -> None: """ Event handler for the "Logs" button. """ selected_job_ids = self.selected_job_ids() if selected_job_ids: url = f'http://{self.current_server_proxy.hostname}:{self.current_server_proxy.port}/api/job/{selected_job_ids[0]}/logs' self.log_viewer_window = LogViewer(url) self.log_viewer_window.show() def stop_job(self, event): """ Event handler for the Stop Job button """ job_ids = self.selected_job_ids() if not job_ids: return if len(job_ids) == 1: job = next((job for job in self.current_server_proxy.get_all_jobs() if job.get('id') == job_ids[0]), None) if job: display_name = job.get('name', os.path.basename(job.get('input_path', ''))) message = f"Are you sure you want to stop the job:\n{display_name}?" else: return # Job not found, handle this case as needed else: message = f"Are you sure you want to stop these {len(job_ids)} jobs?" # Display the message box and check the response in one go msg_box = QMessageBox(QMessageBox.Icon.Warning, "Stop Job", message, QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, self) if msg_box.exec() == QMessageBox.StandardButton.Yes: for job_id in job_ids: self.current_server_proxy.cancel_job(job_id, confirm=True) self.fetch_jobs(clear_table=True) def delete_job(self, event): """ Event handler for the Delete Job button """ job_ids = self.selected_job_ids() if not job_ids: return if len(job_ids) == 1: job = next((job for job in self.current_server_proxy.get_all_jobs() if job.get('id') == job_ids[0]), None) if job: display_name = job.get('name', os.path.basename(job.get('input_path', ''))) message = f"Are you sure you want to delete the job:\n{display_name}?" else: return # Job not found, handle this case as needed else: message = f"Are you sure you want to delete these {len(job_ids)} jobs?" # Display the message box and check the response in one go msg_box = QMessageBox(QMessageBox.Icon.Warning, "Delete Job", message, QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, self) if msg_box.exec() == QMessageBox.StandardButton.Yes: for job_id in job_ids: self.current_server_proxy.delete_job(job_id, confirm=True) self.fetch_jobs(clear_table=True) def download_files(self, event): job_ids = self.selected_job_ids() if not job_ids: return import webbrowser download_url = (f"http://{self.current_server_proxy.hostname}:{self.current_server_proxy.port}" f"/api/job/{job_ids[0]}/download_all") webbrowser.open(download_url) def open_files(self, event): job_ids = self.selected_job_ids() if not job_ids: return for job_id in job_ids: job_info = self.current_server_proxy.get_job_info(job_id) path = os.path.dirname(job_info['output_path']) launch_url(path) def new_job(self) -> None: file_name, _ = QFileDialog.getOpenFileName(self, "Select Scene File") if file_name: self.new_job_window = NewRenderJobForm(file_name) self.new_job_window.show() if __name__ == "__main__": # lazy load GUI frameworks from PyQt6.QtWidgets import QApplication # load application # QtCore.QCoreApplication.setAttribute(QtCore.Qt.ApplicationAttribute.AA_MacDontSwapCtrlAndMeta) app: QApplication = QApplication(sys.argv) # configure main main_window main_window = MainWindow() # main_window.buffer_handler = buffer_handler app.setActiveWindow(main_window) main_window.show() sys.exit(app.exec())