import os
import re
import json
import math
import uuid
import asyncio
import logging
from collections import deque
from datetime import datetime
from typing import List, Optional
from fastapi import WebSocket

from app.config import (
    LOGS_DIR, FIRMWARE_DIR, FIRMWARE_BIN, OTA_MANIFEST_FILE, OTA_STATE_FILE,
    OTA_PUBLIC_URL, ADC_MAX, VREF_MV, VREF_V, GAIN_SHUNT, FF, VOC_DIVIDER,
    PCB_VERSIONS, DEFAULT_PCB_VERSION, PCB_STATE_FILE,
    PYRANO_MV_PER_WM2, AZ_DEV_DEFAULT, STEPS_80, DEG_SPAN, MAX_SAMPLES,
    HISTORY_LIMIT, LOCAL_TZ, BROADCAST_TIMEOUT, LOG_DAILY_MAX_BYTES, get_local_now
)

log = logging.getLogger("tracker_hub.hub")

RE_SAMPLE = re.compile(
    r"^(?:LOG:)?\[(?P<ts>[0-9\-/ :]+)\]\s*X:(?P<x>-?\d+),Y:(?P<y>-?\d+),"
    r"C:(?P<c>-?\d+),V:(?P<v>-?\d+),P:(?P<p>-?\d+),S:\d+,er:(?P<er>-?\d+)\s*$"
)
# Telemetría periódica (promedios cada 2 s) emitida por el firmware nuevo:
#   LOG:[YYYY-MM-DD HH:MM:SS] TEL: t=<uptime>s, steps=<tot>(+<delta>),
#        avgC=<c>, avgV=<v>, avgP=<p>, LPF=<lpf>, X=<x>, Y=<y>
# Desde el firmware 1.1.1 (F6) se agregan dos campos de diagnóstico ENTRE LPF
# y X:  prox=<0..1>  y  CmaxN=<counts>.  Son OPCIONALES en el parser para no
# romper el formato previo (builds sin F6 no los emiten).  X/Y llegan "crudos"
# (getPosX/getPosY); el resto del hub usa posición invertida, así que on_tel()
# los invierte para que las series se superpongan. Con firmware viejo no hay
# líneas TEL: y el parser simplemente no dispara.
RE_TEL = re.compile(
    r"^(?:LOG:)?\[(?P<ts>[0-9\-/ :]+)\]\s*TEL:\s*"
    r"t=(?P<t>\d+)s,\s*"
    r"steps=(?P<steps>\d+)\((?P<stepsDelta>[+\-]?\d+)\),\s*"
    r"avgC=(?P<avgC>-?\d+),\s*"
    r"avgV=(?P<avgV>-?\d+),\s*"
    r"avgP=(?P<avgP>-?\d+),\s*"
    r"LPF=(?P<lpf>-?\d+)"
    r"(?:,\s*prox=(?P<prox>-?\d+(?:\.\d+)?))?"
    r"(?:,\s*CmaxN=(?P<cmaxN>-?\d+))?"
    r",\s*"
    r"X=(?P<x>-?\d+),\s*"
    r"Y=(?P<y>-?\d+)\s*$"
)
RE_AXIS_ACK = re.compile(r"AXISCFG:(?P<x>\d+):(?P<y>\d+):(?P<base>\d+)")
RE_AXIS_ACK_LOG = re.compile(
    r"LOG:\[(?:\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}|sin hora)\] "
    r"AXISCFG:(?P<x>\d+):(?P<y>\d+):(?P<base>\d+)"
)
RE_PARAM = re.compile(
    r"^(?:LOG:)?\[(?P<ts>[0-9\-/ :]+)\]\s*(?P<name>Az_deviation|ErrMax|"
    r"MinErrMax|TiltAngle|delay)\s*=\s*(?P<val>[-\d.]+)\s*$"
)
_TS_FORMATS = ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%d/%m/%Y %H:%M:%S", "%d/%m/%Y %H:%M")

def parse_ts(s: str) -> Optional[datetime]:
    s = s.strip()
    for fmt in _TS_FORMATS:
        try:
            return datetime.strptime(s, fmt).replace(tzinfo=LOCAL_TZ)
        except Exception:
            pass
    return None


