96 lines
3.9 KiB
Python
96 lines
3.9 KiB
Python
import os
|
|
import re
|
|
from datetime import datetime, timezone, timedelta
|
|
from zoneinfo import ZoneInfo
|
|
|
|
# Aguascalientes está en UTC-6 todo el año (sin horario de verano).
|
|
# Usamos un offset fijo para evitar que tzdata desactualizado aplique DST en octubre.
|
|
LOCAL_TZ = timezone(timedelta(hours=-6))
|
|
|
|
def get_local_now() -> datetime:
|
|
"""Devuelve la fecha/hora actual fijada en UTC-6."""
|
|
return datetime.now(LOCAL_TZ)
|
|
|
|
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
# Directorios configurables por env (permite tests con rutas temporales sin
|
|
# tocar datos reales).
|
|
LOGS_DIR = os.environ.get("LOGS_DIR") or os.path.join(BASE_DIR, "data_logs")
|
|
STATIC_DIR = os.environ.get("STATIC_DIR") or os.path.join(BASE_DIR, "static")
|
|
FIRMWARE_DIR = os.environ.get("FIRMWARE_DIR") or os.path.join(BASE_DIR, "firmware")
|
|
|
|
# Estado multitarjeta (identidades, tokens hasheados, ajustes) en SQLite.
|
|
STATE_DIR = os.environ.get("STATE_DIR") or os.path.join(BASE_DIR, "state")
|
|
DATABASE_PATH = os.path.join(STATE_DIR, "devices.sqlite3")
|
|
PROVISION_DIR = os.path.join(STATE_DIR, "provision")
|
|
|
|
FIRMWARE_BIN = os.path.join(FIRMWARE_DIR, "firmware.bin")
|
|
OTA_MANIFEST_FILE = os.path.join(FIRMWARE_DIR, "manifest.json")
|
|
OTA_STATE_FILE = os.path.join(FIRMWARE_DIR, "ota_state.json")
|
|
|
|
OTA_SLOT_MAX = 0x640000
|
|
OTA_PUBLIC_URL = os.environ.get("OTA_PUBLIC_URL", "https://micro.ferrasdh.uk").rstrip("/")
|
|
PCB_STATE_FILE = os.path.join(FIRMWARE_DIR, "pcb_state.json")
|
|
|
|
# Constantes hardware / calibración
|
|
ADC_MAX = 8388607.0
|
|
VREF_MV = 2048.0
|
|
VREF_V = 2.048
|
|
GAIN_SHUNT = 20.0
|
|
R_SHUNT = 0.075 # legacy / valor por defecto; el hub usa PCB_VERSIONS[pcb_version]
|
|
PCB_VERSIONS = {"V1": 0.075, "V2": 0.130} # r_shunt en ohms por versión de PCB
|
|
DEFAULT_PCB_VERSION = "V1"
|
|
FF = 0.82
|
|
VOC_DIVIDER = 2.0
|
|
|
|
PYRANO_MV_PER_WM2 = 0.2
|
|
AZ_DEV_DEFAULT = 5.0
|
|
STEPS_80 = 53000.0
|
|
DEG_SPAN = 80.0
|
|
|
|
MAX_SAMPLES = max(1, int(os.environ.get("MAX_SAMPLES", "15000")))
|
|
HISTORY_LIMIT = MAX_SAMPLES
|
|
|
|
# ----------------------- Límites y seguridad -----------------------
|
|
|
|
# Máximo de tarjetas registradas (buffers solo se reservan para IDs conocidos).
|
|
MAX_DEVICES = int(os.environ.get("MAX_DEVICES", "16"))
|
|
|
|
# Canal de ingesta ESP32: límite por mensaje (bytes) y tasa por tarjeta.
|
|
WS_MESSAGE_MAX = int(os.environ.get("WS_MESSAGE_MAX", str(16 * 1024)))
|
|
WS_RATE_PER_SEC = float(os.environ.get("WS_RATE_PER_SEC", "100"))
|
|
WS_RATE_BURST = int(os.environ.get("WS_RATE_BURST", "200"))
|
|
|
|
# Timeout de envío a un suscriptor de navegador lento (solo corta a ese cliente).
|
|
BROADCAST_TIMEOUT = float(os.environ.get("BROADCAST_TIMEOUT", "5"))
|
|
|
|
# Retención de logs diarios por tarjeta (días). No toca archivos legacy en la
|
|
# raíz de LOGS_DIR.
|
|
LOG_RETENTION_DAYS = int(os.environ.get("LOG_RETENTION_DAYS", "30"))
|
|
LOG_DAILY_MAX_BYTES = max(1, int(os.environ.get("LOG_DAILY_MAX_BYTES", str(50 * 1024 * 1024))))
|
|
|
|
# Sondeo periódico de credenciales (la revocación desde CLI, proceso separado,
|
|
# cierra la sesión activa en este intervalo).
|
|
CREDENTIAL_POLL_SECONDS = float(os.environ.get("CREDENTIAL_POLL_SECONDS", "5"))
|
|
|
|
# Caducidad del token anterior tras rotar (horas).
|
|
TOKEN_GRACE_HOURS = int(os.environ.get("TOKEN_GRACE_HOURS", "24"))
|
|
|
|
# Aliases legacy: solo se habilitan con una tarjeta primaria explícita.
|
|
PRIMARY_DEVICE_ID = os.environ.get("PRIMARY_DEVICE_ID", "").strip() or None
|
|
|
|
# Orígenes permitidos para el WebSocket de navegador (reemplaza CORS "*").
|
|
# El canal ESP32 no usa Origin: se autentica por token.
|
|
_DEFAULT_ORIGINS = "https://micro.ferrasdh.uk,http://localhost:4000,http://127.0.0.1:4000"
|
|
ALLOWED_ORIGINS = {
|
|
o.strip().rstrip("/")
|
|
for o in os.environ.get("ALLOWED_ORIGINS", _DEFAULT_ORIGINS).split(",")
|
|
if o.strip()
|
|
}
|
|
|
|
# Identidad: PCB_ID configurado o MAC eFuse (12 hex minúsculos con prefijo estable).
|
|
DEVICE_ID_RE = re.compile(r"^esp32-[0-9a-f]{12}$")
|
|
|
|
def is_valid_device_id(value) -> bool:
|
|
return isinstance(value, str) and bool(DEVICE_ID_RE.fullmatch(value))
|