''' app/ui/main_window.py ''' import ast import datetime import io import logging import os import sys import threading import time from typing import List, Dict, Any, Optional import PIL import humanize from PIL import Image from PyQt6.QtCore import Qt, QByteArray, QBuffer, QIODevice, QThread, pyqtSignal 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.api.serverproxy_manager import ServerProxyManager from src.render_queue import RenderQueue 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.utilities.misc_helper import get_time_elapsed, resources_dir, is_localhost from src.utilities.misc_helper import launch_url from src.utilities.status_utils import RenderStatus from src.utilities.zeroconf_server import ZeroconfServer 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: Optional[QTableWidget] = None self.server_info_ram: Optional[str] = None self.server_info_cpu: Optional[str] = None self.server_info_os: Optional[str] = None self.server_info_gpu: Optional[List[Dict[str, Any]]] = None self.server_info_hostname: Optional[str] = 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.found_servers = [] self.job_data = {} self.bg_update_thread = BackgroundUpdater(window=self) self.bg_update_thread.updated_signal.connect(self.update_ui_data) 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: QVBoxLayout) -> None: """Setup the main user interface layout. Args: main_layout: The main layout container for the UI widgets. """ # 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) # Setup Job Headers 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) # Job List Layout job_list_group = QGroupBox("Job Preview") job_list_layout = QVBoxLayout(job_list_group) job_list_layout.setContentsMargins(0, 0, 0, 0) job_list_layout.addWidget(self.image_label) job_list_layout.addWidget(self.job_list_view, stretch=True) # Add them all to the window main_layout.addLayout(info_layout) right_layout = QVBoxLayout() right_layout.setContentsMargins(0, 0, 0, 0) right_layout.addWidget(job_list_group) main_layout.addLayout(right_layout) def closeEvent(self, event): """Handle window close event with job running confirmation. Args: event: The close event triggered by user. """ 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 -- # def refresh_job_list(self): """Refresh the job list display.""" self.job_list_view.clearContents() self.bg_update_thread.needs_update = True @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.refresh_job_list() # 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 update_ui_data(self): """Update UI data with current server and job information.""" self.update_servers() if not self.current_server_proxy: return server_job_data = self.job_data.get(self.current_server_proxy.hostname) if server_job_data: num_jobs = len(server_job_data) self.job_list_view.setRowCount(num_jobs) for row, job in enumerate(server_job_data): 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', '')) converted_time = datetime.datetime.fromisoformat(job['date_created']) humanized_time = humanize.naturaltime(converted_time) items = [QTableWidgetItem(job['id']), QTableWidgetItem(name), QTableWidgetItem(engine_name), QTableWidgetItem(priority), QTableWidgetItem(display_status), QTableWidgetItem(time_elapsed), QTableWidgetItem(total_frames), QTableWidgetItem(humanized_time)] 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): """Get list of selected job IDs from the job list. Returns: List[str]: List of selected job ID strings. """ 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 [] # -- Image Code -- # def load_image_path(self, image_path): """Load and display an image from file path. Args: image_path: Path to the image file to load. """ # Load and set 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): # Always make sure local hostname is first if self.found_servers and not is_localhost(self.found_servers[0]): for hostname in self.found_servers: if is_localhost(hostname): self.found_servers.remove(hostname) self.found_servers.insert(0, hostname) break old_count = self.server_list_view.count() # Update proxys for hostname in self.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 self.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 self.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: """Open log viewer for selected job. Opens a log viewer window showing the logs for the currently selected job. """ 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): """Stop selected render jobs with user confirmation. Args: event: The button click event. """ 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 job: {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.refresh_job_list() def delete_job(self, event): """Delete selected render jobs with user confirmation. Args: event: The button click event. """ 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.refresh_job_list() 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() class BackgroundUpdater(QThread): """Worker class to fetch job and server information and update the UI""" updated_signal = pyqtSignal() error_signal = pyqtSignal(str) def __init__(self, window): super().__init__() self.window = window self.needs_update = True def run(self): """Main background thread execution loop. Continuously fetches server and job data, updating the main UI every second or when updates are needed. """ try: last_run = 0 while True: now = time.monotonic() if now - last_run >= 1.0 or self.needs_update: self.window.found_servers = list(set(ZeroconfServer.found_hostnames() + self.window.added_hostnames)) self.window.found_servers = [x for x in self.window.found_servers if ZeroconfServer.get_hostname_properties(x)['api_version'] == API_VERSION] if self.window.current_server_proxy: self.window.job_data[self.window.current_server_proxy.hostname] = \ self.window.current_server_proxy.get_all_jobs(ignore_token=False) self.needs_update = False self.updated_signal.emit() time.sleep(0.05) except Exception as e: print(f"ERROR: {e}") self.error_signal.emit(str(e)) 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())