diff --git a/Dockerfile b/Dockerfile index bbd73f5..a8c1c32 100644 --- a/Dockerfile +++ b/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 diff --git a/check-mate.py b/check-mate.py index 9571534..3c75fc3 100644 --- a/check-mate.py +++ b/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()) diff --git a/status.py b/status.py index 3064a81..68f88cf 100644 --- a/status.py +++ b/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