Enhance collision detection and intervention triggers in AutonomyMetricsLogger

src/autonomy_metrics/autonomy_metrics/db_mgr.py

@@ -1,117 +1,161 @@
-from pymongo import MongoClient
+from __future__ import annotations
+
 from datetime import datetime, timezone
+import math
+
 from bson import ObjectId
+from pymongo import DESCENDING, MongoClient
+
 
-# NEW
 try:
     import numpy as np
-except Exception:
+except Exception:  # pragma: no cover
     np = None
 
 
 class DatabaseMgr:
-    """
-    A class to manage database operations for robot session.
-    """
-
-    def __init__(self, database_name='robot_incidents', host='localhost', port=27017):
+    """MongoDB helper for session summaries, events, and periodic snapshots."""
+
+    def __init__(
+        self,
+        database_name: str = "robot_incidents",
+        host: str = "localhost",
+        port: int = 27017,
+    ):
         self.client = MongoClient(
-            f'mongodb://{host}:{port}/',
-            connectTimeoutMS=None
+            f"mongodb://{host}:{port}/",
+            connectTimeoutMS=1500,
+            serverSelectionTimeoutMS=1500,
         )
         self.db = self.client[database_name]
-        self.sessions_collection = self.db['sessions']
+        self.sessions_collection = self.db["sessions"]
+        self.events_collection = self.db["session_events"]
+        self.snapshots_collection = self.db["session_snapshots"]
         self.session_id = None
 
-    # NEW
+        self.sessions_collection.create_index([("session_start_time", DESCENDING)])
+        self.events_collection.create_index([("session_id", 1), ("time", DESCENDING)])
+        self.snapshots_collection.create_index([("session_id", 1), ("time", DESCENDING)])
+
     def _bson_safe(self, obj):
-        """Recursively convert non-BSON types (notably numpy) into Mongo-friendly types."""
         if np is not None:
-            # numpy arrays -> list
             if isinstance(obj, np.ndarray):
                 return obj.tolist()
-            # numpy scalar -> python scalar
             if isinstance(obj, np.generic):
                 return obj.item()
 
+        if isinstance(obj, float):
+            if math.isfinite(obj):
+                return obj
+            return None
+
         if isinstance(obj, dict):
-            return {k: self._bson_safe(v) for k, v in obj.items()}
+            return {key: self._bson_safe(value) for key, value in obj.items()}
 
         if isinstance(obj, (list, tuple)):
-            return [self._bson_safe(v) for v in obj]
+            return [self._bson_safe(value) for value in obj]
 
         return obj
 
-    def init_session(self, env_variables, aoc_repos_info):
+    def _json_safe(self, obj):
+        if isinstance(obj, ObjectId):
+            return str(obj)
+        if isinstance(obj, datetime):
+            return obj.astimezone(timezone.utc).isoformat()
+        if isinstance(obj, dict):
+            return {key: self._json_safe(value) for key, value in obj.items()}
+        if isinstance(obj, list):
+            return [self._json_safe(value) for value in obj]
+        return obj
+
+    def init_session(self, env_variables: dict, git_repos_info: list[dict], metadata: dict):
         session_start = datetime.now(tz=timezone.utc)
-        session_document = {
-            "session_start_time": session_start,
-            "robot_name": env_variables['robot_name'],
-            "farm_name": env_variables['farm_name'],
-            "field_name": env_variables['field_name'],
-            "application": env_variables['application'],
-            "scenario_name": env_variables['scenario_name'],
-            "aoc_repos_info": aoc_repos_info,
-            "mdbi": None,
-            "incidents": 0,
-            "distance": 0,
-            "autonomous_distance": 0,
-            "collision_incidents": 0,
-            "events": []
-        }
-
-        # NEW: sanitize (in case aoc_repos_info contains numpy types)
-        session_document = self._bson_safe(session_document)
+        session_document = self._bson_safe(
+            {
+                "session_start_time": session_start,
+                "session_end_time": None,
+                "robot_name": env_variables.get("robot_name", "UNDEFINED"),
+                "farm_name": env_variables.get("farm_name", "UNDEFINED"),
+                "field_name": env_variables.get("field_name", "UNDEFINED"),
+                "application": env_variables.get("application", "UNDEFINED"),
+                "scenario_name": env_variables.get("scenario_name", "UNDEFINED"),
+                "aoc_repos_info": git_repos_info,
+                "summary": metadata,
+            }
+        )
 
         result = self.sessions_collection.insert_one(session_document)
         self.session_id = result.inserted_id
+        return str(self.session_id)
 
-    def add_event(self, event):
-        # NEW: sanitize event before writing
-        event = self._bson_safe(event)
-
+    def update_session_summary(self, summary: dict):
+        if self.session_id is None:
+            return False
+        summary = self._bson_safe(summary)
         result = self.sessions_collection.update_one(
-            {"_id": ObjectId(self.session_id)},
-            {"$addToSet": {"events": event}}
+            {"_id": self.session_id},
+            {"$set": {"summary": summary, "last_updated_time": datetime.now(tz=timezone.utc)}},
         )
-        return result.modified_count > 0
-
-    def update_incidents(self, incidents):
-        incidents = self._bson_safe(incidents)  # harmless; keeps pattern consistent
+        return result.acknowledged
+
+    def mark_session_end(self, summary: dict | None = None):
+        if self.session_id is None:
+            return False
+        update = {"session_end_time": datetime.now(tz=timezone.utc)}
+        if summary is not None:
+            update["summary"] = self._bson_safe(summary)
         result = self.sessions_collection.update_one(
-            {"_id": ObjectId(self.session_id)},
-            {"$set": {"incidents": incidents}}
+            {"_id": self.session_id},
+            {"$set": update},
         )
