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")