|
""" |
|
Core Models Module - Simplified redirect to working model loader |
|
""" |
|
|
|
import logging |
|
|
|
logger = logging.getLogger(__name__) |
|
|
|
class ModelManager: |
|
""" |
|
Compatibility wrapper that redirects to the working ModelLoader |
|
Provides the same interface that CoreVideoProcessor expects |
|
""" |
|
|
|
def __init__(self): |
|
self._loader = None |
|
self._device_mgr = None |
|
self._memory_mgr = None |
|
|
|
def _get_loader(self): |
|
"""Lazy initialization of model loader""" |
|
if self._loader is None: |
|
try: |
|
from models.loaders.model_loader import ModelLoader |
|
from utils.hardware.device_manager import DeviceManager |
|
from utils.system.memory_manager import MemoryManager |
|
|
|
self._device_mgr = DeviceManager() |
|
self._memory_mgr = MemoryManager() |
|
self._loader = ModelLoader(self._device_mgr, self._memory_mgr) |
|
|
|
except ImportError as e: |
|
logger.error(f"Failed to import ModelLoader dependencies: {e}") |
|
|
|
class DummyLoader: |
|
def get_sam2(self): return None |
|
def get_matanyone(self): return None |
|
def load_all_models(self, *args, **kwargs): return None, None |
|
def cleanup(self): pass |
|
|
|
self._loader = DummyLoader() |
|
|
|
return self._loader |
|
|
|
def load_all(self): |
|
"""Load all models (SAM2 and MatAnyone)""" |
|
try: |
|
loader = self._get_loader() |
|
sam2, matanyone = loader.load_all_models() |
|
logger.info(f"Models loaded - SAM2: {sam2 is not None}, MatAnyone: {matanyone is not None}") |
|
return sam2, matanyone |
|
except Exception as e: |
|
logger.error(f"Failed to load models: {e}") |
|
return None, None |
|
|
|
def get_sam2(self): |
|
"""Get SAM2 model/predictor""" |
|
loader = self._get_loader() |
|
return loader.get_sam2() if hasattr(loader, 'get_sam2') else None |
|
|
|
def get_matanyone(self): |
|
"""Get MatAnyone model""" |
|
loader = self._get_loader() |
|
return loader.get_matanyone() if hasattr(loader, 'get_matanyone') else None |
|
|
|
def cleanup(self): |
|
"""Cleanup models and free memory""" |
|
loader = self._get_loader() |
|
if hasattr(loader, 'cleanup'): |
|
loader.cleanup() |
|
|
|
def process_frame(self, image, mask=None): |
|
"""Process a single frame - for compatibility""" |
|
|
|
|
|
|
|
logger.warning("ModelManager.process_frame called - this is deprecated") |
|
return { |
|
"alpha": mask if mask is not None else image[:,:,0], |
|
"foreground": image, |
|
"confidence": [0.5] |
|
} |
|
|
|
|
|
ModelType = None |
|
ModelFactory = ModelManager |
|
ModelConfig = None |
|
|
|
__all__ = ['ModelManager'] |