631 lines
29 KiB
Python
631 lines
29 KiB
Python
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.
|