Export state as native metric for state timeline grafana

Author: olaals

src/isar/state_machine/state_machine.py

@@ -14,6 +14,7 @@
 )
 from isar.services.utilities.mqtt_utilities import publish_isar_status
 from isar.state_machine.state import State, Transition
+from isar.state_machine.state_metrics import StateMetricsPublisher
 from isar.state_machine.states.going_to_lockdown import GoingToLockdown
 from isar.state_machine.states.maintenance import Maintenance
 from isar.state_machine.states.unknown_status import UnknownStatus
@@ -71,6 +72,10 @@ def __init__(
             [], settings.STATE_TRANSITIONS_LOG_LENGTH
         )
 
+        self.state_metrics_publisher: StateMetricsPublisher = StateMetricsPublisher(
+            current_state_provider=lambda: self.current_state.name
+        )
+
     #################################################################################
 
     def run(self) -> None:

src/isar/state_machine/state_metrics.py

@@ -0,0 +1,34 @@
+from typing import Callable, List
+
+from opentelemetry import metrics
+from opentelemetry.metrics import CallbackOptions, Meter, Observation
+
+from isar.config.settings import settings
+from isar.state_machine.states_enum import STATE_TO_CODE, States
+
+UNKNOWN_STATE_CODE: int = -1
+
+
+class StateMetricsPublisher:
+    def __init__(self, current_state_provider: Callable[[], States]) -> None:
+        self._current_state_provider = current_state_provider
+
+        self.meter: Meter = metrics.get_meter("isar.state_machine")
+        self.meter.create_observable_gauge(
+            name="isar.state",
+            callbacks=[self._observe_state],
+            description="Current state of the ISAR state machine (see STATE_TO_CODE)",
+        )
+
+    def _observe_state(self, _: CallbackOptions) -> List[Observation]:
+        state: States = self._current_state_provider()
+        code: int = STATE_TO_CODE.get(state, UNKNOWN_STATE_CODE)
+        return [
+            Observation(
+                value=code,
+                attributes={
+                    "robot_name": settings.ROBOT_NAME,
+                    "isar_id": settings.ISAR_ID,
+                },
+            )
+        ]

src/isar/state_machine/states_enum.py

@@ -33,3 +33,35 @@ class States(str, Enum):
 
     def __repr__(self) -> str:
         return self.value
+
+
+STATE_TO_CODE: dict["States", int] = {
+    States.Home: 0,
+    States.AwaitNextMission: 1,
+    States.Monitor: 2,
+    States.ReturningHome: 3,
+    States.Offline: 4,
+    States.InterventionNeeded: 5,
+    States.Maintenance: 6,
+    States.Paused: 7,
+    States.Pausing: 8,
+    States.Resuming: 9,
+    States.PausingReturnHome: 10,
+    States.ResumingReturnHome: 11,
+    States.ReturnHomePaused: 12,
+    States.Stopping: 13,
+    States.StoppingUnknownMission: 14,
+    States.StoppingReturnHome: 15,
+    States.StoppingPausedMission: 16,
+    States.StoppingPausedReturnHome: 17,
+    States.StoppingDueToMaintenance: 18,
+    States.StoppingGoToLockdown: 19,
+    States.StoppingGoToRecharge: 20,
+    States.Recharging: 21,
+    States.RechargingWithMission: 22,
+    States.GoingToRecharging: 23,
+    States.GoingToRechargingWithMission: 24,
+    States.GoingToLockdown: 25,
+    States.Lockdown: 26,
+    States.UnknownStatus: 27,
+}