Get inspections

src/isar_exr/api/energy_robotics_api.py

@@ -1,8 +1,11 @@
 from isar_exr.api.graphql_client import GraphqlClient
+from isar_exr.models.mission import MissionReports
 from isar_exr.models.step_status import ExrMissionStatus
 from robot_interface.models.mission.status import StepStatus
-from typing import Any
+from typing import Any, Type
 from isar_exr.models.exceptions import NoMissionRunningException
+from isar_exr.config.settings import settings
+from robot_interface.models.inspection import Inspection, InspectionMetadata
 
 
 class Api:
@@ -39,3 +42,101 @@ def is_mission_running(self, exr_robot_id: str):
         response_dict: dict[str, Any] = self.client.query(query_string, params)
         is_running: bool = response_dict["isMissionRunning"]
         return is_running
+
+    def get_mission_report_ids_and_endtime_for_robot(
+        self, exr_robot_id: str, number_of_latest_reports: str
+    ) -> str:
+        query_string: str = """
+            query missionReports($robot_id: String!, $number_of_latest_reports: Float!) {
+                missionReports(input: {robotId: $robot_id}
+                filter: {first: $number_of_latest_reports}) {   
+                  page {    
+                    edges {
+                      node {
+                        endTimestamp
+                        id
+                      }
+                    }
+                  }
+                }
+            }
+
+        """
+        params: dict = {
+            "robot_id": exr_robot_id,
+            "number_of_latest_reports": number_of_latest_reports,
+        }
+        response_dict: dict = self.client.query(query_string, params)
+        return response_dict["missionReports"]["page"]["edges"]
+
+    def _return_latest_mission_report_id(self, response: dict[str, Any]):
+        return response[0]["node"]["id"]
+
+    def get_last_mission_report_for_robot(self) -> MissionReports:
+        reports = self.get_mission_report_ids_and_endtime_for_robot(
+            settings.ROBOT_EXR_ID, 1
+        )
+        latest_report_id = self._return_latest_mission_report_id(reports)
+
+        query_string: str = """
+            query missionReport($report_id: String!) {
+                missionReport(id: $report_id) {
+                  dataPayloads{
+                    ... on PhotoDataPayloadType {
+                      dataType
+                      uri
+                      poiName
+                    }
+                    ... on VideoDataPayloadType {
+                      dataType
+                      uri
+                      poiName
+                    }
+                    ... on AudioDataPayloadType {
+                      dataType
+                      uri
+                      poiName
+                    }
+                  }
+                }
+            }
+
+        """
+
+        params = {"report_id": latest_report_id}
+        response_dict: dict = self.client.query(query_string, params)
+        mission_report = MissionReports.from_dict(response_dict[""])
+        return response_dict
+
+    @staticmethod
+    def _get_inspection_uri_for_poi(
+        mission_report: dict, point_of_interest: str
+    ) -> str:
+        return next(
+            (
+                inspection
+                for inspection in mission_report["missionReport"]["dataPayloads"]
+                if inspection["poiName"] == point_of_interest
+            ),
+            None,
+        )["uri"]
+
+    def get_inspection_data_for_poi(self, poi_name: str):
+        inspection_report = self.get_last_mission_report_for_robot()
+        inspection_data_uri = self._get_inspection_uri_for_poi(
+            inspection_report, poi_name
+        )
+
+        raise NotImplementedError
+
+    @staticmethod
+    def _get_inspection_data_type(inspection_report) -> Type[InspectionMetadata]:
+        raise NotImplementedError
+
+
+if __name__ == "__main__":
+    api = Api()
+    report = api.get_last_mission_report_for_robot()
+    # print(report)
+    # print(api.get_mission_report_ids_and_endtime_for_robot(settings.ROBOT_EXR_ID, 10))
+    print(api._get_inspection_for_poi(report, "hvac_1"))

src/isar_exr/models/mission.py

