Implement device fallback to CPU for preconfigured pipelines (#5091)

Author: A-Artemis

application/backend/app/api/dependencies.py

@@ -107,12 +107,18 @@ def get_source_update_service(
     return SourceUpdateService(event_bus=event_bus, db_session=db)
 
 
+def get_system_service() -> SystemService:
+    """Provides a SystemService instance for system-level operations."""
+    return SystemService()
+
+
 def get_pipeline_service(
     event_bus: Annotated[EventBus, Depends(get_event_bus)],
     db: Annotated[Session, Depends(get_db)],
+    system_service: Annotated[SystemService, Depends(get_system_service)],
 ) -> PipelineService:
     """Provides a PipelineService instance ."""
-    return PipelineService(event_bus=event_bus, db_session=db)
+    return PipelineService(event_bus=event_bus, db_session=db, system_service=system_service)
 
 
 def get_pipeline_metrics_service(
@@ -126,11 +132,6 @@ def get_pipeline_metrics_service(
     )
 
 
-def get_system_service() -> SystemService:
-    """Provides a SystemService instance for system-level operations."""
-    return SystemService()
-
-
 def get_model_service(
     data_dir: Annotated[Path, Depends(get_data_dir)],
     db: Annotated[Session, Depends(get_db)],

application/backend/app/services/data_collect/data_collector.py

@@ -104,11 +104,12 @@ def __init__(self, data_dir: Path, event_bus: EventBus) -> None:
         )
 
     def _load_pipeline(self) -> None:
-        from app.services import LabelService, PipelineService, ProjectService
+        from app.services import LabelService, PipelineService, ProjectService, SystemService
 
         with get_db_session() as db:
             label_service = LabelService(db_session=db)
-            pipeline_service = PipelineService(event_bus=self.event_bus, db_session=db)
+            system_service = SystemService()
+            pipeline_service = PipelineService(event_bus=self.event_bus, db_session=db, system_service=system_service)
             pipeline = pipeline_service.get_active_pipeline()
             if pipeline is None:
                 logger.info("No active pipeline found, disabling data collection")

application/backend/app/services/pipeline_service.py

@@ -3,6 +3,7 @@
 
 from uuid import UUID
 
+from loguru import logger
 from sqlalchemy.orm import Session
 
 from app.db.schema import PipelineDB
@@ -12,13 +13,16 @@
 from app.services.event.event_bus import EventBus, EventType
 from app.services.parent_process_guard import parent_process_only
 
+from .system_service import DEFAULT_DEVICE, SystemService
+
 MSG_ERR_DELETE_RUNNING_PIPELINE = "Cannot delete a running pipeline."
 
 
 class PipelineService:
-    def __init__(self, event_bus: EventBus, db_session: Session) -> None:
+    def __init__(self, event_bus: EventBus, db_session: Session, system_service: SystemService) -> None:
         self._event_bus: EventBus = event_bus
         self._db_session: Session = db_session
+        self._system_service: SystemService = system_service
 
     def create_pipeline(self, project_id: UUID) -> Pipeline:
         pipeline_repo = PipelineRepository(self._db_session)
@@ -31,8 +35,19 @@ def create_pipeline(self, project_id: UUID) -> Pipeline:
     def get_active_pipeline(self) -> Pipeline | None:
         """Retrieve an active pipeline."""
         pipeline_repo = PipelineRepository(self._db_session)
-        pipeline = pipeline_repo.get_active_pipeline()
-        return Pipeline.model_validate(pipeline) if pipeline is not None else None
+        pipeline_db = pipeline_repo.get_active_pipeline()
+        if pipeline_db is None:
+            return None
+
+        if not self._system_service.validate_device(pipeline_db.device):
+            logger.warning(
+                "The configured device '{}' is not available for pipeline '{}'. Falling back to 'cpu'.",
+                pipeline_db.device,
+                pipeline_db.project_id,
+            )
+            pipeline_db.device = DEFAULT_DEVICE
+            pipeline_repo.update(pipeline_db)
+        return Pipeline.model_validate(pipeline_db)
 
     def get_pipeline_by_id(self, project_id: UUID) -> Pipeline:
         """Retrieve a pipeline by project ID."""

application/backend/app/services/system_service.py

@@ -9,6 +9,7 @@
 from app.schemas.system import DeviceInfo, DeviceType
 
 DEVICE_PATTERN = re.compile(r"^(cpu|xpu|cuda)(-(\d+))?$")
+DEFAULT_DEVICE = "cpu"
 
 
 class SystemService:

application/backend/tests/integration/project_factory.py

@@ -73,6 +73,7 @@ def with_pipeline(
         model_id: str | None = None,
         source_id: str | None = None,
         sink_id: str | None = None,
+        device: str = "cpu",
     ) -> "ProjectTestDataFactory":
         """Add a pipeline to the project."""
         if not self._project:
@@ -84,6 +85,7 @@ def with_pipeline(
             model_revision_id=model_id,
             source_id=source_id,
             sink_id=sink_id,
+            device=device,
         )
         return self
 

application/backend/tests/integration/services/test_dataset_revision_service.py

@@ -9,7 +9,14 @@
 
 from app.db.schema import DatasetItemDB, DatasetRevisionDB, PipelineDB
 from app.models import DatasetItemAnnotationStatus, DatasetItemSubset, Pipeline, Project
-from app.services import DatasetRevisionService, DatasetService, LabelService, PipelineService, ProjectService
+from app.services import (
+    DatasetRevisionService,
+    DatasetService,
+    LabelService,
+    PipelineService,
+    ProjectService,
+    SystemService,
+)
 from app.services.base import ResourceNotFoundError, ResourceType
 from app.services.event.event_bus import EventBus
 
@@ -21,9 +28,17 @@ def fxt_event_bus() -> EventBus:
 
 
 @pytest.fixture
-def fxt_pipeline_service(fxt_event_bus: EventBus, db_session: Session) -> PipelineService:
+def fxt_system_service() -> SystemService:
+    """Fixture to create a SystemService instance."""
+    return SystemService()
+
+
+@pytest.fixture
+def fxt_pipeline_service(
+    fxt_event_bus: EventBus, db_session: Session, fxt_system_service: SystemService
+) -> PipelineService:
     """Fixture to create a PipelineService instance."""
-    return PipelineService(event_bus=fxt_event_bus, db_session=db_session)
+    return PipelineService(event_bus=fxt_event_bus, db_session=db_session, system_service=fxt_system_service)
 
 
 @pytest.fixture

application/backend/tests/integration/services/test_dataset_service.py

@@ -22,7 +22,7 @@
     Project,
     Rectangle,
 )
-from app.services import LabelService, PipelineService, ProjectService
+from app.services import LabelService, PipelineService, ProjectService, SystemService
 from app.services.base import ResourceNotFoundError, ResourceType
 from app.services.dataset_service import (
     DatasetItemFilters,
@@ -40,9 +40,17 @@ def fxt_event_bus() -> EventBus:
 
 
 @pytest.fixture
-def fxt_pipeline_service(fxt_event_bus: EventBus, db_session: Session) -> PipelineService:
+def fxt_system_service() -> SystemService:
+    """Fixture to create a SystemService instance."""
+    return SystemService()
+
+
+@pytest.fixture
+def fxt_pipeline_service(
+    fxt_event_bus: EventBus, db_session: Session, fxt_system_service: SystemService
+) -> PipelineService:
     """Fixture to create a PipelineService instance."""
-    return PipelineService(event_bus=fxt_event_bus, db_session=db_session)
+    return PipelineService(event_bus=fxt_event_bus, db_session=db_session, system_service=fxt_system_service)
 
 
 @pytest.fixture

application/backend/tests/integration/services/test_pipeline_service.py

@@ -10,7 +10,7 @@
 from app.db.schema import PipelineDB, ProjectDB
 from app.models import PipelineStatus
 from app.models.data_collection_policy import FixedRateDataCollectionPolicy
-from app.services import PipelineService, ResourceNotFoundError, ResourceType
+from app.services import PipelineService, ResourceNotFoundError, ResourceType, SystemService
 from app.services.event.event_bus import EventType
 from tests.integration.project_factory import ProjectTestDataFactory
 
@@ -21,9 +21,15 @@ class PipelineField(StrEnum):
 
 
 @pytest.fixture
-def fxt_pipeline_service(fxt_event_bus, db_session) -> PipelineService:
+def fxt_system_service() -> SystemService:
+    """Fixture to create a SystemService instance."""
+    return SystemService()
+
+
+@pytest.fixture
+def fxt_pipeline_service(fxt_event_bus, db_session, fxt_system_service) -> PipelineService:
     """Fixture to create a PipelineService instance with mocked dependencies."""
-    return PipelineService(fxt_event_bus, db_session)
+    return PipelineService(fxt_event_bus, db_session, fxt_system_service)
 
 
 @pytest.fixture
@@ -33,7 +39,7 @@ def fxt_project_with_pipeline(
     """Fixture to create a ProjectDB with an associated PipelineDB."""
 
     def _create_project_with_pipeline(
-        is_running: bool, data_policies: list[dict] | None = None
+        is_running: bool, data_policies: list[dict] | None = None, device: str = "cpu"
     ) -> tuple[ProjectDB, PipelineDB]:
         db_session.add_all(fxt_db_sources)
         db_session.add_all(fxt_db_sinks)
@@ -46,6 +52,7 @@ def _create_project_with_pipeline(
                 model_id=fxt_db_models[0].id,
                 source_id=fxt_db_sources[0].id,
                 sink_id=fxt_db_sinks[0].id,
+                device=device,
             )
             .with_models(fxt_db_models)
             .with_data_policies(data_policies if data_policies else [])
@@ -104,6 +111,19 @@ def test_get_active_pipeline(self, fxt_pipeline_service, fxt_project_with_pipeli
         assert active_pipeline is not None
         assert active_pipeline.project_id == project_id
 
+    def test_get_active_pipeline_device_change(self, fxt_pipeline_service, fxt_project_with_pipeline, db_session):
+        """Test retrieving a pipeline when its original device is no longer available."""
+        db_project, db_pipeline = fxt_project_with_pipeline(is_running=True, data_policies=[], device="xpu-99")
+
+        assert db_pipeline.device == "xpu-99"
+
+        project_id = UUID(db_project.id)
+        active_pipeline = fxt_pipeline_service.get_active_pipeline()
+
+        assert active_pipeline is not None
+        assert active_pipeline.project_id == project_id
+        assert active_pipeline.device == "cpu"
+
     def test_get_non_existent_pipeline(self, fxt_pipeline_service):
         """Test retrieving a non-existent pipeline raises error."""
         pipeline_id = uuid4()

application/backend/tests/integration/services/test_project_service.py

@@ -8,7 +8,7 @@
 
 from app.db.schema import DatasetItemDB, LabelDB, PipelineDB, ProjectDB
 from app.models import Label, Task, TaskType
-from app.services import LabelService, PipelineService, ResourceWithIdAlreadyExistsError
+from app.services import LabelService, PipelineService, ResourceWithIdAlreadyExistsError, SystemService
 from app.services.base import ResourceInUseError, ResourceNotFoundError, ResourceType
 from app.services.event.event_bus import EventBus
 from app.services.label_service import DuplicateLabelsError
@@ -22,9 +22,17 @@ def fxt_event_bus() -> EventBus:
 
 
 @pytest.fixture
-def fxt_pipeline_service(fxt_event_bus: EventBus, db_session: Session) -> PipelineService:
+def fxt_system_service() -> SystemService:
+    """Fixture to create a SystemService instance."""
+    return SystemService()
+
+
+@pytest.fixture
+def fxt_pipeline_service(
+    fxt_event_bus: EventBus, db_session: Session, fxt_system_service: SystemService
+) -> PipelineService:
     """Fixture to create a PipelineService instance."""
-    return PipelineService(event_bus=fxt_event_bus, db_session=db_session)
+    return PipelineService(event_bus=fxt_event_bus, db_session=db_session, system_service=fxt_system_service)
 
 
 @pytest.fixture