mirror of
https://github.com/jkingsman/Remote-Terminal-for-MeshCore.git
synced 2026-03-28 17:43:05 +01:00
143 lines
5.0 KiB
Python
143 lines
5.0 KiB
Python
import os
|
|
from typing import Any, Literal
|
|
|
|
from fastapi import APIRouter
|
|
from pydantic import BaseModel
|
|
|
|
from app.config import settings
|
|
from app.repository import RawPacketRepository
|
|
from app.services.radio_runtime import radio_runtime as radio_manager
|
|
from app.version_info import get_app_build_info
|
|
|
|
router = APIRouter(tags=["health"])
|
|
|
|
|
|
class RadioDeviceInfoResponse(BaseModel):
|
|
model: str | None = None
|
|
firmware_build: str | None = None
|
|
firmware_version: str | None = None
|
|
max_contacts: int | None = None
|
|
max_channels: int | None = None
|
|
|
|
|
|
class AppInfoResponse(BaseModel):
|
|
version: str
|
|
commit_hash: str | None = None
|
|
|
|
|
|
class HealthResponse(BaseModel):
|
|
status: str
|
|
radio_connected: bool
|
|
radio_initializing: bool = False
|
|
radio_state: str = "disconnected"
|
|
connection_info: str | None
|
|
app_info: AppInfoResponse | None = None
|
|
radio_device_info: RadioDeviceInfoResponse | None = None
|
|
database_size_mb: float
|
|
oldest_undecrypted_timestamp: int | None
|
|
fanout_statuses: dict[str, dict[str, str]] = {}
|
|
bots_disabled: bool = False
|
|
bots_disabled_source: Literal["env", "until_restart"] | None = None
|
|
basic_auth_enabled: bool = False
|
|
|
|
|
|
def _clean_optional_str(value: object) -> str | None:
|
|
if not isinstance(value, str):
|
|
return None
|
|
cleaned = value.strip()
|
|
return cleaned or None
|
|
|
|
|
|
def _read_optional_bool_setting(name: str) -> bool:
|
|
value = getattr(settings, name, False)
|
|
return value if isinstance(value, bool) else False
|
|
|
|
|
|
async def build_health_data(radio_connected: bool, connection_info: str | None) -> dict:
|
|
"""Build the health status payload used by REST endpoint and WebSocket broadcasts."""
|
|
app_build_info = get_app_build_info()
|
|
db_size_mb = 0.0
|
|
try:
|
|
db_size_bytes = os.path.getsize(settings.database_path)
|
|
db_size_mb = round(db_size_bytes / (1024 * 1024), 2)
|
|
except OSError:
|
|
pass
|
|
|
|
oldest_ts = None
|
|
try:
|
|
oldest_ts = await RawPacketRepository.get_oldest_undecrypted()
|
|
except RuntimeError:
|
|
pass # Database not connected
|
|
|
|
# Fanout module statuses
|
|
fanout_statuses: dict[str, Any] = {}
|
|
bots_disabled_source = "env" if _read_optional_bool_setting("disable_bots") else None
|
|
try:
|
|
from app.fanout.manager import fanout_manager
|
|
|
|
fanout_statuses = fanout_manager.get_statuses()
|
|
manager_bots_disabled_source = fanout_manager.get_bots_disabled_source()
|
|
if manager_bots_disabled_source is not None:
|
|
bots_disabled_source = manager_bots_disabled_source
|
|
except Exception:
|
|
pass
|
|
|
|
setup_in_progress = getattr(radio_manager, "is_setup_in_progress", False)
|
|
setup_complete = getattr(radio_manager, "is_setup_complete", radio_connected)
|
|
if not radio_connected:
|
|
setup_complete = False
|
|
|
|
connection_desired = getattr(radio_manager, "connection_desired", True)
|
|
is_reconnecting = getattr(radio_manager, "is_reconnecting", False)
|
|
|
|
radio_initializing = bool(radio_connected and (setup_in_progress or not setup_complete))
|
|
if not connection_desired:
|
|
radio_state = "paused"
|
|
elif radio_initializing:
|
|
radio_state = "initializing"
|
|
elif radio_connected:
|
|
radio_state = "connected"
|
|
elif is_reconnecting:
|
|
radio_state = "connecting"
|
|
else:
|
|
radio_state = "disconnected"
|
|
|
|
radio_device_info = None
|
|
device_info_loaded = getattr(radio_manager, "device_info_loaded", False)
|
|
if radio_connected and device_info_loaded:
|
|
radio_device_info = {
|
|
"model": _clean_optional_str(getattr(radio_manager, "device_model", None)),
|
|
"firmware_build": _clean_optional_str(getattr(radio_manager, "firmware_build", None)),
|
|
"firmware_version": _clean_optional_str(
|
|
getattr(radio_manager, "firmware_version", None)
|
|
),
|
|
"max_contacts": getattr(radio_manager, "max_contacts", None),
|
|
"max_channels": getattr(radio_manager, "max_channels", None),
|
|
}
|
|
|
|
return {
|
|
"status": "ok" if radio_connected and not radio_initializing else "degraded",
|
|
"radio_connected": radio_connected,
|
|
"radio_initializing": radio_initializing,
|
|
"radio_state": radio_state,
|
|
"connection_info": connection_info,
|
|
"app_info": {
|
|
"version": app_build_info.version,
|
|
"commit_hash": app_build_info.commit_hash,
|
|
},
|
|
"radio_device_info": radio_device_info,
|
|
"database_size_mb": db_size_mb,
|
|
"oldest_undecrypted_timestamp": oldest_ts,
|
|
"fanout_statuses": fanout_statuses,
|
|
"bots_disabled": bots_disabled_source is not None,
|
|
"bots_disabled_source": bots_disabled_source,
|
|
"basic_auth_enabled": _read_optional_bool_setting("basic_auth_enabled"),
|
|
}
|
|
|
|
|
|
@router.get("/health", response_model=HealthResponse)
|
|
async def healthcheck() -> HealthResponse:
|
|
"""Check if the API is running and if the radio is connected."""
|
|
data = await build_health_data(radio_manager.is_connected, radio_manager.connection_info)
|
|
return HealthResponse(**data)
|