core/services/ardupilot_manager/mavlink_proxy: Add TLogCondition

Author: joaoantoniocardoso

core/services/ardupilot_manager/mavlink_proxy/AbstractRouter.py

@@ -4,6 +4,7 @@
 import shlex
 import shutil
 import tempfile
+from enum import Enum
 from typing import Any, List, Optional, Set, Type
 
 from loguru import logger
@@ -18,6 +19,13 @@
 )
 
 
+class TLogCondition(str, Enum):
+    """When to write tlog files."""
+
+    Always = "always"
+    WhileArmed = "while_armed"
+
+
 class AbstractRouter(metaclass=abc.ABCMeta):
     def __init__(self) -> None:
         self._endpoints: Set[Endpoint] = set()
@@ -28,6 +36,7 @@ def __init__(self) -> None:
         # to avoid any problem in __del__
         self._binary = shutil.which(self.binary_name())
         self._logdir = pathlib.Path(tempfile.gettempdir())
+        self._tlog_condition = TLogCondition.WhileArmed
         self._version = self._get_version()
 
     @staticmethod
@@ -182,6 +191,12 @@ def set_logdir(self, directory: pathlib.Path) -> None:
             raise ValueError(f"Logging directory {directory} does not exist.")
         self._logdir = directory
 
+    def tlog_condition(self) -> TLogCondition:
+        return self._tlog_condition
+
+    def set_tlog_condition(self, tlog_condition: TLogCondition) -> None:
+        self._tlog_condition = tlog_condition
+
     def add_endpoint(self, endpoint: Endpoint) -> None:
         self._validate_endpoint(endpoint)
 

core/services/ardupilot_manager/mavlink_proxy/Manager.py

@@ -10,7 +10,7 @@
 import mavlink_proxy.MAVLinkServer
 import mavlink_proxy.MAVP2P
 import mavlink_proxy.MAVProxy
-from mavlink_proxy.AbstractRouter import AbstractRouter
+from mavlink_proxy.AbstractRouter import AbstractRouter, TLogCondition
 from mavlink_proxy.Endpoint import Endpoint
 from mavlink_proxy.exceptions import (
     EndpointAlreadyExists,
@@ -171,6 +171,9 @@ def router_name(self) -> str:
     def set_logdir(self, log_dir: pathlib.Path) -> None:
         self.tool.set_logdir(log_dir)
 
+    def set_tlog_condition(self, tlog_condtion: TLogCondition) -> None:
+        self.tool.set_tlog_condition(tlog_condtion)
+
     async def auto_restart_router(self) -> None:
         """Auto-restart Mavlink router process if it dies."""
         while True: