Update core/app.py

core/app.py

@@ -3,12 +3,11 @@
 BackgroundFX Pro – Main Application Entry Point
 Refactored modular architecture – orchestrates specialised components
 
-Changes in this version (2025-08-27):
-- Robust two-stage import with clear diagnostics (tries multiple locations, logs traceback)
-- Properly reports "Two-stage available" based on actual import + init
-- Defensive config patching: adds missing attributes (e.g., use_nvenc) at runtime to avoid AttributeError
-- Cleaner progress handling and error messages
-- Minor hardening around background preparation and status reporting
+2025-08-27 update:
+- Robust Two-Stage importer: tries package import, then direct file import (bypasses
+  any side-effect errors in processing/__init__.py). Clear logging of where it loaded.
+- Config hardening: adds safe defaults for fields like max_model_size/use_nvenc/etc.
+- Defensive error logs with tracebacks for easier diagnosis.
 """
 
 from __future__ import annotations
@@ -23,6 +22,8 @@
 import traceback
 import sys
 import os
+import importlib
+import importlib.util
 from pathlib import Path
 from typing import Optional, Tuple, Dict, Any, Callable
 
@@ -35,9 +36,7 @@
 )
 logger = logging.getLogger("core.app")
 
-# ─────────────────────────────────────────────────────────────────────────────
-# 1.1 Ensure project root is importable (helps HF Spaces and local runs)
-# ─────────────────────────────────────────────────────────────────────────────
+# Ensure project root is importable (helps HF Spaces and local runs)
 PROJECT_FILE = Path(__file__).resolve()
 CORE_DIR = PROJECT_FILE.parent
 ROOT = CORE_DIR.parent
@@ -82,33 +81,63 @@ def _patched_get_type(schema):
 from utils.cv_processing import validate_video_file
 
 # ─────────────────────────────────────────────────────────────────────────────
-# 3.1) Optional: Two-stage import helper with diagnostics
+# 3.1) Optional: Two-stage import with package and file fallbacks
 # ─────────────────────────────────────────────────────────────────────────────
-def _import_two_stage() -> tuple[bool, Any, Dict[str, Any], str]:
+def _import_two_stage() -> tuple[bool, Any, Dict[str, Any], str, str]:
     """
-    Try multiple import paths for the TwoStageProcessor module.
-    Returns (available, TwoStageProcessor or None, CHROMA_PRESETS or {}, import_path_string)
+    Returns (available, TwoStageProcessor|None, CHROMA_PRESETS|{}, import_origin, error_text)
+
+    Tries:
+      1) package imports (preferred)
+      2) direct file imports (bypass processing/__init__.py side-effects)
     """
-    paths_to_try = [
+    pkg_paths = [
         "processing.two_stage.two_stage_processor",
         "two_stage_processor",
         "processing.two_stage_processor",
     ]
-    last_err = None
-    for mod_path in paths_to_try:
+    fs_paths = [
+        ROOT / "processing" / "two_stage" / "two_stage_processor.py",
+        ROOT / "processing" / "two_stage_processor.py",
+        ROOT / "two_stage_processor.py",
+    ]
+
+    # Package imports
+    last_err_text = ""
+    for mod_path in pkg_paths:
+        try:
+            mod = importlib.import_module(mod_path)
+            TSP = getattr(mod, "TwoStageProcessor")
+            PRESETS = getattr(mod, "CHROMA_PRESETS", {"standard": {}})
+            logger.info(f"Two-stage import OK from package '{mod_path}'")
+            return True, TSP, PRESETS, f"pkg:{mod_path}", ""
+        except Exception as e:
+            tb = traceback.format_exc()
+            last_err_text += f"[pkg:{mod_path}] {repr(e)}\n{tb}\n"
+
+    # Direct file imports (bypass __init__.py)
+    for path in fs_paths:
         try:
-            mod = __import__(mod_path, fromlist=["TwoStageProcessor", "CHROMA_PRESETS"])
-            TwoStageProcessor = getattr(mod, "TwoStageProcessor")
-            CHROMA_PRESETS = getattr(mod, "CHROMA_PRESETS", {"standard": {}})
-            logger.info(f"Two-stage import OK from '{mod_path}'")
-            return True, TwoStageProcessor, CHROMA_PRESETS, mod_path
+            if not path.exists():
+                continue
+            spec = importlib.util.spec_from_file_location("bx_two_stage", str(path))
+            if not spec or not spec.loader:
+                continue
+            mod = importlib.util.module_from_spec(spec)
+            sys.modules["bx_two_stage"] = mod
+            spec.loader.exec_module(mod)  # type: ignore[attr-defined]
+            TSP = getattr(mod, "TwoStageProcessor")
+            PRESETS = getattr(mod, "CHROMA_PRESETS", {"standard": {}})
+            logger.info(f"Two-stage import OK from file '{path}'")
+            return True, TSP, PRESETS, f"file:{path}", ""
         except Exception as e:
-            last_err = e
-            logger.debug("Two-stage import failed from '%s': %r\n%s", mod_path, e, traceback.format_exc())
-    logger.warning("Two-stage import failed from all known paths: %r\n%s", last_err, traceback.format_exc())
-    return False, None, {"standard": {}}, ""
+            tb = traceback.format_exc()
+            last_err_text += f"[file:{path}] {repr(e)}\n{tb}\n"
 
-TWO_STAGE_AVAILABLE, _TwoStageProcessor, CHROMA_PRESETS, TWO_STAGE_IMPORT_PATH = _import_two_stage()
+    logger.error("Two-stage import failed from all paths:\n%s", last_err_text or "(no traceback)")
+    return False, None, {"standard": {}}, "", last_err_text or "(no traceback)"
+
+TWO_STAGE_AVAILABLE, _TwoStageProcessor, CHROMA_PRESETS, TWO_STAGE_IMPORT_ORIGIN, TWO_STAGE_IMPORT_ERROR = _import_two_stage()
 
 # ╔══════════════════════════════════════════════════════════════════════════╗
 # ║                           VideoProcessor class                           ║
@@ -118,9 +147,6 @@ class VideoProcessor:
     Main orchestrator – coordinates all specialised components.
     """
 
