mirror of
https://git.ianrenton.com/ian/spothole.git
synced 2025-10-27 08:49:27 +00:00
60 lines
2.5 KiB
Python
60 lines
2.5 KiB
Python
import os
|
|
from datetime import datetime
|
|
from threading import Timer
|
|
|
|
import psutil
|
|
import pytz
|
|
|
|
from core.config import SERVER_OWNER_CALLSIGN
|
|
from core.constants import SOFTWARE_VERSION
|
|
|
|
|
|
# Provides a timed update of the application's status data.
|
|
class StatusReporter:
|
|
|
|
# Constructor
|
|
def __init__(self, status_data, run_interval, web_server, cleanup_timer, spots, spot_providers, alerts,
|
|
alert_providers):
|
|
self.status_data = status_data
|
|
self.run_interval = run_interval
|
|
self.web_server = web_server
|
|
self.cleanup_timer = cleanup_timer
|
|
self.spots = spots
|
|
self.spot_providers = spot_providers
|
|
self.alerts = alerts
|
|
self.alert_providers = alert_providers
|
|
self.run_timer = None
|
|
self.startup_time = datetime.now(pytz.UTC)
|
|
|
|
self.status_data["software-version"] = SOFTWARE_VERSION
|
|
self.status_data["server-owner-callsign"] = SERVER_OWNER_CALLSIGN
|
|
|
|
# Start the cleanup timer
|
|
def start(self):
|
|
self.run()
|
|
|
|
# Stop any threads and prepare for application shutdown
|
|
def stop(self):
|
|
self.run_timer.cancel()
|
|
|
|
# Write status information and reschedule next timer
|
|
def run(self):
|
|
self.status_data["uptime"] = str(datetime.now(pytz.UTC) - self.startup_time).split(".")[0]
|
|
self.status_data["mem_use_mb"] = round(psutil.Process(os.getpid()).memory_info().rss / (1024 * 1024), 3)
|
|
self.status_data["num_spots"] = len(self.spots)
|
|
self.status_data["num_alerts"] = len(self.alerts)
|
|
self.status_data["spot_providers"] = list(
|
|
map(lambda p: {"name": p.name, "enabled": p.enabled, "status": p.status, "last_updated": p.last_update_time,
|
|
"last_spot": p.last_spot_time}, self.spot_providers))
|
|
self.status_data["alert_providers"] = list(
|
|
map(lambda p: {"name": p.name, "enabled": p.enabled, "status": p.status,
|
|
"last_updated": p.last_update_time}, self.alert_providers))
|
|
self.status_data["cleanup"] = {"status": self.cleanup_timer.status,
|
|
"last_ran": self.cleanup_timer.last_cleanup_time}
|
|
self.status_data["webserver"] = {"status": self.web_server.status,
|
|
"last_api_access": self.web_server.last_api_access_time,
|
|
"last_page_access": self.web_server.last_page_access_time}
|
|
|
|
self.run_timer = Timer(self.run_interval, self.run)
|
|
self.run_timer.start()
|