-        return result.modified_count > 0
-
-    def update_distance(self, distance):
-        distance = self._bson_safe(distance)
-        result = self.sessions_collection.update_one(
-            {"_id": ObjectId(self.session_id)},
-            {"$set": {"distance": distance}}
+        return result.acknowledged
+
+    def add_event(self, event: dict):
+        if self.session_id is None:
+            return False
+        document = self._bson_safe(
+            {
+                "session_id": self.session_id,
+                **event,
+            }
         )
-        return result.modified_count > 0
-
-    def update_autonomous_distance(self, autonomous_distance):
-        autonomous_distance = self._bson_safe(autonomous_distance)
-        result = self.sessions_collection.update_one(
-            {"_id": ObjectId(self.session_id)},
-            {"$set": {"autonomous_distance": autonomous_distance}}
+        result = self.events_collection.insert_one(document)
+        return bool(result.inserted_id)
+
+    def add_snapshot(self, snapshot: dict):
+        if self.session_id is None:
+            return False
+        document = self._bson_safe(
+            {
+                "session_id": self.session_id,
+                **snapshot,
+            }
         )
-        return result.modified_count > 0
+        result = self.snapshots_collection.insert_one(document)
+        return bool(result.inserted_id)
 
-    def update_mdbi(self, mdbi):
-        mdbi = self._bson_safe(mdbi)
-        result = self.sessions_collection.update_one(
-            {"_id": ObjectId(self.session_id)},
-            {"$set": {"mdbi": mdbi}}
+    def fetch_latest_session(self) -> dict | None:
+        document = self.sessions_collection.find_one(
+            sort=[("session_start_time", DESCENDING)]
         )
-        return result.modified_count > 0
-
-    def update_collision_incidents(self, collision_incidents):
-        collision_incidents = self._bson_safe(collision_incidents)
-        result = self.sessions_collection.update_one(
-            {"_id": ObjectId(self.session_id)},
-            {"$set": {"collision_incidents": collision_incidents}}
+        return self._json_safe(document) if document else None
+
+    def fetch_recent_events(self, limit: int = 20, session_id: str | None = None) -> list[dict]:
+        query = {}
+        if session_id:
+            query["session_id"] = ObjectId(session_id)
+        documents = list(
+            self.events_collection.find(query).sort("time", DESCENDING).limit(limit)
+        )
+        return self._json_safe(documents)
+
+    def fetch_recent_snapshots(
+        self, limit: int = 10, session_id: str | None = None
+    ) -> list[dict]:
+        query = {}
+        if session_id:
+            query["session_id"] = ObjectId(session_id)
+        documents = list(
+            self.snapshots_collection.find(query).sort("time", DESCENDING).limit(limit)
         )
-        return result.modified_count > 0
+        return self._json_safe(documents)

src/autonomy_metrics/autonomy_metrics/git_utils.py

@@ -0,0 +1,89 @@
+"""Helpers for collecting git metadata for the configured repositories."""
+
+from __future__ import annotations
+
+import os
+import subprocess
+
+
+DEFAULT_GIT_REPOS_ROOT = (
+    "/home/ros/aoc_strawberry_scenario_ws/src/aoc_strawberry_scenario"
+)
+
+
+def _run_git(repo_path: str, *args: str) -> str:
+    result = subprocess.run(
+        ["git", "-C", repo_path, *args],
+        capture_output=True,
+        text=True,
+        timeout=5.0,
+        check=False,
+    )
+    if result.returncode != 0:
+        raise RuntimeError(result.stderr.strip() or "git command failed")
+    return result.stdout.strip()
+
+
+def get_git_info(repo_path: str) -> dict:
+    """Return a git metadata snapshot for one repository path."""
+    info = {
+        "path": repo_path,
+        "exists": False,
+        "remote": None,
+        "branch": None,
+        "commit": None,
+        "short_commit": None,
+        "commit_message": None,
+        "tags": [],
+        "describe": None,
+        "dirty": None,
+        "error": None,
+    }
+
+    try:
+        if not repo_path or not os.path.isdir(repo_path):
+            info["error"] = "Path does not exist or is not a directory"
+            return info
+
+        info["exists"] = True
+
+        for key, command in (
+            ("remote", ("config", "--get", "remote.origin.url")),
+            ("branch", ("rev-parse", "--abbrev-ref", "HEAD")),
+            ("commit", ("rev-parse", "HEAD")),
+            ("short_commit", ("rev-parse", "--short", "HEAD")),
+            ("commit_message", ("log", "-1", "--pretty=%s")),
+            ("describe", ("describe", "--tags", "--always")),
+        ):
+            try:
+                info[key] = _run_git(repo_path, *command)
+            except Exception:
+                info[key] = None
+
+        try:
+            tags = _run_git(repo_path, "tag", "--points-at", "HEAD")
+            info["tags"] = tags.splitlines() if tags else []
+        except Exception:
+            info["tags"] = []
+
+        try:
+            info["dirty"] = bool(_run_git(repo_path, "status", "--porcelain"))
+        except Exception:
+            info["dirty"] = None
+
+    except Exception as exc:
+        info["error"] = str(exc)
+
+    return info
+
+
+def collect_git_repos_info(config: dict) -> list[dict]:
+    """Collect git metadata for all configured repositories."""
+    repos_root = config.get("git_repos_base_path", DEFAULT_GIT_REPOS_ROOT)
+    repos_config = config.get("git_repos", {}) or {}
+
+    results = []
+    for label, rel_path in repos_config.items():
+        repo_path = os.path.join(repos_root, rel_path)
+        results.append({label: get_git_info(repo_path)})
+    return results