Spaces:
Runtime error
Runtime error
File size: 1,672 Bytes
df4a21a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 | """
Fusion service for combining submodel predictions.
"""
from typing import Any, Dict
from app.core.errors import FusionError
from app.core.logging import get_logger
from app.services.model_registry import get_model_registry
logger = get_logger(__name__)
class FusionService:
"""
Service for running fusion predictions.
"""
def __init__(self):
self._registry = get_model_registry()
def fuse(
self,
submodel_outputs: Dict[str, Dict[str, Any]],
**kwargs
) -> Dict[str, Any]:
"""
Run fusion on submodel outputs.
Args:
submodel_outputs: Dictionary mapping submodel name to its prediction output
**kwargs: Additional arguments for the fusion model
Returns:
Standardized prediction dictionary
Raises:
FusionError: If fusion fails
"""
try:
fusion = self._registry.get_fusion()
return fusion.predict(submodel_outputs=submodel_outputs, **kwargs)
except FusionError:
raise
except Exception as e:
logger.error(f"Fusion failed: {e}")
raise FusionError(
message="Fusion prediction failed",
details={"error": str(e)}
)
# Global singleton instance
_fusion_service = None
def get_fusion_service() -> FusionService:
"""
Get the global fusion service instance.
Returns:
FusionService instance
"""
global _fusion_service
if _fusion_service is None:
_fusion_service = FusionService()
return _fusion_service
|