-    # ─────────────────────────────────────────────────────────────────────
-    # Init
-    # ─────────────────────────────────────────────────────────────────────
     def __init__(self):
         self.config = get_config()
         self._patch_config_defaults(self.config)  # ensure missing attrs exist
@@ -146,14 +172,20 @@ def __init__(self):
     @staticmethod
     def _patch_config_defaults(cfg: Any) -> None:
         """
-        Some downstream modules (FFmpeg/OpenCV writers) may read fields that
-        older configs don't define (e.g., use_nvenc). Add safe defaults here.
+        Some downstream modules may read fields that older configs don't define.
+        Add safe defaults here so AttributeErrors never occur.
         """
         defaults = {
-            "use_nvenc": False,         # guards "'ProcessingConfig' has no attribute 'use_nvenc'"
+            # video i/o & writer
+            "use_nvenc": False,
             "prefer_mp4": True,
-            "video_codec": "mp4v",      # OpenCV fallback codec
-            "audio_copy": True,         # for ffmpeg muxing, if used downstream
+            "video_codec": "mp4v",
+            "audio_copy": True,
+            "ffmpeg_path": "ffmpeg",
+            # model/resource guards
+            "max_model_size": 0,            # bytes or MB; downstream should treat 0/None as "no limit"
+            "max_model_size_bytes": 0,
+            # housekeeping
             "output_dir": str(ROOT / "outputs"),
         }
         for k, v in defaults.items():
@@ -164,7 +196,6 @@ def _patch_config_defaults(cfg: Any) -> None:
         try:
             Path(cfg.output_dir).mkdir(parents=True, exist_ok=True)
         except Exception:
-            # Last-resort fallback to a temp-like folder under ROOT
             out = ROOT / "outputs"
             out.mkdir(parents=True, exist_ok=True)
             cfg.output_dir = str(out)
@@ -223,10 +254,7 @@ def load_models(self, progress_callback: Optional[Callable] = None) -> str:
                         self.two_stage_processor = _TwoStageProcessor(
                             sam2_predictor=sam2_predictor, matanyone_model=mat_model
                         )
-                        logger.info(
-                            "Two-stage processor initialised (imported from '%s')",
-                            TWO_STAGE_IMPORT_PATH or "unknown",
-                        )
+                        logger.info("Two-stage processor initialised (%s)", TWO_STAGE_IMPORT_ORIGIN or "unknown")
                     except Exception as e:
                         logger.warning("Two-stage init failed: %r\n%s", e, traceback.format_exc())
                         self.two_stage_processor = None
@@ -264,13 +292,10 @@ def process_video(
         progress_callback: Optional[Callable] = None,
         use_two_stage: bool = False,
         chroma_preset: str = "standard",
-        key_color_mode: str = "auto",          # NEW
+        key_color_mode: str = "auto",
         preview_mask: bool = False,
         preview_greenscreen: bool = False,
     ) -> Tuple[Optional[str], str]:
-        """
-        Dispatch to single-stage or two-stage pipeline.
-        """
         if not self.models_loaded or not self.core_processor:
             return None, "Models not loaded. Please click “Load Models” first."
         if self.cancel_event.is_set():
