Simplify uploader and make tests deterministic

Author: andchiind

src/isar/models/events.py

@@ -16,8 +16,6 @@
 from robot_interface.models.mission.task import InspectionTask
 from robot_interface.telemetry.mqtt_client import MQTTQueueType
 
-InspectionQueueTuple = tuple[Inspection, Mission]
-
 
 class EmptyMessage:
     def __str__(self) -> str:
@@ -28,8 +26,8 @@ def __str__(self) -> str:
 
 
 class Event[T](Queue[T]):
-    def __init__(self, name: str) -> None:
-        super().__init__(maxsize=1)
+    def __init__(self, name: str, maxsize: int = 1) -> None:
+        super().__init__(maxsize=maxsize)
         self.name = name
 
     def trigger_event(self, data: T, timeout: int | None = None) -> None:
@@ -90,8 +88,8 @@ def __init__(self) -> None:
         self.state_machine_events: StateMachineEvents = StateMachineEvents()
         self.robot_service_events: RobotServiceEvents = RobotServiceEvents()
 
-        self.upload_queue: Queue[InspectionQueueTuple] = Queue[InspectionQueueTuple](
-            maxsize=10
+        self.upload_event: Event[tuple[Inspection, Mission]] = Event(
+            "uploader", maxsize=10
         )
 
         self.mqtt_queue: Queue[MQTTQueueType] = Queue[MQTTQueueType]()

src/isar/modules.py

@@ -85,22 +85,18 @@ class ApplicationContainer(containers.DeclarativeContainer):
         mqtt_publisher=mqtt_client,
     )
 
-    # Inspection data service
-    inspection_service = providers.Singleton(
-        RobotInspectionService,
-        events=events,
-        robot=robot_interface,
-        mqtt_publisher=mqtt_client,
-    )
-
     # Uploader
     uploader = providers.Singleton(
         Uploader,
-        events=events,
         storage_handlers=storage_handlers,
         mqtt_publisher=mqtt_client,
     )
 
+    # Inspection data service
+    inspection_service = providers.Singleton(
+        RobotInspectionService, events=events, robot=robot_interface, uploader=uploader
+    )
+
 
 def get_injector() -> ApplicationContainer:
     container = ApplicationContainer()

src/isar/robot/robot_inspection_service.py

@@ -1,18 +1,12 @@
 import logging
 from collections.abc import Callable
-from queue import Queue
 from threading import Event as ThreadEvent
 from threading import Thread
 
 from isar.config.settings import settings
-from isar.models.events import (
-    EventConflictError,
-    Events,
-    EventTimeoutError,
-    RobotServiceEvents,
-    StateMachineEvents,
-)
+from isar.models.events import Event, EventConflictError, Events, EventTimeoutError
 from isar.robot.function_thread import FunctionThread
