Use ISAR version 2.2.0

Author: andchiind

pyproject.toml

@@ -20,7 +20,7 @@ classifiers = [
     "Topic :: Scientific/Engineering :: Physics",
     "Topic :: Software Development :: Libraries",
 ]
-dependencies = ["alitra", "isar>=2.1.4"]
+dependencies = ["alitra", "isar>=2.2.0"]
 dynamic = ["version"]
 
 [tool.uv.sources]

src/isar_robot/inspections.py

@@ -1,7 +1,7 @@
 import logging
 import os
 import random
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from pathlib import Path
 
 from robot_interface.models.exceptions.robot_exceptions import (
@@ -72,7 +72,7 @@
 
 
 def create_image(task: TakeImage, telemetry: Telemetry) -> Image:
-    now: datetime = datetime.now(timezone.utc)
+    now: datetime = datetime.now(UTC)
 
     image_metadata: ImageMetadata = ImageMetadata(
         start_time=now,
@@ -86,11 +86,11 @@ def create_image(task: TakeImage, telemetry: Telemetry) -> Image:
     filepath: Path = _select_image_filepath(task)
     data = _read_data_from_file(filepath)
 
-    return Image(metadata=image_metadata, id=task.inspection_id, data=data)
+    return Image(metadata=image_metadata, id=task.id, data=data)
 
 
 def create_thermal_image(task: TakeThermalImage, telemetry: Telemetry) -> Image:
-    now: datetime = datetime.now(timezone.utc)
+    now: datetime = datetime.now(UTC)
 
     image_metadata: ThermalImageMetadata = ThermalImageMetadata(
         start_time=now,
@@ -104,11 +104,11 @@ def create_thermal_image(task: TakeThermalImage, telemetry: Telemetry) -> Image:
     filepath: Path = example_thermal_image
     data = _read_data_from_file(filepath)
 
-    return ThermalImage(metadata=image_metadata, id=task.inspection_id, data=data)
+    return ThermalImage(metadata=image_metadata, id=task.id, data=data)
 
 
 def create_video(task: TakeVideo, telemetry: Telemetry) -> Video:
-    now: datetime = datetime.now(timezone.utc)
+    now: datetime = datetime.now(UTC)
     video_metadata: VideoMetadata = VideoMetadata(
         start_time=now,
         robot_pose=telemetry.get_pose(),
@@ -122,11 +122,11 @@ def create_video(task: TakeVideo, telemetry: Telemetry) -> Video:
     filepath: Path = example_video
     data = _read_data_from_file(filepath)
 
-    return Video(metadata=video_metadata, id=task.inspection_id, data=data)
+    return Video(metadata=video_metadata, id=task.id, data=data)
 
 
 def create_thermal_video(task: TakeThermalVideo, telemetry: Telemetry):
-    now: datetime = datetime.now(timezone.utc)
+    now: datetime = datetime.now(UTC)
     thermal_video_metadata: ThermalVideoMetadata = ThermalVideoMetadata(
         start_time=now,
         robot_pose=telemetry.get_pose(),
@@ -140,13 +140,11 @@ def create_thermal_video(task: TakeThermalVideo, telemetry: Telemetry):
     filepath: Path = example_thermal_video
     data = _read_data_from_file(filepath)
 
-    return ThermalVideo(
-        metadata=thermal_video_metadata, id=task.inspection_id, data=data
-    )
+    return ThermalVideo(metadata=thermal_video_metadata, id=task.id, data=data)
 
 
 def create_audio(task: RecordAudio, telemetry: Telemetry):
-    now: datetime = datetime.now(timezone.utc)
+    now: datetime = datetime.now(UTC)
     audio_metadata: AudioMetadata = AudioMetadata(
         start_time=now,
         robot_pose=telemetry.get_pose(),
@@ -160,11 +158,11 @@ def create_audio(task: RecordAudio, telemetry: Telemetry):
     filepath: Path = example_audio
     data = _read_data_from_file(filepath)
 
-    return Audio(metadata=audio_metadata, id=task.inspection_id, data=data)
+    return Audio(metadata=audio_metadata, id=task.id, data=data)
 
 
 def create_co2_measurement(task: TakeCO2Measurement, telemetry: Telemetry):
-    now: datetime = datetime.now(timezone.utc)
+    now: datetime = datetime.now(UTC)
     gas_measurement_metadata: GasMeasurementMetadata = GasMeasurementMetadata(
         start_time=now,
         robot_pose=telemetry.get_pose(),
@@ -176,7 +174,7 @@ def create_co2_measurement(task: TakeCO2Measurement, telemetry: Telemetry):
 
     return CO2Measurement(
         metadata=gas_measurement_metadata,
-        id=task.inspection_id,
+        id=task.id,
         value=random.normalvariate(0.043, 0.005),
         unit="% v/v",
     )
@@ -185,7 +183,7 @@ def create_co2_measurement(task: TakeCO2Measurement, telemetry: Telemetry):
 def create_acoustic_measurement(
     task: TakeAcousticMeasurement, telemetry: Telemetry
 ) -> AcousticMeasurement:
-    now: datetime = datetime.now(timezone.utc)
+    now: datetime = datetime.now(UTC)
     metadata: AcousticMeasurementMetadata = AcousticMeasurementMetadata(
         start_time=now,
         robot_pose=telemetry.get_pose(),
@@ -207,7 +205,7 @@ def create_acoustic_measurement(
 
     data = _read_data_from_file(example_video)
 
-    return AcousticMeasurement(metadata=metadata, id=task.inspection_id, data=data)
+    return AcousticMeasurement(metadata=metadata, id=task.id, data=data)
 
 
 def _select_image_filepath(task: TakeImage) -> Path:

src/isar_robot/robotinterface.py

@@ -1,10 +1,10 @@
 import logging
 import random
 import time
-from datetime import datetime, timezone
+from collections.abc import Callable
+from datetime import UTC, datetime
 from queue import Queue
 from threading import Thread
-from typing import Callable, List, Optional
 
 from alitra import Position
 from robot_interface.models.exceptions.robot_exceptions import (
@@ -41,9 +41,9 @@ def __init__(self, robot_name: str, isar_id: str) -> None:
         super().__init__(robot_name=robot_name, isar_id=isar_id)
 
         self.telemetry = telemetry.Telemetry()
-        self.last_task_completion_time: datetime = datetime.now(timezone.utc)
+        self.last_task_completion_time: datetime = datetime.now(UTC)
         self.robot_is_home: bool = settings.SHOULD_START_AT_HOME
-        self.mission_simulation: Optional[MissionSimulation] = None
+        self.mission_simulation: MissionSimulation | None = None
 
     def initiate_mission(self, mission: Mission) -> None:
         if (
@@ -116,7 +116,7 @@ def get_inspection(self, task: InspectionTask) -> Inspection:
 
     def register_inspection_callback(
         self, callback_function: Callable[[Inspection, Mission], None]
-    ) -> Optional[Thread]:
+    ) -> Thread | None:
 
         if settings.SHOULD_SIMULATE_INSPECTION_CALLBACK_CRASH:
             return None
@@ -140,7 +140,7 @@ def initialize(self) -> None:
         return
 
     def _get_pose_telemetry(self, isar_id: str, robot_name: str) -> str:
-        current_target: Optional[Position] = None
+        current_target: Position | None = None
         if self.mission_simulation:
             current_task = self.mission_simulation.current_task()
             if current_task and isinstance(current_task, InspectionTask):
@@ -157,8 +157,8 @@ def _get_battery_telemetry(self, isar_id: str, robot_name: str) -> str:
 
     def get_telemetry_publishers(
         self, queue: Queue, isar_id: str, robot_name: str
-    ) -> List[Thread]:
-        publisher_threads: List[Thread] = []
+    ) -> list[Thread]:
+        publisher_threads: list[Thread] = []
 
         pose_publisher: MqttTelemetryPublisher = MqttTelemetryPublisher(
             mqtt_queue=queue,
@@ -249,7 +249,7 @@ def resume(self) -> None:
             )
         self.mission_simulation.resume_mission()
 
-    def generate_media_config(self) -> Optional[MediaConfig]:
+    def generate_media_config(self) -> MediaConfig | None:
         return None
 
     def get_battery_level(self):

src/isar_robot/telemetry.py

@@ -1,6 +1,5 @@
 import random
-from datetime import datetime, timezone
-from typing import Optional
+from datetime import UTC, datetime
 
 from alitra import Frame, Orientation, Pose, Position
 from robot_interface.models.robots.battery_state import BatteryState
@@ -49,7 +48,7 @@ def __init__(self) -> None:
     def get_pose(self) -> Pose:
         return self.current_pose
 
-    def _get_pose(self, current_target: Optional[Position]) -> Pose:
+    def _get_pose(self, current_target: Position | None) -> Pose:
         if not current_target:
             return self.current_pose
 
@@ -64,7 +63,7 @@ def _get_pose(self, current_target: Optional[Position]) -> Pose:
         )
         return self.current_pose
 
-    def _get_battery_level(self, is_home: Optional[bool] = None) -> float:
+    def _get_battery_level(self, is_home: bool | None = None) -> float:
         if settings.SHOULD_HAVE_RANDOM_BATTERY_LEVEL or is_home is None:
             # Return random float in the range [50, 100]
             return random.randint(500, 1000) / 10.0
@@ -80,32 +79,32 @@ def _get_battery_level(self, is_home: Optional[bool] = None) -> float:
             )
         return self.current_battery_level
 
-    def _get_battery_state(self, is_home: Optional[bool] = None) -> BatteryState:
+    def _get_battery_state(self, is_home: bool | None = None) -> BatteryState:
         if settings.SHOULD_HAVE_RANDOM_BATTERY_LEVEL or is_home is None:
             return BatteryState.Normal
 
         return BatteryState.Charging if is_home else BatteryState.Normal
 
     def get_battery_telemetry(
-        self, isar_id: str, robot_name: str, is_home: Optional[bool] = None
+        self, isar_id: str, robot_name: str, is_home: bool | None = None
     ) -> str:
         battery_payload: TelemetryBatteryPayload = TelemetryBatteryPayload(
             battery_level=self._get_battery_level(is_home=is_home),
             battery_state=self._get_battery_state(is_home=is_home),
             isar_id=isar_id,
             robot_name=robot_name,
-            timestamp=datetime.now(timezone.utc),
+            timestamp=datetime.now(UTC),
         )
         return battery_payload.model_dump_json()
 
     def get_pose_telemetry(
-        self, isar_id: str, robot_name: str, current_target: Optional[Position]
+        self, isar_id: str, robot_name: str, current_target: Position | None
     ) -> str:
         pose_payload: TelemetryPosePayload = TelemetryPosePayload(
             pose=self._get_pose(current_target=current_target),
             isar_id=isar_id,
             robot_name=robot_name,
-            timestamp=datetime.now(timezone.utc),
+            timestamp=datetime.now(UTC),
         )
         return pose_payload.model_dump_json()
 
@@ -115,7 +114,7 @@ def get_obstacle_status_telemetry(self, isar_id: str, robot_name: str) -> str:
                 obstacle_status=_get_obstacle_status(),
                 isar_id=isar_id,
                 robot_name=robot_name,
-                timestamp=datetime.now(timezone.utc),
+                timestamp=datetime.now(UTC),
             )
         )
         return obstacle_status_payload.model_dump_json()
@@ -125,6 +124,6 @@ def get_pressure_telemetry(self, isar_id: str, robot_name: str) -> str:
             pressure_level=_get_pressure_level(),
             isar_id=isar_id,
             robot_name=robot_name,
-            timestamp=datetime.now(timezone.utc),
+            timestamp=datetime.now(UTC),
         )
         return pressure_payload.model_dump_json()

tests/test_inspections.py

@@ -20,31 +20,33 @@
 
 
 def test_create_image() -> None:
-    task_actions = TakeImage(target=target, robot_pose=robot_pose)
+    task_actions = TakeImage(id="id", target=target, robot_pose=robot_pose)
 
     inspection_image = inspections.create_image(task_actions, telemetryModule)
 
     assert inspection_image.metadata.file_type == "jpg"
 
 
 def test_create_thermal_image() -> None:
-    task_actions = TakeThermalImage(target=target, robot_pose=robot_pose)
+    task_actions = TakeThermalImage(id="id", target=target, robot_pose=robot_pose)
 
     inspection_image = inspections.create_thermal_image(task_actions, telemetryModule)
 
     assert inspection_image.metadata.file_type == "fff"
 
 
 def test_create_video() -> None:
-    task_actions = TakeImage(target=target, robot_pose=robot_pose)
+    task_actions = TakeImage(id="id", target=target, robot_pose=robot_pose)
 
     inspection_video = inspections.create_video(task_actions, telemetryModule)
 
     assert inspection_video.metadata.file_type == "mp4"
 
 
 def test_create_thermal_video() -> None:
-    task_actions = TakeThermalVideo(target=target, duration=10, robot_pose=robot_pose)
+    task_actions = TakeThermalVideo(
+        id="id", target=target, duration=10, robot_pose=robot_pose
+    )
 
     inspection_video = inspections.create_thermal_video(task_actions, telemetryModule)
 
@@ -53,7 +55,9 @@ def test_create_thermal_video() -> None:
 
 
 def test_create_audio() -> None:
-    task_actions = RecordAudio(target=target, duration=10, robot_pose=robot_pose)
+    task_actions = RecordAudio(
+        id="id", target=target, duration=10, robot_pose=robot_pose
+    )
 
     inspection_recording = inspections.create_audio(task_actions, telemetryModule)
 
@@ -63,6 +67,7 @@ def test_create_audio() -> None:
 
 def test_create_acoustic_measurement() -> None:
     task_actions = TakeAcousticMeasurement(
+        id="id",
         target=target,
         robot_pose=robot_pose,
         frequency_from=35000,
@@ -79,7 +84,9 @@ def test_create_acoustic_measurement() -> None:
 
 def test_select_image_filepath_cloe_kaa(monkeypatch) -> None:
     monkeypatch.setattr(inspections.settings, "PLANT_SHORT_NAME", "kaa")
-    task = TakeImage(target=target, robot_pose=robot_pose, analysis_types=["CLOE"])
+    task = TakeImage(
+        id="id", target=target, robot_pose=robot_pose, analysis_types=["CLOE"]
+    )
 
     assert (
         inspections._select_image_filepath(task) == inspections.example_cloe_image_kaa
@@ -88,7 +95,9 @@ def test_select_image_filepath_cloe_kaa(monkeypatch) -> None:
 
 def test_select_image_filepath_fencilla(monkeypatch) -> None:
     monkeypatch.setattr(inspections.settings, "PLANT_SHORT_NAME", "nls")
-    task = TakeImage(target=target, robot_pose=robot_pose, analysis_types=["Fencilla"])
+    task = TakeImage(
+        id="id", target=target, robot_pose=robot_pose, analysis_types=["Fencilla"]
+    )
 
     assert (
         inspections._select_image_filepath(task) == inspections.example_fencilla_image