class HubManager:
    """Estado de UNA tarjeta (parsers, buffers, calibración, sesión ESP32).

    Constructible en modo standalone (`HubManager()`): efímero, sin SQLite ni
    I/O de estado global. Con `store` se persisten calibración/PCB/OTA-auto.
    """

    def __init__(self, device_id: Optional[str] = None, store=None):
        self.device_id = device_id
        self._store = store
        self.esp32_socket: Optional[WebSocket] = None
        self.esp32_connected = False
        self.client_sockets: List[WebSocket] = []
        self.last_pos_x = 0
        self.last_pos_y = 0
        self.last_seen: Optional[datetime] = None
        self.last_telemetry_at: Optional[datetime] = None
        self.connected_at: Optional[datetime] = None
        self.connection_id: Optional[str] = None
        self.pyrano = PYRANO_MV_PER_WM2
        self.azdev = AZ_DEV_DEFAULT
        self.samples: deque = deque(maxlen=MAX_SAMPLES)
        self.tel_samples: deque = deque(maxlen=MAX_SAMPLES)

        self.esp32_firmware_version: Optional[str] = None
        self.ota_manifest: Optional[dict] = None
        self.ota_auto_enabled: bool = False
        self.ota_url: str = OTA_PUBLIC_URL + "/ota/firmware"

        self.pcb_version: str = DEFAULT_PCB_VERSION
        self._send_lock = asyncio.Lock()
        self._auto_task: Optional[asyncio.Task] = None
        self.config_revision = -1
        self.log_error_count = 0
        self.log_error = None
        self._log_error_day = None
        self._version_session_id = None
        self.authorize_session = None

        if store is not None and device_id is not None:
            self._load_settings()

    # --------------------------- sesión ESP32 ---------------------------

    async def connect_esp32(self, websocket: WebSocket, on_connect=None) -> str:
        """Acepta una sesión autenticada. Si ya había otra del MISMO ID, la
        cierra explícitamente; los eventos tardíos de la antigua se ignoran
        por connection_id."""
        await websocket.accept()
        old = self.esp32_socket
        conn_id = uuid.uuid4().hex[:12]
        self.connection_id = conn_id
        self.esp32_socket = websocket
        self.esp32_connected = True
        self.connected_at = get_local_now()
        self.last_seen = self.connected_at
        self._version_session_id = None
        if on_connect is not None:
            on_connect(conn_id)
        if self._auto_task is not None:
            self._auto_task.cancel()
            self._auto_task = None
        log.info("esp32_connect", extra={"event": "esp32_connect", "device_id": self.device_id,
                                         "connection_id": conn_id})
        if old is not None and old is not websocket:
            try:
                await old.close(code=1000)
            except Exception:
                pass
        await self.broadcast_status()
        await self.broadcast_ota()
        return conn_id

    def disconnect_esp32(self, conn_id: Optional[str] = None):
        """Cierre de sesión. Un `finally` tardío de una sesión reemplazada NO
        desconecta a la nueva: se verifica connection_id."""
        if conn_id is not None and conn_id != self.connection_id:
            log.info("esp32_late_event_ignored", extra={"event": "esp32_late_event_ignored",
                                                         "device_id": self.device_id})
            return
        if self._auto_task is not None:
            self._auto_task.cancel()
            self._auto_task = None
        was_connected = self.esp32_connected
        self.esp32_socket = None
        self.esp32_connected = False
        self.connection_id = None
        if was_connected:
            log.info("esp32_disconnect", extra={"event": "esp32_disconnect", "device_id": self.device_id,
                                                "connection_id": conn_id})

    async def force_close(self, conn_id=None):
        """Cierre forzado (revocación de credenciales): cierra el socket y
        marca desconectado; publica el cambio de estado."""
        if conn_id is not None and conn_id != self.connection_id:
            return
        if self._auto_task is not None:
            self._auto_task.cancel()
            self._auto_task = None
        sock, self.esp32_socket = self.esp32_socket, None
        self.esp32_connected = False
        self.connection_id = None
        if sock is not None:
            try:
                await sock.close(code=1008)
            except Exception:
                pass
        log.warning("esp32_forced_close", extra={"event": "esp32_forced_close", "device_id": self.device_id})
        await self.broadcast_status()
        await self.broadcast_ota()

    async def handle_esp32_message(self, message: str, conn_id: Optional[str] = None):
        if conn_id is not None and (conn_id != self.connection_id or not self.esp32_connected):
            log.info("esp32_late_message_ignored", extra={"event": "esp32_late_message_ignored",
                                                           "device_id": self.device_id})
            return
        self.last_seen = get_local_now()
        line = message.strip()
        self._append_log(line)
        if not line:
            return
        try:
            await self._process_line(line)
        except (ValueError, TypeError, OverflowError):
            log.warning("line_rejected", extra={"event": "line_rejected", "device_id": self.device_id})

    async def _process_line(self, line):
        if line.startswith("POS:"):
            parts = line.split(":")
            if len(parts) >= 3:
                try:
                    self.last_pos_x, self.last_pos_y = int(parts[1]), int(parts[2])
                except ValueError:
                    pass

        axis_ack = RE_AXIS_ACK.fullmatch(line) or RE_AXIS_ACK_LOG.fullmatch(line)
        if axis_ack:
            x_factor, y_factor, base = (int(axis_ack.group(k)) for k in ("x", "y", "base"))
            if (1 <= x_factor <= 8 and 1 <= y_factor <= 8 and 1 <= base <= 64
                    and x_factor * base <= 64 and y_factor * base <= 64):
                await self._broadcast("AXISCFG:" + json.dumps({"x": x_factor, "y": y_factor, "base": base,
                        "device_id": self.device_id,
                        "x_budget": x_factor * base, "y_budget": y_factor * base}))

        if line.startswith("VER:"):
            v = line[4:].strip()
            self._version_session_id = self.connection_id
            if v and v != self.esp32_firmware_version:
                self.esp32_firmware_version = v
                await self._broadcast("FW:" + v)
                await self.broadcast_ota()
            if v:
                self._maybe_auto_update()

        m = RE_SAMPLE.match(line)
        if m:
            pt = self.on_sample(m.groupdict())
            self.last_telemetry_at = get_local_now()
            if pt:
                await self._broadcast("DATA:" + json.dumps(pt))

        # Telemetría promediada (firmware nuevo). Con firmware viejo no hay
        # líneas TEL: y este bloque no se ejecuta (compatibilidad hacia atrás).
        mt = RE_TEL.match(line)
        if mt:
            tpt = self.on_tel(mt.groupdict())
            self.last_telemetry_at = get_local_now()
            if tpt:
                await self._broadcast("TEL:" + json.dumps(tpt))

        pm = RE_PARAM.match(line)
        if pm:
            name, val = pm.group("name"), float(pm.group("val"))
            if name == "Az_deviation" and abs(val - self.azdev) > 1e-6:
                # Persistir ANTES de mutar el estado en memoria.
                if self._store is not None:
                    try:
                        self._store.save_settings(self.device_id, az_dev=val)
                    except Exception as e:
                        log.error("azdev_persist_failed", extra={"event": "azdev_persist_failed",
                                                                  "device_id": self.device_id,
                                                                  "error_type": type(e).__name__})
                    else:
                        self.azdev = val
                        self._recompute_position()
                        await self._broadcast("CONFIG:" + self._config_json())
                else:
                    self.azdev = val
                    self._recompute_position()
                    await self._broadcast("CONFIG:" + self._config_json())

        await self._broadcast("LOG:" + line + "\n")

    # ------------------------- conversión de datos ------------------------

    def _units(self, x, y, c, v, p):
        r_shunt = PCB_VERSIONS[self.pcb_version]
        gpoa = p / ADC_MAX * (VREF_MV / self.pyrano)
        voc = v / ADC_MAX * VREF_V * VOC_DIVIDER
        isc = (c / ADC_MAX * (VREF_MV / GAIN_SHUNT)) / 1000.0 / r_shunt
        pcpv = voc * isc * FF
        return gpoa, voc, isc, pcpv

    def _pos_deg(self, x, y):
        x_base = y - self.azdev / 90.0 * x
        el = 90.0 - (x_base - STEPS_80) / STEPS_80 * DEG_SPAN
        az = 180.0 + (x - STEPS_80) / STEPS_80 * DEG_SPAN
        return az, el

    def on_sample(self, d):
        ts = parse_ts(d["ts"])
        t = ts.timestamp() if ts else get_local_now().timestamp()
        x, y, c, v, p = int(d["x"]), int(d["y"]), int(d["c"]), int(d["v"]), int(d["p"])
        gpoa, voc, isc, pcpv = self._units(x, y, c, v, p)
        az, el = self._pos_deg(x, y)
        pt = {
            "ts": round(t, 3), "x": x, "y": y, "p_raw": p, "c_raw": c, "v_raw": v,
            "gpoa": round(gpoa, 1), "voc": round(voc, 3),
            "isc_ma": round(isc * 1000, 2), "pcpv_mw": round(pcpv * 1000, 2),
            "az": round(az, 2), "el": round(el, 2)
        }
        if self.device_id:
            pt["device_id"] = self.device_id
        self.samples.append(pt)
        return pt

    def on_tel(self, d):
        ts = parse_ts(d["ts"])
        t_wall = ts.timestamp() if ts else get_local_now().timestamp()
        avg_c, avg_v, avg_p = int(d["avgC"]), int(d["avgV"]), int(d["avgP"])
        # X/Y crudos del firmware -> se invierten para alinear con el sistema
        # de coordenadas del hub (muestras y POS usan valores negados).
        x, y = -int(d["x"]), -int(d["y"])
        gpoa, voc, isc, pcpv = self._units(x, y, avg_c, avg_v, avg_p)
        az, el = self._pos_deg(x, y)
        pt = {
            "ts": round(t_wall, 3), "t": int(d["t"]),
            "steps": int(d["steps"]), "steps_delta": int(d["stepsDelta"]),
            "avg_c": avg_c, "avg_v": avg_v, "avg_p": avg_p,
            "lpf": int(d["lpf"]), "x": x, "y": y,
            # F6 (firmware >= 1.1.1): campos de diagnóstico opcionales.
            "prox": float(d["prox"]) if d.get("prox") is not None else None,
            "cmaxn": int(d["cmaxN"]) if d.get("cmaxN") is not None else None,
            "gpoa": round(gpoa, 1), "voc": round(voc, 3),
            "isc_ma": round(isc * 1000, 2), "pcpv_mw": round(pcpv * 1000, 2),
            "az": round(az, 2), "el": round(el, 2),
        }
        if self.device_id:
            pt["device_id"] = self.device_id
        self.tel_samples.append(pt)
        return pt

    def _recompute_position(self):
        for pt in self.samples:
            az, el = self._pos_deg(pt["x"], pt["y"])
            pt["az"], pt["el"] = round(az, 2), round(el, 2)
        for pt in self.tel_samples:
            az, el = self._pos_deg(pt["x"], pt["y"])
            pt["az"], pt["el"] = round(az, 2), round(el, 2)

    def _config_json(self):
        return json.dumps({
            "device_id": self.device_id,
            "pyrano": self.pyrano, "azdev": self.azdev, "adc_max": ADC_MAX,
            "vref_mv": VREF_MV, "vref_v": VREF_V, "gain_shunt": GAIN_SHUNT,
            "r_shunt": PCB_VERSIONS[self.pcb_version], "ff": FF, "voc_div": VOC_DIVIDER,
            "pcb_version": self.pcb_version,
            "steps80": STEPS_80, "deg_span": DEG_SPAN, "pos": [self.last_pos_x, self.last_pos_y]
        })

    # ---------------------------- suscriptores ----------------------------

    async def connect_client(self, websocket: WebSocket):
        await websocket.accept()
        self.client_sockets.append(websocket)
        try:
            # CONFIG va como instantánea inicial completa (marcador de
            # "listo") ANTES de cualquier historial, para que la UI no vea
            # datos con configuración vieja.
            await self._send_to(websocket, "STATUS:%s" % ("ONLINE" if self.esp32_connected else "OFFLINE"))
            await self._send_to(websocket, "CONFIG:" + self._config_json())
            await self._send_to(websocket, "HISTORY:" + json.dumps(list(self.samples)[-HISTORY_LIMIT:]))
            await self._send_to(websocket, "TEL_HISTORY:" + json.dumps(list(self.tel_samples)[-HISTORY_LIMIT:]))
            await self._send_to(websocket, "POS:%d:%d" % (self.last_pos_x, self.last_pos_y))
            if self.esp32_firmware_version:
                await self._send_to(websocket, "FW:" + self.esp32_firmware_version)
            await self._send_to(websocket, "OTA_STATE:" + json.dumps(self._ota_payload()))

            today = f"log_{get_local_now():%Y-%m-%d}.txt"
            base = LOGS_DIR if self.device_id is None else os.path.join(LOGS_DIR, self.device_id)
            path = os.path.join(base, today)
            if os.path.exists(path):
                with open(path, "rb") as f:
                    f.seek(max(0, os.path.getsize(path) - 655360))
                    tail = "".join(f.read().decode("utf-8", errors="replace").splitlines(keepends=True)[-40:])
                if tail:
                    await self._send_to(websocket, "LOG:" + tail)
        except Exception as e:
            log.error("connect_client_error", extra={"event": "connect_client_error",
                                                     "device_id": self.device_id,
                                                     "error_type": type(e).__name__})

    def disconnect_client(self, websocket):
        if websocket in self.client_sockets:
            self.client_sockets.remove(websocket)

    async def _send_to(self, client, msg: str):
        """Envío acotado por timeout: un navegador lento solo se corta a él."""
        try:
            await asyncio.wait_for(client.send_text(msg), timeout=BROADCAST_TIMEOUT)
        except Exception:
            self.disconnect_client(client)
            try:
                await asyncio.wait_for(client.close(code=1001), timeout=BROADCAST_TIMEOUT)
            except Exception:
                pass

    async def _broadcast(self, msg):
        await asyncio.gather(*(self._send_to(c, msg) for c in list(self.client_sockets)))

    async def broadcast_status(self):
        await self._broadcast("STATUS:%s" % ("ONLINE" if self.esp32_connected else "OFFLINE"))

    async def send_command_to_esp32(self, command: str, expected_connection=None) -> bool:
        """Envía al socket ACTUAL de esta tarjeta; escrituras serializadas.
        Sin ACK del firmware solo existe estado 'enviado', no 'ejecutado'."""
        if not (self.esp32_socket and self.esp32_connected):
            return False
        async with self._send_lock:
            sock = self.esp32_socket
            connection_id = self.connection_id
            if (sock is None or not self.esp32_connected
                    or (expected_connection is not None and expected_connection != connection_id)):
                return False
            if self._store is not None and (self.authorize_session is None or not self.authorize_session()):
                return False
            try:
                await asyncio.wait_for(sock.send_text(command), timeout=BROADCAST_TIMEOUT)
                log.info("command_sent", extra={"event": "command_sent", "device_id": self.device_id,
                                                "connection_id": self.connection_id})
                return True
            except Exception as e:
                log.error("command_failed", extra={"event": "command_failed", "device_id": self.device_id,
                                                    "error_type": type(e).__name__})
                self.disconnect_esp32(connection_id)
                await self.broadcast_status()
                await self.broadcast_ota()
                return False

    # --------------------- calibración / PCB (persiste 1ro) ---------------------

    def apply_calibration(self, mV_per_WM2=None, az_dev=None) -> dict:
        """Valida, persiste (si hay store) y luego muta. Si la persistencia
        falla, el estado previo se conserva y se propaga el error."""
        new_pyrano, new_az = self.pyrano, self.azdev
        changed_pyrano = changed_az = False
        if mV_per_WM2 is not None:
            v = float(mV_per_WM2)
            if not math.isfinite(v) or v <= 0:
                raise ValueError("mV_per_WM2 debe ser finita y positiva")
            new_pyrano, changed_pyrano = v, True
        if az_dev is not None:
            v = float(az_dev)
            if not math.isfinite(v):
                raise ValueError("az_dev debe ser finita")
            new_az, changed_az = v, True
        if not (changed_pyrano or changed_az):
            return {"mV_per_WM2": self.pyrano, "az_dev": self.azdev}
        if self._store is not None:
            self._store.save_settings(self.device_id, pyrano=new_pyrano, az_dev=new_az)
        if changed_pyrano:
            self.pyrano = new_pyrano
            for pt in self.samples:
                if pt.get("p_raw") is not None:
                    pt["gpoa"] = round(pt["p_raw"] / ADC_MAX * (VREF_MV / self.pyrano), 1)
            for pt in self.tel_samples:
                if pt.get("avg_p") is not None:
                    pt["gpoa"] = round(pt["avg_p"] / ADC_MAX * (VREF_MV / self.pyrano), 1)
        if changed_az:
            self.azdev = new_az
            self._recompute_position()
        log.info("calibration_updated", extra={"event": "calibration_updated", "device_id": self.device_id,
                                               "result": {"pyrano": self.pyrano, "az_dev": self.azdev}})
        return {"mV_per_WM2": self.pyrano, "az_dev": self.azdev}

    def set_pcb_version(self, version) -> bool:
        if not isinstance(version, str) or version not in PCB_VERSIONS:
            return False
        if self._store is not None:
            self._store.save_settings(self.device_id, pcb_version=version)
        self.pcb_version = version
        r = PCB_VERSIONS[version]
        for pt in self.samples:
            c = pt.get("c_raw")
            if c is None:
                continue
            isc = (c / ADC_MAX * (VREF_MV / GAIN_SHUNT)) / 1000.0 / r
            pt["isc_ma"] = round(isc * 1000, 2)
            if pt.get("voc") is not None:
                pt["pcpv_mw"] = round(pt["voc"] * isc * FF * 1000, 2)
        for pt in self.tel_samples:
            c = pt.get("avg_c")
            if c is None:
                continue
            isc = (c / ADC_MAX * (VREF_MV / GAIN_SHUNT)) / 1000.0 / r
            pt["isc_ma"] = round(isc * 1000, 2)
            if pt.get("voc") is not None:
                pt["pcpv_mw"] = round(pt["voc"] * isc * FF * 1000, 2)
        log.info("pcb_updated", extra={"event": "pcb_updated", "device_id": self.device_id,
                                      "result": version})
        return True

    async def rebroadcast_config(self):
        await self._broadcast("CONFIG:" + self._config_json())
        await self._broadcast("HISTORY:" + json.dumps(list(self.samples)[-HISTORY_LIMIT:]))
        await self._broadcast("TEL_HISTORY:" + json.dumps(list(self.tel_samples)[-HISTORY_LIMIT:]))

    # ------------------------------- OTA --------------------------------

    def _ota_payload(self) -> dict:
        return {
            "device_id": self.device_id,
            "device": self.esp32_firmware_version, "available": self.ota_manifest,
            "auto": self.ota_auto_enabled, "url": self.ota_url, "connected": self.esp32_connected,
        }

    async def broadcast_ota(self):
        await self._broadcast("OTA_STATE:" + json.dumps(self._ota_payload()))

    async def set_ota_auto(self, enabled: bool):
        """OTA automática POR TARJETA (default False; no se hereda de un
        estado global)."""
        if self._store is not None:
            self._store.save_settings(self.device_id, ota_auto=bool(enabled))
        self.ota_auto_enabled = bool(enabled)
        if not self.ota_auto_enabled and self._auto_task is not None:
            self._auto_task.cancel()
            self._auto_task = None
        await self.broadcast_ota()

    async def push_ota(self, sha256: Optional[str] = None, expected_connection=None) -> bool:
        """Push manual de OTA a ESTA tarjeta. Si se indica sha256, debe
        coincidir con el manifiesto actual (previene carreras de archivo)."""
        if not (self.esp32_connected and self.esp32_socket and self.ota_manifest and os.path.exists(FIRMWARE_BIN)):
            return False
        if sha256:
            cur = str(self.ota_manifest.get("sha256") or "").lower()
            if not cur or not _const_eq(str(sha256).lower(), cur):
                return False
        manifest = self.ota_manifest.copy()
        ok = await self.send_command_to_esp32("OTA:" + manifest["url"], expected_connection)
        if ok:
            await self._broadcast(
                f"LOG:[hub] Empujando firmware '{manifest.get('version', '?')}' a la ESP32 (enviado, no ejecutado)...\n"
            )
        return ok

    def _maybe_auto_update(self):
        """Flujo auto POR tarjeta, ligado a identidad Y sesión; se cancela al
        desconectar. La subida de artefacto NUNCA invoca esto."""
        try:
            if self._auto_task is not None:
                self._auto_task.cancel()
                self._auto_task = None
            if not self.ota_auto_enabled or not self.ota_manifest or not os.path.exists(FIRMWARE_BIN):
                return
            if not self.esp32_firmware_version or self._version_session_id != self.connection_id:
                return
            if self.esp32_firmware_version == self.ota_manifest.get("version"):
                return
            cid = self.connection_id
            sha = self.ota_manifest.get("sha256")
            version = self.ota_manifest.get("version")

            async def _do():
                await asyncio.sleep(3)
                if (self.ota_auto_enabled and self.esp32_connected and cid is not None
                        and self.connection_id == cid and self.esp32_firmware_version != version):
                    await self.push_ota(sha, cid)

            self._auto_task = asyncio.create_task(_do())
        except Exception as e:
            log.error("auto_update_error", extra={"event": "auto_update_error",
                                                  "device_id": self.device_id,
                                                  "error_type": type(e).__name__})

    # --------------------------- persistencia ---------------------------

    def _load_settings(self):
        s = self._store.get_settings(self.device_id)
        if s is None:
            raise ValueError("Missing registered device settings")
        if s["config_revision"] == self.config_revision:
            return
        self.pcb_version = s["pcb_version"]
        self.pyrano, self.azdev = s["pyrano"], s["az_dev"]
        self.ota_auto_enabled = s["ota_auto"]
        self.config_revision = s["config_revision"]
        for points, keys in ((self.samples, ("c_raw", "v_raw", "p_raw")),
                             (self.tel_samples, ("avg_c", "avg_v", "avg_p"))):
            for point in points:
                g, v, i, p = self._units(point["x"], point["y"], *(point[k] for k in keys))
                point.update(gpoa=round(g, 1), voc=round(v, 3), isc_ma=round(i * 1000, 2), pcpv_mw=round(p * 1000, 2))
        self._recompute_position()

    def _append_log(self, line: str):
        if not line:
            return
        fn = f"log_{get_local_now():%Y-%m-%d}.txt"
        base = LOGS_DIR if self.device_id is None else os.path.join(LOGS_DIR, self.device_id)
        try:
            os.makedirs(base, exist_ok=True)
            path = os.path.join(base, fn)
            content = (line + "\n").encode("utf-8")
            if (os.path.getsize(path) if os.path.exists(path) else 0) + len(content) > LOG_DAILY_MAX_BYTES:
                raise OSError("daily log quota")
            with open(path, "ab") as f:
                f.write(content)
            self.log_error = None
        except Exception as e:
            # Modo degradado visible: la ingesta continua, el log lo registra.
            self.log_error_count += 1
            self.log_error = type(e).__name__
            if self._log_error_day == fn:
                return
            self._log_error_day = fn
            log.error("log_write_failed", extra={"event": "log_write_failed",
                                                 "device_id": self.device_id,
                                                 "error_type": type(e).__name__})

    def status_payload(self, alias: Optional[str] = None) -> dict:
        return {
            "device_id": self.device_id,
            "alias": alias,
            "esp32_connected": self.esp32_connected,
            "connected_at": self.connected_at.isoformat() if self.connected_at else None,
            "last_seen": self.last_seen.isoformat() if self.last_seen else None,
            "last_telemetry_at": self.last_telemetry_at.isoformat() if self.last_telemetry_at else None,
            "connection_id": self.connection_id,
            "pos_x": self.last_pos_x, "pos_y": self.last_pos_y,
            "pyrano_mV_per_Wm2": self.pyrano, "az_dev_deg": self.azdev,
            "pcb_version": self.pcb_version,
            "active_clients": len(self.client_sockets),
            "samples_in_buffer": len(self.samples),
            "tel_samples_in_buffer": len(self.tel_samples),
            "firmware_device": self.esp32_firmware_version,
            "log_degraded": self.log_error is not None,
            "log_error_count": self.log_error_count,
            "ota": self._ota_payload(),
        }


def _const_eq(a: str, b: str) -> bool:
    import hmac
    return hmac.compare_digest(a, b)


# Nota: HubManager() sigue siendo constructible standalone (sin argumentos,
# sin SQLite ni I/O de estado global) para pruebas efímeras. No hay
# singleton global: cada proceso/instancia gestiona sus propias tarjetas.