+from isar.storage.uploader import Uploader
 from robot_interface.models.exceptions.robot_exceptions import (
     RobotException,
     RobotRetrieveInspectionException,
@@ -21,18 +15,17 @@
 from robot_interface.models.mission.mission import Mission
 from robot_interface.models.mission.task import InspectionTask
 from robot_interface.robot_interface import RobotInterface
-from robot_interface.telemetry.mqtt_client import MqttClientInterface
 
 
-def robot_upload_inspection(
-    robot: RobotInterface,
+def fetch_and_upload_inspection(
+    get_inspection_function: Callable[[InspectionTask], Inspection],
     logger: logging.Logger,
+    upload_function: Callable[[Inspection, Mission], None],
     task: InspectionTask,
     mission: Mission,
-    upload_queue: Queue,
 ) -> None:
     try:
-        inspection: Inspection = robot.get_inspection(task=task)
+        inspection: Inspection = get_inspection_function(task)
         if task.id != inspection.id:
             logger.warning(
                 f"The id of task ({task.id}) "
@@ -45,33 +38,30 @@ def robot_upload_inspection(
         return
 
     if not inspection:
-        logger.warning(
-            f"No inspection result data retrieved for task {str(task.id)[:8]}"
-        )
+        logger.error(f"No inspection result data retrieved for task {str(task.id)[:8]}")
+        return
 
     inspection.metadata.tag_id = task.tag_id
     inspection.metadata.analysis_types = task.analysis_types
 
-    message: tuple[Inspection, Mission] = (
-        inspection,
-        mission,
-    )
-    upload_queue.put(message)
-    logger.info(f"Inspection result: {str(inspection.id)[:8]} queued for upload")
+    upload_function(inspection, mission)
 
 
 class RobotInspectionService:
     def __init__(
         self,
         events: Events,
         robot: RobotInterface,
-        mqtt_publisher: MqttClientInterface,
+        uploader: Uploader,
     ) -> None:
         self.logger = logging.getLogger("uploader")
-        self.state_machine_events: StateMachineEvents = events.state_machine_events
-        self.robot_service_events: RobotServiceEvents = events.robot_service_events
-        self.mqtt_publisher: MqttClientInterface = mqtt_publisher
-        self.upload_queue: Queue = events.upload_queue
+        self.upload_task_event: Event[tuple[InspectionTask, Mission]] = (
+            events.robot_service_events.request_inspection_upload
+        )
+        self.upload_inspection_event: Event[tuple[Inspection, Mission]] = (
+            events.upload_event
+        )
+        self.uploader: Uploader = uploader
         self.robot: RobotInterface = robot
         self.upload_inspection_threads: list[FunctionThread] = []
         self.signal_exit: ThreadEvent = ThreadEvent()
@@ -107,7 +97,7 @@ def _restart_inspection_thread_if_stopped(self) -> None:
 
     def register_and_monitor_inspection_callback(
         self,
-        callback_function: Callable,
+        callback_function: Callable[[Inspection, Mission], None],
     ) -> None:
         self.inspection_callback_function = callback_function
 
@@ -121,18 +111,33 @@ def register_and_monitor_inspection_callback(
     def run(self) -> None:
         try:
             while not self.signal_exit.wait(0):
-                upload_request: tuple[(InspectionTask, Mission)] | None = (
-                    self.robot_service_events.request_inspection_upload.consume_event()
+
+                upload_task_request: tuple[(InspectionTask, Mission)] | None = (
+                    self.upload_task_event.consume_event()
                 )
-                if upload_request is not None:
+
+                if upload_task_request is not None:
                     self.upload_inspection_threads.append(
                         FunctionThread(
-                            robot_upload_inspection,
-                            self.robot,
+                            fetch_and_upload_inspection,
+                            self.robot.get_inspection,
                             self.logger,
-                            upload_request[0],
-                            upload_request[1],
-                            self.upload_queue,
+                            self.uploader.upload_inspection,
+                            upload_task_request[0],
+                            upload_task_request[1],
+                        )
+                    )
+
+                upload_inspection_request: tuple[Inspection, Mission] | None = (
+                    self.upload_inspection_event.consume_event()
+                )
+
+                if upload_inspection_request is not None:
+                    self.upload_inspection_threads.append(
+                        FunctionThread(
+                            self.uploader.upload_inspection,
+                            upload_inspection_request[0],
+                            upload_inspection_request[1],
                         )
                     )
 

src/isar/script.py

@@ -10,7 +10,7 @@
 from isar.config.log import setup_loggers
 from isar.config.open_telemetry import instrument_fastapi, setup_open_telemetry
 from isar.config.settings import robot_settings, settings
-from isar.models.events import Events, InspectionQueueTuple
+from isar.models.events import Events
 from isar.modules import ApplicationContainer, get_injector
 from isar.robot.robot_inspection_service import RobotInspectionService
 from isar.robot.robot_service import RobotService
@@ -22,7 +22,6 @@
     RobotInfoPublisher,
 )
 from isar.state_machine.state_machine import StateMachine
-from isar.storage.uploader import Uploader
 from robot_interface.models.inspection.inspection import Inspection
 from robot_interface.models.mission.mission import Mission
 from robot_interface.robot_interface import RobotInterface
@@ -83,7 +82,6 @@ def start() -> None:
     print_startup_info()
 
     state_machine: StateMachine = injector.state_machine()
-    uploader: Uploader = injector.uploader()
     robot_interface: RobotInterface = injector.robot_interface()
     events: Events = injector.events()
     robot: RobotService = injector.robot()
@@ -110,12 +108,6 @@ def start() -> None:
     inspection_service_thread.start()
     threads.append(inspection_service_thread)
 
-    uploader_thread: Thread = Thread(
-        target=uploader.run, name="ISAR Uploader", daemon=True
-    )
-    uploader_thread.start()
-    threads.append(uploader_thread)
-
     robot_service_thread: Thread = Thread(
         target=robot.run, name="Robot service", daemon=True
     )
@@ -125,11 +117,7 @@ def start() -> None:
     if settings.UPLOAD_INSPECTIONS_ASYNC:
 
         def inspections_callback(inspection: Inspection, mission: Mission) -> None:
-            message: InspectionQueueTuple = (
-                inspection,
-                mission,
-            )
-            state_machine.events.upload_queue.put(message)
+            state_machine.events.upload_event.trigger_event((inspection, mission))
 
         inspection_service.register_and_monitor_inspection_callback(
             inspections_callback

src/isar/services/utilities/robot_utilities.py

@@ -1,5 +1,3 @@
-import logging
-
 from robot_interface.models.robots.media import MediaConfig
 from robot_interface.robot_interface import RobotInterface
 
@@ -14,7 +12,6 @@ def __init__(
         robot: RobotInterface,
     ):
         self.robot: RobotInterface = robot
-        self.logger = logging.getLogger("api")
 
     def generate_media_config(self) -> MediaConfig | None:
         return self.robot.generate_media_config()