brett/Zordon
1
0
Fork 0
mirror of https://github.com/blw1138/Zordon.git synced 2026-02-05 21:56:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

More backend refactoring / improvements

Browse Source
This commit is contained in:
Brett
2026-01-07 11:27:25 -06:00
parent f6ee57fb55
commit d2a30a887f
8 changed files with 166 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/engines/blender/scripts/get_system_info.py
View File

@@ -1,6 +1,10 @@
import bpy
import json
# Force CPU rendering
bpy.context.preferences.addons["cycles"].preferences.compute_device_type = "NONE"
bpy.context.scene.cycles.device = "CPU"
# Ensure Cycles is available
bpy.context.preferences.addons['cycles'].preferences.get_devices()

2
src/engines/core/base_engine.py
View File

@@ -4,7 +4,7 @@ import platform
import subprocess
logger = logging.getLogger()
SUBPROCESS_TIMEOUT = 5
SUBPROCESS_TIMEOUT = 10
class BaseRenderEngine(object):

76
src/engines/engine_manager.py
View File

@@ -3,17 +3,17 @@ import os
import shutil
import threading
import concurrent.futures
from typing import Type
from 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
logger = logging.getLogger()
ENGINE_CLASSES = [Blender, FFMPEG]
class EngineManager:
"""Class that manages different versions of installed render engines and handles fetching and downloading new versions,
if possible.
@@ -23,37 +23,37 @@ class EngineManager:
download_tasks = []
@staticmethod
def supported_engines():
def supported_engines() -> list[type[BaseRenderEngine]]:
return ENGINE_CLASSES
# --- Installed Engines ---
@classmethod
def engine_for_project_path(cls, path):
def engine_class_for_project_path(cls, path) -> type[BaseRenderEngine]:
_, extension = os.path.splitext(path)
extension = extension.lower().strip('.')
for engine_class in cls.supported_engines():
engine = cls.get_latest_engine_instance(engine_class)
if extension in engine.supported_extensions():
return engine
return engine_class
undefined_renderer_support = [x for x in cls.supported_engines() if not cls.get_latest_engine_instance(x).supported_extensions()]
return undefined_renderer_support[0]
@classmethod
def engine_with_name(cls, engine_name):
def engine_class_with_name(cls, engine_name: str) -> type[BaseRenderEngine] | None:
for obj in cls.supported_engines():
if obj.name().lower() == engine_name.lower():
return obj
return None
@classmethod
def get_latest_engine_instance(cls, engine_class):
def get_latest_engine_instance(cls, engine_class: Type[BaseRenderEngine]) -> BaseRenderEngine:
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):
def get_installed_engine_data(cls, filter_name=None, include_corrupt=False, ignore_system=False) -> list:
if not cls.engines_path:
raise FileNotFoundError("Engine path is not set")
@@ -75,7 +75,7 @@ class EngineManager:
# Initialize binary_name with engine name
binary_name = result_dict['engine'].lower()
# Determine the correct binary name based on the engine and system_os
eng = cls.engine_with_name(result_dict['engine'])
eng = cls.engine_class_with_name(result_dict['engine'])
binary_name = eng.binary_names.get(result_dict['system_os'], binary_name)
# Find the path to the binary file
@@ -141,13 +141,13 @@ class EngineManager:
cls.download_engine(engine.name(), update_available['version'], background=True)
@classmethod
def all_version_data_for_engine(cls, engine_name, include_corrupt=False, ignore_system=False):
def all_version_data_for_engine(cls, engine_name:str, include_corrupt=False, ignore_system=False) -> list:
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, system_os=None, cpu=None, ignore_system=None):
def newest_installed_engine_data(cls, engine_name:str, system_os=None, cpu=None, ignore_system=None) -> list:
system_os = system_os or current_system_os()
cpu = cpu or current_system_cpu()
@@ -157,37 +157,37 @@ class EngineManager:
return filtered[0]
except IndexError:
logger.error(f"Cannot find newest engine version for {engine_name}-{system_os}-{cpu}")
return None
return []
@classmethod
def is_version_installed(cls, engine, version, system_os=None, cpu=None, ignore_system=False):
def is_version_installed(cls, engine_name:str, version, system_os=None, cpu=None, ignore_system=False):
system_os = system_os or current_system_os()
cpu = cpu or current_system_cpu()
filtered = [x for x in cls.get_installed_engine_data(filter_name=engine, ignore_system=ignore_system) if
filtered = [x for x in cls.get_installed_engine_data(filter_name=engine_name, ignore_system=ignore_system) if
x['system_os'] == system_os and x['cpu'] == cpu and x['version'] == version]
return filtered[0] if filtered else False
@classmethod
def version_is_available_to_download(cls, engine, version, system_os=None, cpu=None):
def version_is_available_to_download(cls, engine_name:str, version, system_os=None, cpu=None):
try:
downloader = cls.engine_with_name(engine).downloader()
downloader = cls.engine_class_with_name(engine_name).downloader()
return downloader.version_is_available_to_download(version=version, system_os=system_os, cpu=cpu)
except Exception as e:
logger.debug(f"Exception in version_is_available_to_download: {e}")
return None
@classmethod
def find_most_recent_version(cls, engine=None, system_os=None, cpu=None, lts_only=False):
def find_most_recent_version(cls, engine_name:str, system_os=None, cpu=None, lts_only=False) -> dict:
try:
downloader = cls.engine_with_name(engine).downloader()
downloader = cls.engine_class_with_name(engine_name).downloader()
return downloader.find_most_recent_version(system_os=system_os, cpu=cpu)
except Exception as e:
logger.debug(f"Exception in find_most_recent_version: {e}")
return None
return {}
@classmethod
def is_engine_update_available(cls, engine_class, ignore_system_installs=False):
def is_engine_update_available(cls, engine_class: Type[BaseRenderEngine], ignore_system_installs=False):
logger.debug(f"Checking for updates to {engine_class.name()}")
latest_version = engine_class.downloader().find_most_recent_version()
@@ -209,23 +209,23 @@ class EngineManager:
return [engine for engine in cls.supported_engines() if hasattr(engine, "downloader") and engine.downloader()]
@classmethod
def get_existing_download_task(cls, engine, version, system_os=None, cpu=None):
def get_existing_download_task(cls, engine_name, version, system_os=None, cpu=None):
for task in cls.download_tasks:
task_parts = task.name.split('-')
task_engine, task_version, task_system_os, task_cpu = task_parts[:4]
if engine == task_engine and version == task_version:
if engine_name == task_engine and version == task_version:
if system_os in (task_system_os, None) and cpu in (task_cpu, None):
return task
return None
@classmethod
def download_engine(cls, engine, version, system_os=None, cpu=None, background=False, ignore_system=False):
def download_engine(cls, engine_name, version, system_os=None, cpu=None, background=False, ignore_system=False):
engine_to_download = cls.engine_with_name(engine)
existing_task = cls.get_existing_download_task(engine, version, system_os, cpu)
engine_to_download = cls.engine_class_with_name(engine_name)
existing_task = cls.get_existing_download_task(engine_name, version, system_os, cpu)
if existing_task:
logger.debug(f"Already downloading {engine} {version}")
logger.debug(f"Already downloading {engine_name} {version}")
if not background:
existing_task.join() # If download task exists, wait until it's done downloading
return None
@@ -235,7 +235,7 @@ class EngineManager:
elif not cls.engines_path:
raise FileNotFoundError("Engines path must be set before requesting downloads")
thread = EngineDownloadWorker(engine, version, system_os, cpu)
thread = EngineDownloadWorker(engine_name, version, system_os, cpu)
cls.download_tasks.append(thread)
thread.start()
@@ -243,29 +243,29 @@ class EngineManager:
return thread
thread.join()
found_engine = cls.is_version_installed(engine, version, system_os, cpu, ignore_system) # Check that engine downloaded
found_engine = cls.is_version_installed(engine_name, version, system_os, cpu, ignore_system) # Check that engine downloaded
if not found_engine:
logger.error(f"Error downloading {engine}")
logger.error(f"Error downloading {engine_name}")
return found_engine
@classmethod
def delete_engine_download(cls, engine, version, system_os=None, cpu=None):
logger.info(f"Requested deletion of engine: {engine}-{version}")
def delete_engine_download(cls, engine_name, version, system_os=None, cpu=None):
logger.info(f"Requested deletion of engine: {engine_name}-{version}")
found = cls.is_version_installed(engine, version, system_os, cpu)
found = cls.is_version_installed(engine_name, version, system_os, cpu)
if found and found['type'] == 'managed': # don't delete system installs
# find the root directory of the engine executable
root_dir_name = '-'.join([engine, version, found['system_os'], found['cpu']])
root_dir_name = '-'.join([engine_name, version, found['system_os'], found['cpu']])
remove_path = os.path.join(found['path'].split(root_dir_name)[0], root_dir_name)
# delete the file path
logger.info(f"Deleting engine at path: {remove_path}")
shutil.rmtree(remove_path, ignore_errors=False)
logger.info(f"Engine {engine}-{version}-{found['system_os']}-{found['cpu']} successfully deleted")
logger.info(f"Engine {engine_name}-{version}-{found['system_os']}-{found['cpu']} successfully deleted")
return True
elif found: # these are managed by the system / user. Don't delete these.
logger.error(f'Cannot delete requested {engine} {version}. Managed externally.')
logger.error(f'Cannot delete requested {engine_name} {version}. Managed externally.')
else:
logger.error(f"Cannot find engine: {engine}-{version}")
logger.error(f"Cannot find engine: {engine_name}-{version}")
return False
# --- Background Tasks ---
@@ -277,7 +277,7 @@ class EngineManager:
@classmethod
def create_worker(cls, engine_name, input_path, output_path, engine_version=None, args=None, parent=None, name=None):
worker_class = cls.engine_with_name(engine_name).worker_class()
worker_class = cls.engine_class_with_name(engine_name).worker_class()
# check to make sure we have versions installed
all_versions = cls.all_version_data_for_engine(engine_name)
@@ -342,7 +342,7 @@ class EngineDownloadWorker(threading.Thread):
return existing_download
# Get the appropriate downloader class based on the engine type
downloader = EngineManager.engine_with_name(self.engine).downloader()
downloader = EngineManager.engine_class_with_name(self.engine).downloader()
downloader.download_engine( self.version, download_location=EngineManager.engines_path,
system_os=self.system_os, cpu=self.cpu, timeout=300, progress_callback=self._update_progress)
except Exception as e: