brett/Zordon
1
0
Fork 0
mirror of https://github.com/blw1138/Zordon.git synced 2026-06-09 13:39:24 -05:00
Code Issues Packages Projects Releases Wiki Activity

Windows path fixes (#129)

Browse Source 
* Change uses of os.path to use Pathlib

* Add return types and type hints

* Add more docstrings

* Add missing import to api_server
This commit is contained in:
Brett
2026-01-18 00:18:43 -06:00
committed by GitHub
parent ee9f44e4c4
commit 74dce5cc3d
15 changed files with 536 additions and 262 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/engines/engine_manager.py
+146 -17
View File
@@ -3,12 +3,13 @@ import os
import shutil
import threading
import concurrent.futures
from typing import Type
from pathlib import Path
from typing import Type, List, Dict, Any, Optional
from src.engines.core.base_engine import BaseRenderEngine
from src.engines.blender.blender_engine import Blender
from src.engines.ffmpeg.ffmpeg_engine import FFMPEG
from src.utilities.misc_helper import system_safe_path, current_system_os, current_system_cpu
from src.utilities.misc_helper import current_system_os, current_system_cpu
logger = logging.getLogger()
@@ -19,17 +20,30 @@ class EngineManager:
if possible.
"""
engines_path = None
download_tasks = []
engines_path: Optional[str] = None
download_tasks: List[Any] = []
@staticmethod
def supported_engines() -> list[type[BaseRenderEngine]]:
"""Return list of supported engine classes.
Returns:
List[Type[BaseRenderEngine]]: List of available engine classes.
"""
return ENGINE_CLASSES
# --- Installed Engines ---
@classmethod
def engine_class_for_project_path(cls, path) -> type[BaseRenderEngine]:
def engine_class_for_project_path(cls, path: str) -> Type[BaseRenderEngine]:
"""Find engine class that can handle the given project file.
Args:
path: Path to project file.
Returns:
Type[BaseRenderEngine]: Engine class that can handle the file.
"""
_, extension = os.path.splitext(path)
extension = extension.lower().strip('.')
for engine_class in cls.supported_engines():
@@ -40,7 +54,15 @@ class EngineManager:
return undefined_renderer_support[0]
@classmethod
def engine_class_with_name(cls, engine_name: str) -> type[BaseRenderEngine] | None:
def engine_class_with_name(cls, engine_name: str) -> Optional[Type[BaseRenderEngine]]:
"""Find engine class by name.
Args:
engine_name: Name of engine to find.
Returns:
Optional[Type[BaseRenderEngine]]: Engine class if found, None otherwise.
"""
for obj in cls.supported_engines():
if obj.name().lower() == engine_name.lower():
return obj
@@ -48,13 +70,34 @@ class EngineManager:
@classmethod
def get_latest_engine_instance(cls, engine_class: Type[BaseRenderEngine]) -> BaseRenderEngine:
"""Create instance of latest installed engine version.
Args:
engine_class: Engine class to instantiate.
Returns:
BaseRenderEngine: Instance of engine with latest version.
"""
newest = cls.newest_installed_engine_data(engine_class.name())
engine = engine_class(newest["path"])
return engine
@classmethod
def get_installed_engine_data(cls, filter_name=None, include_corrupt=False, ignore_system=False) -> list:
def get_installed_engine_data(cls, filter_name: Optional[str] = None, include_corrupt: bool = False,
ignore_system: bool = False) -> List[Dict[str, Any]]:
"""Get data about installed render engines.
Args:
filter_name: Optional engine name to filter by.
include_corrupt: Whether to include potentially corrupted installations.
ignore_system: Whether to ignore system-installed engines.
Returns:
List[Dict[str, Any]]: List of installed engine data.
Raises:
FileNotFoundError: If engines path is not set.
"""
if not cls.engines_path:
raise FileNotFoundError("Engine path is not set")
@@ -79,11 +122,9 @@ class EngineManager:
binary_name = eng.binary_names.get(result_dict['system_os'], binary_name)
# Find the path to the binary file
path = next(
(os.path.join(root, binary_name) for root, _, files in
os.walk(system_safe_path(os.path.join(cls.engines_path, directory))) if binary_name in files),
None
)
search_root = cls.engines_path / directory
match = next((p for p in search_root.rglob(binary_name) if p.is_file()), None)
path = str(match) if match else None
result_dict['path'] = path
# fetch version number from binary - helps detect corrupted downloads - disabled due to perf issues
@@ -133,7 +174,8 @@ class EngineManager:
# --- Check for Updates ---
@classmethod
def update_all_engines(cls):
def update_all_engines(cls) -> None:
"""Check for and download updates for all downloadable engines."""
for engine in cls.downloadable_engines():
update_available = cls.is_engine_update_available(engine)
if update_available:
@@ -142,12 +184,33 @@ class EngineManager:
@classmethod
def all_version_data_for_engine(cls, engine_name:str, include_corrupt=False, ignore_system=False) -> list:
"""Get all version data for a specific engine.
Args:
engine_name: Name of engine to query.
include_corrupt: Whether to include corrupt installations.
ignore_system: Whether to ignore system installations.
Returns:
list: Sorted list of engine version data (newest first).
"""
versions = cls.get_installed_engine_data(filter_name=engine_name, include_corrupt=include_corrupt, ignore_system=ignore_system)
sorted_versions = sorted(versions, key=lambda x: x['version'], reverse=True)
return sorted_versions
@classmethod
def newest_installed_engine_data(cls, engine_name:str, system_os=None, cpu=None, ignore_system=None) -> list:
"""Get newest installed engine data for specific platform.
Args:
engine_name: Name of engine to query.
system_os: Operating system to filter by (defaults to current).
cpu: CPU architecture to filter by (defaults to current).
ignore_system: Whether to ignore system installations.
Returns:
list: Newest engine data or empty list if not found.
"""
system_os = system_os or current_system_os()
cpu = cpu or current_system_cpu()
@@ -160,7 +223,19 @@ class EngineManager:
return []
@classmethod
def is_version_installed(cls, engine_name:str, version, system_os=None, cpu=None, ignore_system=False):
def is_version_installed(cls, engine_name:str, version:str, system_os=None, cpu=None, ignore_system=False):
"""Check if specific engine version is installed.
Args:
engine_name: Name of engine to check.
version: Version string to check.
system_os: Operating system to check (defaults to current).
cpu: CPU architecture to check (defaults to current).
ignore_system: Whether to ignore system installations.
Returns:
Engine data if found, False otherwise.
"""
system_os = system_os or current_system_os()
cpu = cpu or current_system_cpu()
@@ -206,6 +281,11 @@ class EngineManager:
@classmethod
def downloadable_engines(cls):
"""Get list of engines that support downloading.
Returns:
List[Type[BaseRenderEngine]]: Engines with downloader capability.
"""
return [engine for engine in cls.supported_engines() if hasattr(engine, "downloader") and engine.downloader()]
@classmethod
@@ -272,10 +352,41 @@ class EngineManager:
@classmethod
def active_downloads(cls) -> list:
"""Get list of currently active download tasks.
Returns:
list: List of active EngineDownloadWorker threads.
"""
return [x for x in cls.download_tasks if x.is_alive()]
@classmethod
def create_worker(cls, engine_name, input_path, output_path, engine_version=None, args=None, parent=None, name=None):
def create_worker(cls, engine_name: str, input_path: Path, output_path: Path, engine_version=None, args=None, parent=None, name=None):
"""
Create and return a worker instance for a specific engine.
This resolves the appropriate engine binary/path for the requested engine and version,
downloading the engine if necessary (when a specific version is requested and not found
locally). The returned worker is constructed with string paths for compatibility with
worker implementations that expect `str` rather than `Path`.
Args:
engine_name: The engine name used to resolve an engine class and its worker.
input_path: Path to the input file/folder for the worker to process.
output_path: Path where the worker should write output.
engine_version: Optional engine version to use. If `None` or `'latest'`, the newest
installed version is used. If a specific version is provided and not installed,
the engine will be downloaded.
args: Optional arguments passed through to the worker (engine-specific).
parent: Optional Qt/GUI parent object passed through to the worker constructor.
name: Optional name/label passed through to the worker constructor.
Returns:
An instance of the engine-specific worker class.
Raises:
FileNotFoundError: If no versions of the engine are installed, if the requested
version cannot be found or downloaded, or if the engine path cannot be resolved.
"""
worker_class = cls.engine_class_with_name(engine_name).worker_class()
@@ -306,7 +417,7 @@ class EngineManager:
if not engine_path:
raise FileNotFoundError(f"Cannot find requested engine version {engine_version}")
return worker_class(input_path=input_path, output_path=output_path, engine_path=engine_path, args=args,
return worker_class(input_path=str(input_path), output_path=str(output_path), engine_path=engine_path, args=args,
parent=parent, name=name)
@@ -323,6 +434,14 @@ class EngineDownloadWorker(threading.Thread):
cpu (str, optional): Requested CPU architecture. Defaults to system CPU type.
"""
def __init__(self, engine, version, system_os=None, cpu=None):
"""Initialize download worker for specific engine version.
Args:
engine: Name of engine to download.
version: Version of engine to download.
system_os: Target operating system (defaults to current).
cpu: Target CPU architecture (defaults to current).
"""
super().__init__()
self.engine = engine
self.version = version
@@ -331,9 +450,19 @@ class EngineDownloadWorker(threading.Thread):
self.percent_complete = 0
def _update_progress(self, current_progress):
"""Update download progress.
Args:
current_progress: Current download progress percentage (0-100).
"""
self.percent_complete = current_progress
def run(self):
def run(self):
"""Execute the download process.
Checks if engine version already exists, then downloads if not found.
Handles cleanup and error reporting.
"""
try:
existing_download = EngineManager.is_version_installed(self.engine, self.version, self.system_os, self.cpu,
ignore_system=True)