| """ |
| Gestión de estado global y métricas de telemetría del nodo. |
| 100% thread-safe con gestión de CurrentJob idéntica a main.go. |
| """ |
|
|
| import time |
| import threading |
| from collections import deque |
| from typing import Optional, Dict, Any, List, Callable |
| import subprocess |
|
|
| from config import START_TIME, logger |
|
|
|
|
| class CurrentJob: |
| """Representa un trabajo activo en ejecución o pendiente de subida.""" |
|
|
| def __init__(self, task_id: str, file_name: str, upload_url: str): |
| self.task_id: str = task_id |
| self.file_name: str = file_name |
| self.upload_url: str = upload_url |
| self.done: bool = False |
| self.proc: Optional[subprocess.Popen] = None |
| self.cancel_func: Optional[Callable[[], None]] = None |
| self.start_time: float = time.time() |
| self.percent: float = 0.0 |
| self.speed: str = "--" |
| self.eta: str = "--" |
| self.status: str = "Iniciando..." |
|
|
|
|
| class ClusterState: |
| """Contenedor de estado compartido entre el cliente WS, FFmpeg y la UI de Gradio.""" |
|
|
| def __init__(self): |
| self.lock = threading.RLock() |
| self.status: str = "Iniciando..." |
| self.is_connected: bool = False |
| self.last_error: str = "" |
| self.was_connected: bool = False |
|
|
| |
| self.active_jobs: Dict[str, CurrentJob] = {} |
|
|
| |
| self.total_completed: int = 0 |
| self.total_failed: int = 0 |
| self.total_orig_bytes: int = 0 |
| self.total_comp_bytes: int = 0 |
| self.logs: deque = deque(maxlen=200) |
|
|
| def log(self, message: str, level: str = "INFO"): |
| ts = time.strftime("%H:%M:%S") |
| entry = f"[{ts}] {message}" |
| with self.lock: |
| self.logs.append(entry) |
| if level == "ERROR": |
| logger.error(message) |
| elif level == "WARNING": |
| logger.warning(message) |
| else: |
| logger.info(message) |
|
|
| def set_connection(self, connected: bool, error: str = ""): |
| with self.lock: |
| self.is_connected = connected |
| self.last_error = error |
| if connected: |
| self.was_connected = True |
| self.status = "🟢 Conectado" |
| else: |
| self.status = "🔴 Reconectando" if error else "⚪ Desconectado" |
|
|
| def get_state(self) -> tuple[bool, str]: |
| with self.lock: |
| return self.is_connected, self.last_error |
|
|
| def add_job(self, job: CurrentJob): |
| with self.lock: |
| self.active_jobs[job.task_id] = job |
|
|
| def get_job(self, task_id: str) -> Optional[CurrentJob]: |
| with self.lock: |
| return self.active_jobs.get(task_id) |
|
|
| def remove_job(self, task_id: str): |
| with self.lock: |
| self.active_jobs.pop(task_id, None) |
|
|
| def get_active_tasks_for_handshake(self) -> List[Dict[str, str]]: |
| """Devuelve las tareas activas para reportarlas en el Handshake si reconecta.""" |
| with self.lock: |
| tasks = [] |
| for j in self.active_jobs.values(): |
| status = "completed" if j.done else "converting" |
| tasks.append({ |
| "task_id": j.task_id, |
| "status": status, |
| "upload_url": j.upload_url, |
| }) |
| return tasks |
|
|
| def record_success(self, orig_size: int, comp_size: int): |
| with self.lock: |
| self.total_completed += 1 |
| self.total_orig_bytes += orig_size |
| self.total_comp_bytes += comp_size |
|
|
| def record_fail(self): |
| with self.lock: |
| self.total_failed += 1 |
|
|
| def get_summary(self) -> Dict[str, Any]: |
| with self.lock: |
| saved_bytes = max(0, self.total_orig_bytes - self.total_comp_bytes) |
| saved_mb = saved_bytes / (1024 * 1024) |
|
|
| |
| active_info = None |
| if self.active_jobs: |
| first_job = next(iter(self.active_jobs.values())) |
| active_info = { |
| "task_id": first_job.task_id, |
| "file_name": first_job.file_name, |
| "percent": first_job.percent, |
| "speed": first_job.speed, |
| "eta": first_job.eta, |
| "status": first_job.status, |
| } |
|
|
| return { |
| "status": self.status, |
| "connected": self.is_connected, |
| "error": self.last_error, |
| "job": active_info, |
| "active_count": len(self.active_jobs), |
| "completed": self.total_completed, |
| "failed": self.total_failed, |
| "saved_mb": round(saved_mb, 2), |
| "uptime": int(time.time() - START_TIME), |
| "logs": "\n".join(self.logs), |
| } |
|
|
|
|
| STATE = ClusterState() |
|
|
