Allow status directory to be specified, to make ECS healthchecks easier

Dockerfile

@@ -22,6 +22,6 @@ RUN --mount=type=cache,target=/root/.cache/pip \
 USER appuser
 COPY . .
 
-HEALTHCHECK --interval=60s --timeout=10s --start-period=30s --retries=3 CMD python3 check-mate --status
+HEALTHCHECK --interval=60s --timeout=10s --start-period=30s --retries=3 CMD python3 -m check-mate --status
 
 ENTRYPOINT python3 -m check-mate 

check-mate.py

@@ -13,7 +13,7 @@
 import meshtastic
 import meshtastic.tcp_interface
 
-from status import readStatus, writeStatus
+from status import StatusManager
 import json
 
 """Max frequency with which to report healthchecks."""
@@ -26,14 +26,16 @@
 class CheckMate:
     """Manages connection with meshtastic node, monitoring private channels and responds to radio checks"""
 
-    def __init__(self, host, location=None, healthCheckURL=None):
+    def __init__(self, statusManager, host, location, healthCheckURL):
+        self.statusManager = statusManager
         self.host = host
         self.location = location
+        self.healthCheckURL = healthCheckURL
+        self.lastHealthCheck = None
+
         self.users = {}
         self.iface = None
         self.connected = False
-        self.lastHealthCheck = None
-        self.healthCheckURL = healthCheckURL
         self.logger = logging.getLogger(__name__)
         self.status = {
             "status": "starting",
@@ -86,7 +88,7 @@ def setStatus(self, status, ping=False):
         self.status["user_count"] = len(self.users)
         if ping:
             self.status["last_device_ping"] = time.time()
-        writeStatus(self.status)
+        self.statusManager.writeStatus(self.status)
 
     def onConnect(self, interface, topic=pub.AUTO_TOPIC):
         """called when we (re)connect to the radio"""
@@ -302,6 +304,13 @@ def getLogFormat():
         required=False,
         help="Get status of the current check-mate process",
     )
+    parser.add_argument(
+        "--status-dir",
+        dest="statusDir",
+        required=False,
+        help="Override default location of the status dir",
+        default=os.environ.get("STATUS_DIR"),
+    )
     parser.add_argument(
         "--host",
         dest="host",
@@ -326,18 +335,15 @@ def getLogFormat():
     )
     args = parser.parse_args()
 
+    statusManager = StatusManager(args.statusDir)
+
     if args.status:
-        status = readStatus()
-        print(json.dumps(status))
-        if status["status"] == "active":
-            sys.exit(0)
-        else:
-            sys.exit(1)
+        sys.exit(statusManager.dump())
 
     if not args.host:
         parser.error(
             "Please provide a host via --host or the $HOST environment variable"
         )
 
-    checkmate = CheckMate(args.host, args.location, args.healthCheckURL)
+    checkmate = CheckMate(statusManager, args.host, args.location, args.healthCheckURL)
     sys.exit(checkmate.start())

status.py

@@ -4,29 +4,35 @@
 import json
 
 
-def getStatusFilePath() -> Path:
-    """Determine the appropriate status file path based on the OS."""
-    if platform.system() == "Darwin":  # macOS
-        base_dir = Path.home() / "Library" / "Application Support" / "check-mate"
-    else:  # Linux and others
-        base_dir = Path.home() / ".local" / "share" / "check-mate"
-
-    base_dir.mkdir(parents=True, exist_ok=True)
-    return base_dir / "status.json"
-
-
-STATUS_FILE = getStatusFilePath()
-
-
-def writeStatus(status: Dict[str, any]):
-    """Write the current status to the status file."""
-    with open(STATUS_FILE, "w") as f:
-        json.dump(status, f)
-
-
-def readStatus() -> Dict[str, any]:
-    """Read the current status from the status file."""
-    if not STATUS_FILE.exists():
-        return {"status": "unknown"}
-    with open(STATUS_FILE, "r") as f:
-        return json.load(f)
+class StatusManager:
+    def __init__(self, baseDir=None):
+        if baseDir is None or baseDir == "":
+            if platform.system() == "Darwin":  # macOS
+                self.baseDir = (
+                    Path.home() / "Library" / "Application Support" / "check-mate"
+                )
+            else:  # Linux and others
+                self.baseDir = Path.home() / ".local" / "share" / "check-mate"
+        else:
+            self.baseDir = Path(baseDir)
+
+        self.baseDir.mkdir(parents=True, exist_ok=True)
+        self.statusFile = self.baseDir / "status.json"
+
+    def writeStatus(self, status: Dict[str, any]):
+        """Write the current status to the status file."""
+        with open(self.statusFile, "w") as f:
+            json.dump(status, f)
+
+    def readStatus(self) -> Dict[str, any]:
+        """Read the current status from the status file."""
+        if not self.statusFile.exists():
+            return {"status": "unknown"}
+        with open(self.statusFile, "r") as f:
+            return json.load(f)
+
+    def dump(self):
+        """Print the status file."""
+        status = self.readStatus()
+        print(json.dumps(status))
+        return 0 if status["status"] == "active" else 1