mirror of
https://github.com/yellowcooln/meshcore-mqtt-live-map.git
synced 2026-04-20 23:23:36 +00:00
312 lines
12 KiB
Python
312 lines
12 KiB
Python
import os
|
|
|
|
# =========================
|
|
# Env / Config
|
|
# =========================
|
|
MQTT_HOST = os.getenv("MQTT_HOST", "localhost")
|
|
MQTT_PORT = int(os.getenv("MQTT_PORT", "1883"))
|
|
MQTT_USERNAME = os.getenv("MQTT_USERNAME", "")
|
|
MQTT_PASSWORD = os.getenv("MQTT_PASSWORD", "")
|
|
|
|
MQTT_TOPIC = os.getenv("MQTT_TOPIC", "meshcore/#")
|
|
MQTT_TOPICS = [t.strip() for t in MQTT_TOPIC.split(",") if t.strip()]
|
|
|
|
MQTT_TLS = os.getenv("MQTT_TLS", "false").lower() == "true"
|
|
MQTT_TLS_INSECURE = os.getenv("MQTT_TLS_INSECURE", "false").lower() == "true"
|
|
MQTT_CA_CERT = os.getenv("MQTT_CA_CERT", "") # optional path to CA bundle
|
|
|
|
MQTT_TRANSPORT = os.getenv("MQTT_TRANSPORT",
|
|
"tcp").strip().lower() # tcp | websockets
|
|
MQTT_WS_PATH = os.getenv("MQTT_WS_PATH", "/mqtt") # often "/" or "/mqtt"
|
|
|
|
MQTT_CLIENT_ID = os.getenv("MQTT_CLIENT_ID", "")
|
|
|
|
STATE_DIR = os.getenv("STATE_DIR", "/data")
|
|
STATE_FILE = os.getenv("STATE_FILE", os.path.join(STATE_DIR, "state.json"))
|
|
DEVICE_ROLES_FILE = os.getenv("DEVICE_ROLES_FILE",
|
|
os.path.join(STATE_DIR, "device_roles.json"))
|
|
DEVICE_COORDS_FILE = os.getenv("DEVICE_COORDS_FILE",
|
|
os.path.join(STATE_DIR, "device_coords.json"))
|
|
NEIGHBOR_OVERRIDES_FILE = os.getenv(
|
|
"NEIGHBOR_OVERRIDES_FILE",
|
|
os.path.join(STATE_DIR, "neighbor_overrides.json"),
|
|
)
|
|
CHANNEL_SECRETS_FILE = os.getenv(
|
|
"CHANNEL_SECRETS_FILE",
|
|
os.path.join(STATE_DIR, "channel_secrets.json"),
|
|
)
|
|
BACKUP_ENABLED = os.getenv("BACKUP_ENABLED", "false").lower() == "true"
|
|
try:
|
|
BACKUP_INTERVAL_SECONDS = int(os.getenv("BACKUP_INTERVAL_SECONDS", "43200"))
|
|
except ValueError:
|
|
BACKUP_INTERVAL_SECONDS = 43200
|
|
if BACKUP_INTERVAL_SECONDS < 60:
|
|
BACKUP_INTERVAL_SECONDS = 60
|
|
BACKUP_DIR = os.getenv("BACKUP_DIR", "/backup").strip()
|
|
try:
|
|
BACKUP_RETENTION_DAYS = int(os.getenv("BACKUP_RETENTION_DAYS", "7"))
|
|
except ValueError:
|
|
BACKUP_RETENTION_DAYS = 7
|
|
if BACKUP_RETENTION_DAYS < 0:
|
|
BACKUP_RETENTION_DAYS = 0
|
|
STATE_SAVE_INTERVAL = float(os.getenv("STATE_SAVE_INTERVAL", "5"))
|
|
|
|
DEVICE_TTL_HOURS = float(os.getenv("DEVICE_TTL_HOURS", "96")) # 4 days default
|
|
DEVICE_TTL_WINDOW_SECONDS = int(DEVICE_TTL_HOURS * 3600)
|
|
PATH_TTL_SECONDS = int(os.getenv("PATH_TTL_SECONDS", "172800")) # 48 hours
|
|
TRAIL_LEN = int(os.getenv("TRAIL_LEN", "30"))
|
|
ROUTE_TTL_SECONDS = int(os.getenv("ROUTE_TTL_SECONDS", "120"))
|
|
ROUTE_PAYLOAD_TYPES = os.getenv("ROUTE_PAYLOAD_TYPES", "8,9,2,5,4")
|
|
ROUTE_PATH_MAX_LEN = int(os.getenv("ROUTE_PATH_MAX_LEN", "16"))
|
|
ROUTE_MAX_HOP_DISTANCE = float(os.getenv("ROUTE_MAX_HOP_DISTANCE", "100"))
|
|
ROUTE_INFRA_ONLY = os.getenv("ROUTE_INFRA_ONLY", "false").lower() == "true"
|
|
ROUTE_ALLOW_AMBIGUOUS_ONE_BYTE_FALLBACK = (
|
|
os.getenv("ROUTE_ALLOW_AMBIGUOUS_ONE_BYTE_FALLBACK", "false").lower() ==
|
|
"true"
|
|
)
|
|
ROUTE_HISTORY_ENABLED = os.getenv("ROUTE_HISTORY_ENABLED",
|
|
"true").lower() == "true"
|
|
ROUTE_HISTORY_HOURS = float(os.getenv("ROUTE_HISTORY_HOURS", "24"))
|
|
ROUTE_HISTORY_MAX_SEGMENTS = int(
|
|
os.getenv("ROUTE_HISTORY_MAX_SEGMENTS", "40000")
|
|
)
|
|
ROUTE_HISTORY_FILE = os.getenv(
|
|
"ROUTE_HISTORY_FILE", os.path.join(STATE_DIR, "route_history.jsonl")
|
|
)
|
|
ROUTE_HISTORY_PAYLOAD_TYPES = os.getenv(
|
|
"ROUTE_HISTORY_PAYLOAD_TYPES", ROUTE_PAYLOAD_TYPES
|
|
)
|
|
ROUTE_HISTORY_ALLOWED_MODES = os.getenv("ROUTE_HISTORY_ALLOWED_MODES", "path")
|
|
ROUTE_HISTORY_COMPACT_INTERVAL = float(
|
|
os.getenv("ROUTE_HISTORY_COMPACT_INTERVAL", "120")
|
|
)
|
|
HISTORY_EDGE_SAMPLE_LIMIT = 3
|
|
MESSAGE_ORIGIN_TTL_SECONDS = int(os.getenv("MESSAGE_ORIGIN_TTL_SECONDS", "300"))
|
|
HEAT_TTL_SECONDS = int(os.getenv("HEAT_TTL_SECONDS", "600"))
|
|
MQTT_ONLINE_SECONDS = int(os.getenv("MQTT_ONLINE_SECONDS", "300"))
|
|
MQTT_ONLINE_STATUS_TTL_SECONDS = int(
|
|
os.getenv("MQTT_ONLINE_STATUS_TTL_SECONDS", str(MQTT_ONLINE_SECONDS))
|
|
)
|
|
MQTT_ONLINE_INTERNAL_TTL_SECONDS = int(
|
|
os.getenv("MQTT_ONLINE_INTERNAL_TTL_SECONDS", str(MQTT_ONLINE_SECONDS))
|
|
)
|
|
MQTT_ACTIVITY_PACKETS_TTL_SECONDS = int(
|
|
os.getenv("MQTT_ACTIVITY_PACKETS_TTL_SECONDS", str(MQTT_ONLINE_SECONDS))
|
|
)
|
|
MQTT_SEEN_BROADCAST_MIN_SECONDS = float(
|
|
os.getenv("MQTT_SEEN_BROADCAST_MIN_SECONDS", "5")
|
|
)
|
|
MQTT_ONLINE_TOPIC_SUFFIXES = tuple(
|
|
s.strip()
|
|
for s in os.getenv("MQTT_ONLINE_TOPIC_SUFFIXES", "/status,/internal"
|
|
).split(",") if s.strip()
|
|
)
|
|
MQTT_ONLINE_FORCE_NAMES = tuple(
|
|
s.strip()
|
|
for s in os.getenv("MQTT_ONLINE_FORCE_NAMES", "").split(",") if s.strip()
|
|
)
|
|
MQTT_ONLINE_FORCE_NAMES_SET = {s.lower() for s in MQTT_ONLINE_FORCE_NAMES}
|
|
MQTT_STATUS_OFFLINE_VALUES = tuple(
|
|
s.strip().lower()
|
|
for s in os.getenv(
|
|
"MQTT_STATUS_OFFLINE_VALUES", "offline,disconnected"
|
|
).split(",") if s.strip()
|
|
)
|
|
MQTT_STATUS_OFFLINE_VALUES_SET = set(MQTT_STATUS_OFFLINE_VALUES)
|
|
try:
|
|
PEERS_DEFAULT_LIMIT = int(os.getenv("PEERS_DEFAULT_LIMIT", "8"))
|
|
except ValueError:
|
|
PEERS_DEFAULT_LIMIT = 8
|
|
if PEERS_DEFAULT_LIMIT < 1:
|
|
PEERS_DEFAULT_LIMIT = 1
|
|
|
|
DEBUG_PAYLOAD = os.getenv("DEBUG_PAYLOAD", "false").lower() == "true"
|
|
DEBUG_PAYLOAD_MAX = int(os.getenv("DEBUG_PAYLOAD_MAX", "400"))
|
|
|
|
DECODE_WITH_NODE = os.getenv("DECODE_WITH_NODE", "true").lower() == "true"
|
|
NODE_DECODE_TIMEOUT_SECONDS = float(
|
|
os.getenv("NODE_DECODE_TIMEOUT_SECONDS", "2.0")
|
|
)
|
|
DEBUG_LAST_MAX = int(os.getenv("DEBUG_LAST_MAX", "50"))
|
|
DEBUG_STATUS_MAX = int(os.getenv("DEBUG_STATUS_MAX", "50"))
|
|
PAYLOAD_PREVIEW_MAX = int(os.getenv("PAYLOAD_PREVIEW_MAX", "800"))
|
|
DIRECT_COORDS_MODE = os.getenv("DIRECT_COORDS_MODE", "topic").strip().lower()
|
|
DIRECT_COORDS_TOPIC_REGEX = os.getenv(
|
|
"DIRECT_COORDS_TOPIC_REGEX", r"(position|location|gps|coords)"
|
|
)
|
|
DIRECT_COORDS_ALLOW_ZERO = (
|
|
os.getenv("DIRECT_COORDS_ALLOW_ZERO", "false").lower() == "true"
|
|
)
|
|
|
|
ROUTE_HISTORY_ALLOWED_MODES_SET = {
|
|
s.strip()
|
|
for s in ROUTE_HISTORY_ALLOWED_MODES.split(",") if s.strip()
|
|
}
|
|
|
|
SITE_TITLE = os.getenv("SITE_TITLE", "Greater Boston Mesh Live Map")
|
|
SITE_DESCRIPTION = os.getenv(
|
|
"SITE_DESCRIPTION",
|
|
"Live view of Greater Boston Mesh nodes, message routes, and advert paths.",
|
|
)
|
|
SITE_OG_IMAGE = os.getenv("SITE_OG_IMAGE", "")
|
|
SITE_URL = os.getenv("SITE_URL", "/")
|
|
SITE_ICON = os.getenv("SITE_ICON", "/static/logo.png")
|
|
SITE_FEED_NOTE = os.getenv("SITE_FEED_NOTE", "Feed: Boston MQTT.")
|
|
CUSTOM_LINK_URL = os.getenv("CUSTOM_LINK_URL", "").strip()
|
|
PACKET_ANALYZER_URL = os.getenv("PACKET_ANALYZER_URL", "").strip()
|
|
QR_CODE_BUTTON_ENABLED = (
|
|
os.getenv("QR_CODE_BUTTON_ENABLED", "false").lower() == "true"
|
|
)
|
|
GIT_CHECK_ENABLED = os.getenv("GIT_CHECK_ENABLED", "false").lower() == "true"
|
|
GIT_CHECK_FETCH = os.getenv("GIT_CHECK_FETCH", "false").lower() == "true"
|
|
GIT_CHECK_PATH = os.getenv("GIT_CHECK_PATH", os.getcwd()).strip()
|
|
try:
|
|
GIT_CHECK_INTERVAL_SECONDS = float(
|
|
os.getenv("GIT_CHECK_INTERVAL_SECONDS", "43200")
|
|
)
|
|
except ValueError:
|
|
GIT_CHECK_INTERVAL_SECONDS = 43200.0
|
|
DISTANCE_UNITS = os.getenv("DISTANCE_UNITS", "km").strip().lower()
|
|
if DISTANCE_UNITS not in ("km", "mi"):
|
|
DISTANCE_UNITS = "km"
|
|
try:
|
|
NODE_MARKER_RADIUS = float(os.getenv("NODE_MARKER_RADIUS", "8"))
|
|
except ValueError:
|
|
NODE_MARKER_RADIUS = 8.0
|
|
if NODE_MARKER_RADIUS <= 0:
|
|
NODE_MARKER_RADIUS = 8.0
|
|
try:
|
|
HISTORY_LINK_SCALE = float(os.getenv("HISTORY_LINK_SCALE", "1"))
|
|
except ValueError:
|
|
HISTORY_LINK_SCALE = 1.0
|
|
if HISTORY_LINK_SCALE <= 0:
|
|
HISTORY_LINK_SCALE = 1.0
|
|
try:
|
|
MAP_START_LAT = float(os.getenv("MAP_START_LAT", "42.3601"))
|
|
except ValueError:
|
|
MAP_START_LAT = 42.3601
|
|
try:
|
|
MAP_START_LON = float(os.getenv("MAP_START_LON", "-71.1500"))
|
|
except ValueError:
|
|
MAP_START_LON = -71.1500
|
|
try:
|
|
MAP_START_ZOOM = float(os.getenv("MAP_START_ZOOM", "10"))
|
|
except ValueError:
|
|
MAP_START_ZOOM = 10
|
|
MAP_DEFAULT_LAYER = os.getenv("MAP_DEFAULT_LAYER", "light").strip().lower()
|
|
try:
|
|
MAP_RADIUS_KM = float(os.getenv("MAP_RADIUS_KM", "0"))
|
|
except ValueError:
|
|
MAP_RADIUS_KM = 0.0
|
|
if MAP_RADIUS_KM < 0:
|
|
MAP_RADIUS_KM = 0.0
|
|
MAP_RADIUS_SHOW = os.getenv("MAP_RADIUS_SHOW", "false").lower() == "true"
|
|
MAP_BOUNDARY_MODE = os.getenv("MAP_BOUNDARY_MODE", "radius").strip().lower()
|
|
if MAP_BOUNDARY_MODE not in ("radius", "polygon"):
|
|
MAP_BOUNDARY_MODE = "radius"
|
|
MAP_BOUNDARY_FILE = os.getenv(
|
|
"MAP_BOUNDARY_FILE", os.path.join(STATE_DIR, "map_boundary.json")
|
|
).strip()
|
|
MAP_BOUNDARY_SHOW = os.getenv("MAP_BOUNDARY_SHOW", "false").lower() == "true"
|
|
|
|
PROD_MODE = os.getenv("PROD_MODE", "false").lower() == "true"
|
|
PROD_TOKEN = os.getenv("PROD_TOKEN", "").strip()
|
|
|
|
LOS_ELEVATION_URL = os.getenv(
|
|
"LOS_ELEVATION_URL", "https://api.opentopodata.org/v1/srtm90m"
|
|
)
|
|
LOS_ELEVATION_PROXY_URL = os.getenv(
|
|
"LOS_ELEVATION_PROXY_URL", "/los/elevations"
|
|
).strip()
|
|
LOS_SAMPLE_MIN = int(os.getenv("LOS_SAMPLE_MIN", "10"))
|
|
LOS_SAMPLE_MAX = int(os.getenv("LOS_SAMPLE_MAX", "80"))
|
|
LOS_SAMPLE_STEP_METERS = int(os.getenv("LOS_SAMPLE_STEP_METERS", "250"))
|
|
ELEVATION_CACHE_TTL = int(os.getenv("ELEVATION_CACHE_TTL", "21600"))
|
|
LOS_CURVATURE_ENABLED = (
|
|
os.getenv("LOS_CURVATURE_ENABLED", "true").lower() == "true"
|
|
)
|
|
try:
|
|
LOS_CURVATURE_FACTOR = float(os.getenv("LOS_CURVATURE_FACTOR", "1.333333"))
|
|
except ValueError:
|
|
LOS_CURVATURE_FACTOR = 1.333333
|
|
if LOS_CURVATURE_FACTOR <= 0:
|
|
LOS_CURVATURE_FACTOR = 1.333333
|
|
LOS_PEAKS_MAX = int(os.getenv("LOS_PEAKS_MAX", "4"))
|
|
|
|
COVERAGE_API_URL = os.getenv("COVERAGE_API_URL", "").strip()
|
|
COVERAGE_API_KEY = os.getenv("COVERAGE_API_KEY", "").strip()
|
|
COVERAGE_MAX_AGE_DAYS = float(os.getenv("COVERAGE_MAX_AGE_DAYS", "30"))
|
|
COVERAGE_RATE_LIMIT_COOLDOWN_SECONDS = int(
|
|
os.getenv("COVERAGE_RATE_LIMIT_COOLDOWN_SECONDS", "3600")
|
|
)
|
|
COVERAGE_CACHE_FILE = os.getenv(
|
|
"COVERAGE_CACHE_FILE", os.path.join(STATE_DIR, "coverage_cache.json")
|
|
).strip()
|
|
COVERAGE_SYNC_INTERVAL_SECONDS = int(
|
|
os.getenv("COVERAGE_SYNC_INTERVAL_SECONDS", "3600")
|
|
)
|
|
WEATHER_RADAR_ENABLED = (
|
|
os.getenv("WEATHER_RADAR_ENABLED", "true").lower() == "true"
|
|
)
|
|
WEATHER_RADAR_COUNTRY_BOUNDS_ENABLED = (
|
|
os.getenv("WEATHER_RADAR_COUNTRY_BOUNDS_ENABLED", "false").lower() == "true"
|
|
)
|
|
WEATHER_RADAR_COUNTRY_LOOKUP_URL = os.getenv(
|
|
"WEATHER_RADAR_COUNTRY_LOOKUP_URL", "/weather/radar/country-bounds"
|
|
).strip()
|
|
WEATHER_WIND_ENABLED = (
|
|
os.getenv("WEATHER_WIND_ENABLED", "true").lower() == "true"
|
|
)
|
|
WEATHER_WIND_API_URL = os.getenv(
|
|
"WEATHER_WIND_API_URL", "https://api.open-meteo.com/v1/forecast"
|
|
).strip()
|
|
try:
|
|
WEATHER_WIND_GRID_SIZE = int(os.getenv("WEATHER_WIND_GRID_SIZE", "3"))
|
|
except ValueError:
|
|
WEATHER_WIND_GRID_SIZE = 3
|
|
if WEATHER_WIND_GRID_SIZE < 1:
|
|
WEATHER_WIND_GRID_SIZE = 1
|
|
if WEATHER_WIND_GRID_SIZE > 5:
|
|
WEATHER_WIND_GRID_SIZE = 5
|
|
try:
|
|
WEATHER_WIND_REFRESH_SECONDS = int(
|
|
os.getenv("WEATHER_WIND_REFRESH_SECONDS", "180")
|
|
)
|
|
except ValueError:
|
|
WEATHER_WIND_REFRESH_SECONDS = 180
|
|
if WEATHER_WIND_REFRESH_SECONDS < 30:
|
|
WEATHER_WIND_REFRESH_SECONDS = 30
|
|
|
|
TURNSTILE_ENABLED_RAW = os.getenv("TURNSTILE_ENABLED", "false").lower() == "true"
|
|
# Turnstile protection is only allowed when PROD_MODE is enabled.
|
|
TURNSTILE_ENABLED = PROD_MODE and TURNSTILE_ENABLED_RAW
|
|
TURNSTILE_SITE_KEY = os.getenv("TURNSTILE_SITE_KEY", "").strip()
|
|
TURNSTILE_SECRET_KEY = os.getenv("TURNSTILE_SECRET_KEY", "").strip()
|
|
TURNSTILE_API_URL = os.getenv(
|
|
"TURNSTILE_API_URL", "https://challenges.cloudflare.com/turnstile/v0/siteverify"
|
|
)
|
|
TURNSTILE_TOKEN_TTL_SECONDS = int(os.getenv("TURNSTILE_TOKEN_TTL_SECONDS", "86400"))
|
|
TURNSTILE_BOT_BYPASS = os.getenv("TURNSTILE_BOT_BYPASS", "true").lower() == "true"
|
|
TURNSTILE_BOT_ALLOWLIST = os.getenv(
|
|
"TURNSTILE_BOT_ALLOWLIST",
|
|
(
|
|
"discordbot,twitterbot,slackbot,facebookexternalhit,"
|
|
"linkedinbot,telegrambot,whatsapp,skypeuripreview,redditbot"
|
|
),
|
|
).strip()
|
|
|
|
APP_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
VERSION_FILE_CANDIDATES = (
|
|
"/repo/VERSION.txt",
|
|
os.path.join(os.path.dirname(APP_DIR), "VERSION.txt"),
|
|
os.path.join(APP_DIR, "VERSION.txt"),
|
|
)
|
|
APP_VERSION = "dev"
|
|
for VERSION_FILE in VERSION_FILE_CANDIDATES:
|
|
try:
|
|
with open(VERSION_FILE, "r", encoding="utf-8") as handle:
|
|
APP_VERSION = handle.read().strip() or "dev"
|
|
break
|
|
except Exception:
|
|
continue
|
|
NODE_SCRIPT_PATH = os.path.join(APP_DIR, "meshcore_decode.mjs")
|