@@ -292,7 +317,7 @@ def process_video(
                     custom_background_path,
                     progress_callback,
                     chroma_preset,
-                    key_color_mode,           # NEW
+                    key_color_mode,
                 )
             else:
                 return self._process_single_stage(
@@ -373,7 +398,7 @@ def _process_two_stage(
         custom_background_path: Optional[str],
         progress_callback: Optional[Callable],
         chroma_preset: str,
-        key_color_mode: str,               # NEW
+        key_color_mode: str,
     ) -> Tuple[Optional[str], str]:
         if self.two_stage_processor is None:
             return None, "Two-stage processor not available"
@@ -407,15 +432,15 @@ def _process_two_stage(
 
         chroma_cfg = CHROMA_PRESETS.get(chroma_preset, CHROMA_PRESETS.get("standard", {}))
         logger.info(
-            "Two-stage with preset: %s | key_color_mode=%s | import=%s",
-            chroma_preset, key_color_mode, TWO_STAGE_IMPORT_PATH or "unknown"
+            "Two-stage with preset: %s | key_color_mode=%s | origin=%s",
+            chroma_preset, key_color_mode, TWO_STAGE_IMPORT_ORIGIN or "unknown"
         )
 
         result, message = self.two_stage_processor.process_full_pipeline(
             video_path,
             background,
             final_out,
-            key_color_mode=key_color_mode,   # NEW
+            key_color_mode=key_color_mode,
             chroma_settings=chroma_cfg,
             progress_callback=progress_callback,
         )
@@ -428,7 +453,7 @@ def _process_two_stage(
                 original_video=video_path, processed_video=result
             )
         except Exception as e:
-            logger.warning("Audio mux failed for two-stage, returning video without audio: %r", e)
+            logger.warning("Audio mux failed for two-stage; returning video without audio: %r", e)
             final_path = result
 
         msg = (
@@ -446,7 +471,8 @@ def get_status(self) -> Dict[str, Any]:
         status = {
             "models_loaded": self.models_loaded,
             "two_stage_available": bool(TWO_STAGE_AVAILABLE and (self.two_stage_processor is not None)),
-            "two_stage_import_path": TWO_STAGE_IMPORT_PATH or "",
+            "two_stage_origin": TWO_STAGE_IMPORT_ORIGIN or "",
+            "two_stage_error": TWO_STAGE_IMPORT_ERROR[:5000] if TWO_STAGE_IMPORT_ERROR else "",
             "device": str(self.device_manager.get_optimal_device()),
             "core_processor_loaded": self.core_processor is not None,
             "config": self._safe_config_dict(),
@@ -467,8 +493,8 @@ def _safe_config_dict(self) -> Dict[str, Any]:
         try:
             return self.config.to_dict()
         except Exception:
-            # Build a minimal dict view with known fields
-            keys = ["use_nvenc", "prefer_mp4", "video_codec", "audio_copy", "output_dir"]
+            keys = ["use_nvenc", "prefer_mp4", "video_codec", "audio_copy",
+                    "ffmpeg_path", "max_model_size", "max_model_size_bytes", "output_dir"]
             return {k: getattr(self.config, k, None) for k in keys}
 
     def _safe_memory_usage(self) -> Dict[str, Any]:
@@ -508,7 +534,7 @@ def process_video_fixed(
     progress_callback: Optional[Callable] = None,
     use_two_stage: bool = False,
     chroma_preset: str = "standard",
-    key_color_mode: str = "auto",          # NEW
+    key_color_mode: str = "auto",
     preview_mask: bool = False,
     preview_greenscreen: bool = False,
 ) -> Tuple[Optional[str], str]:
@@ -519,7 +545,7 @@ def process_video_fixed(
         progress_callback,
         use_two_stage,
         chroma_preset,
-        key_color_mode,                     # NEW
+        key_color_mode,
         preview_mask,
         preview_greenscreen,
     )
@@ -528,7 +554,6 @@ def get_model_status() -> Dict[str, Any]:
     return processor.get_status()
 
 def get_cache_status() -> Dict[str, Any]:
-    # Placeholder – could expose FS cache size, etc.
     return processor.get_status()
 
 PROCESS_CANCELLED = processor.cancel_event
@@ -542,13 +567,11 @@ def main():
         logger.info("Starting BackgroundFX Pro")
         logger.info(f"Device: {processor.device_manager.get_optimal_device()}")
         logger.info(
-            "Two-stage available at import-time: %s (path='%s')",
-            TWO_STAGE_AVAILABLE, TWO_STAGE_IMPORT_PATH or "unknown"
+            "Two-stage available at import-time: %s (origin='%s')",
+            TWO_STAGE_AVAILABLE, TWO_STAGE_IMPORT_ORIGIN or "unknown"
         )
 
-        # UI lives in ui/ui_components.py (recent rename from ui/components.py)
         from ui.ui_components import create_interface
-
         demo = create_interface()
         demo.queue().launch(
             server_name="0.0.0.0",