Include telemetry metadata in publisher

Author: andchiind

src/isar/script.py

@@ -155,10 +155,7 @@ def inspections_callback(inspection: Inspection, mission: Mission) -> None:
     robot_heartbeat_thread.start()
     threads.append(robot_heartbeat_thread)
 
-    telemetry_params: list[TelemetryParameters] = robot_interface.get_telemetry_publishers(
-        robot_name=settings.ROBOT_NAME,
-        isar_id=settings.ISAR_ID,
-    )
+    telemetry_params: list[TelemetryParameters] = robot_interface.get_telemetry_publishers()
     for telemetry_parameter in telemetry_params:
         telemetry_thread = MqttTelemetryPublisher(events.mqtt_queue, telemetry_parameter)
         telemetry_thread.start()

src/robot_interface/robot_interface.py

@@ -216,7 +216,7 @@ def generate_media_config(self) -> MediaConfig | None:
 
     @abstractmethod
     def get_telemetry_publishers(
-        self, isar_id: str, robot_name: str
+        self
     ) -> list[TelemetryParameters]:
         """
         Set up telemetry publisher threads to publish regular updates for pose, battery

src/robot_interface/telemetry/mqtt_client.py

@@ -52,7 +52,12 @@ def run(self) -> None:
         while True:
             time.sleep(self.interval)
             try:
-                payload = self.telemetry_method()
+                payload_str = self.telemetry_method()
+                payload_dict = json.loads(payload_str)
+                payload_dict["isar_id"] = isar_id
+                payload_dict["robot_name"] = robot_name
+                payload_dict["timestamp"] = str(datetime.now(UTC))
+                payload = json.dumps(payload_dict)
                 topic = self.topic
             except RobotTelemetryPoseException, RobotTelemetryNoUpdateException:
                 continue
@@ -61,7 +66,7 @@ def run(self) -> None:
                     CloudHealthPayload(
                         isar_id=isar_id,
                         robot_name=robot_name,
-                        timestamp=datetime.now(UTC),
+                        timestamp=str(datetime.now(UTC)),
                     )
                 )
                 topic = f"isar/{isar_id}/cloud_health"

src/robot_interface/telemetry/payloads.py

@@ -12,32 +12,26 @@
 from robot_interface.models.robots.battery_state import BatteryState
 
 
-class TelemetryPayload(BaseModel):
-    isar_id: str
-    robot_name: str
-    timestamp: datetime
-
-
 class CloudHealthPayload(BaseModel):
     isar_id: str
     robot_name: str
     timestamp: datetime
 
 
-class TelemetryPosePayload(TelemetryPayload, BaseModel):
+class TelemetryPosePayload(BaseModel):
     pose: Pose
 
 
-class TelemetryBatteryPayload(TelemetryPayload, BaseModel):
+class TelemetryBatteryPayload(BaseModel):
     battery_level: float
     battery_state: BatteryState | None = None
 
 
-class TelemetryObstacleStatusPayload(TelemetryPayload, BaseModel):
+class TelemetryObstacleStatusPayload(BaseModel):
     obstacle_status: bool
 
 
-class TelemetryPressurePayload(TelemetryPayload, BaseModel):
+class TelemetryPressurePayload(BaseModel):
     pressure_level: float
 
 

tests/test_mocks/robot_interface.py

@@ -85,7 +85,7 @@ def register_inspection_callback(
         return
 
     def get_telemetry_publishers(
-        self, isar_id: str, robot_name: str
+        self
     ) -> list[TelemetryParameters]:
         return []