@@ -0,0 +1,72 @@
+from enum import Enum
+from pydantic import BaseModel
+from typing import Union, Optional, Any
+from datetime import datetime
+
+
+class MissionReport:
+    # TODO: fill in types?
+    def __init__(
+        self,
+        id=None,
+        key=None,
+        startTimestamp=None,
+        endTimestamp: datetime = None,
+        robot=None,
+        diagnostics: list = None,
+        trajectory: list = None,
+        distance=None,
+        duration=None,
+        missionDefinition: list = None,
+        data_payloads: list = None,
+    ):
+        self.id = id
+        self.key = key
+        self.start_time_stamp = startTimestamp
+        self.end_time_stamp = endTimestamp
+        self.robot = robot
+        self.diagnostics = diagnostics
+        self.trajectory = trajectory
+        self.distance = distance
+        self.duration = duration
+        self.misssionDefinition = missionDefinition
+        self.dataPayloads = data_payloads
+
+
+class MissionReports:
+    reports: list[MissionReport] = []
+
+    def from_dict(self, reports_dict):
+        if not isinstance(reports_dict, list) or len(reports_dict) == 0:
+            raise TypeError("No inspections in report")
+
+        for report_dict in reports_dict:
+            report = MissionReport(**report_dict["node"])
+            self.append_to_report_list(report)
+        return self
+
+    def append_to_report_list(self, report: MissionReport):
+        self.reports.append(report)
+
+
+class AbstractDataPayloadType:
+    id: Union[str, int, float]
+    key: str
+    parentPayloadKey: str
+    dataType: DataPayload
+    producer: Any
+    dataLabel: str
+    regionOfInterest: Any
+    poiName: str
+    isSynced: bool
+    pass
+
+
+class DataPayload(Enum):
+    Photo = "PHOTO"
+    Video = "VIDEO"
+    Audio = "AUDIO"
+    TimeSeries1D = "TIME_SERIES_1D"
+    PhotoOverlay = "PHOTO_OVERLAY"
+    JsonData = "JSON_DATA"
+    Group = "GROUP"

tests/api/test_report.py

@@ -0,0 +1,34 @@
+import json
+
+from isar_exr.api.energy_robotics_api import Api
+from isar_exr.models.mission import MissionReport, MissionReports
+
+
+def test_return_latest_mission_report_id():
+    file = open("tests/data/mission_report_response.json", "r")
+    response_dict = json.load(file)
+    api = Api()
+    latest_report_id = api._return_latest_mission_report_id(
+        response_dict["missionReports"]["page"]["edges"]
+    )
+    assert latest_report_id == "606dd56c023c8866b43f7f7e"
+
+
+def test_return_inspection_uri_for_poi():
+    file = open("tests/data/inspection_report_response.json", "r")
+    response_dict = json.load(file)
+    poi_name = "beds_1"
+    expected_uri = "https://actual.png"
+    actual_uri = Api()._get_inspection_uri_for_poi(response_dict, poi_name)
+    assert expected_uri == actual_uri
+
+
+def test_construct_mission_reports_type_from_json():
+    file = open("tests/data/mission_report_response.json", "r")
+    response_dict = json.load(file)["missionReports"]
+    mission_reports = MissionReports().from_dict(response_dict["page"]["edges"])
+    assert isinstance(mission_reports, MissionReports)
+
+
+if __name__ == "__main__":
+    test_construct_mission_reports_type_from_json()

tests/data/inspection_report_response.json

@@ -0,0 +1,36 @@
+{
+    "missionReport": {
+      "dataPayloads": [
+        {
+          "dataType": "PHOTO",
+          "uri": "https://dummy1.png",
+          "poiName": "stairway_1"
+        },
+        {
+          "dataType": "PHOTO",
+          "uri": "https://dummy2.png",
+          "poiName": "box"
+        },
+        {
+          "dataType": "PHOTO",
+          "uri": "https://dummy3.png",
+          "poiName": "stairway_2"
+        },
+        {
+          "dataType": "PHOTO",
+          "uri": "https://dummy4.png",
+          "poiName": "hvac_1"
+        },
+        {
+          "dataType": "PHOTO",
+          "uri": "https://actual.png",
+          "poiName": "beds_1"
+        },
+        {
+          "dataType": "PHOTO",
+          "uri": "https://dummy6.png",
+          "poiName": "stair_1"
+        }
+      ]
+    }
+  }

tests/data/mission_report_response.json

@@ -0,0 +1,26 @@
+{
+    "missionReports": {
+      "page": {
+        "edges": [
+          {
+            "node": {
+              "endTimestamp": 1712503226920,
+              "id": "606dd56c023c8866b43f7f7e"
+            }
+          },
+          {
+            "node": {
+              "endTimestamp": 1688748047627,
+              "id": "606ef3d8472ff0610e3bce38"
+            }
+          },
+          {
+            "node": {
+              "endTimestamp": 1620565032722,
+              "id": "606ea4984f01b71ea6a28461"
+            }
+          }
+        ]
+      }
+    }
+  }