"""Lo que une el nucleo con la cara: que se empuja al HUD y que se hace al pulsar. `servidor.py` solo mueve bytes; aqui vive el significado. Es el equivalente de lo que hacia `jarvis.py` dentro de Newelle —el fichero de 1.244 lineas que era lo unico acoplado al framework— pero sin GTK, sin WebKit y sin el ciclo de vida de la aplicacion de por medio. ## Lo que se empuja y cada cuanto telemetria 1 s cpu, memoria, grafica, disco, procesos pulso 15/s solo mientras escucha o habla; es lo que mueve el reactor estado cuando cambia tarea cuando cambia una La telemetria cuesta procesos (`nvidia-smi`, `ps`), asi que va en su propio hilo y a su ritmo, no en el bucle de voz. Y el pulso solo se manda si hay alguien mirando: sin panel abierto son quince mensajes por segundo a nadie. """ import os import shutil import signal import subprocess import threading import time from manos.preguntar import confirmacion, es_si, pregunta_por AQUI = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) def _lee(ruta, defecto=""): try: with open(ruta) as f: return f.read().strip() except OSError: return defecto def _manda(orden, tope=4000): try: r = subprocess.run(orden, capture_output=True, text=True, timeout=10) return (r.stdout or "").strip()[:tope] except (OSError, subprocess.TimeoutExpired): return "" class Telemetria: """Las constantes de la maquina, leidas baratas. Todo de /proc salvo la grafica, que necesita nvidia-smi. Se lee /proc a mano en vez de usar psutil para no arrastrar una dependencia por cuatro numeros que estan en ficheros de texto. """ def __init__(self): self._cpu_previo = None self._nucleos_previo = None self._hay_nvidia = shutil.which("nvidia-smi") is not None def _cpu(self): try: with open("/proc/stat") as f: lineas = [l for l in f if l.startswith("cpu")] except OSError: return 0, [] def uso(campos): n = [int(x) for x in campos[1:11]] ocupado = sum(n) - n[3] - n[4] return ocupado, sum(n) ahora = [uso(l.split()) for l in lineas] previo, self._cpu_previo = self._cpu_previo, ahora if not previo or len(previo) != len(ahora): return 0, [0] * (len(ahora) - 1) def pct(a, b): d_ocu, d_tot = a[0] - b[0], a[1] - b[1] return round(100 * d_ocu / d_tot) if d_tot > 0 else 0 return pct(ahora[0], previo[0]), [pct(a, b) for a, b in zip(ahora[1:], previo[1:])] def _memoria(self): datos = {} try: with open("/proc/meminfo") as f: for l in f: k, _, v = l.partition(":") datos[k] = int(v.split()[0]) // 1024 # MB except OSError: return 0, 0, 0, 0 total = datos.get("MemTotal", 0) libre = datos.get("MemAvailable", 0) stotal = datos.get("SwapTotal", 0) slibre = datos.get("SwapFree", 0) return total, total - libre, stotal, stotal - slibre def _grafica(self): if not self._hay_nvidia: return None s = _manda(["nvidia-smi", "--query-gpu=utilization.gpu,memory.used,memory.total", "--format=csv,noheader,nounits"]) try: u, usada, total = (int(x.strip()) for x in s.split(",")[:3]) return {"uso": u, "usada": usada, "total": total} except (ValueError, IndexError): return None def _temperatura(self): mejor = 0 for zona in range(12): t = _lee(f"/sys/class/thermal/thermal_zone{zona}/temp") if t.isdigit(): mejor = max(mejor, int(t) // 1000) return mejor def _procesos(self, tope=5): """Los que mas CPU comen. cpu y mem van como NUMEROS, no como las cadenas que devuelve ps. El HUD hace `p.cpu.toFixed(0)`, y una cadena no tiene toFixed: la excepcion aborta el manejador entero de telemetria y se lleva por delante todo lo que se pinta despues —cpu, ram y grafica se quedaban en "--"—. El sintoma no apuntaba a los procesos en absoluto. """ s = _manda(["ps", "-eo", "pid,comm,pcpu,pmem", "--sort=-pcpu"]) salida = [] for l in s.splitlines()[1:tope + 1]: c = l.split(None, 3) if len(c) < 4: continue try: salida.append({"pid": int(c[0]), "nombre": c[1][:18], "cpu": float(c[2]), "mem": float(c[3].split()[0])}) except ValueError: continue return salida def _disco(self): try: st = os.statvfs("/") total = st.f_blocks * st.f_frsize / 1e9 libre = st.f_bavail * st.f_frsize / 1e9 return round(total - libre, 1), round(total, 1) except OSError: return 0, 0 def _red(self): """Bytes acumulados por todas las interfaces menos la de loopback.""" baja = sube = 0 try: with open("/proc/net/dev") as f: for l in f.readlines()[2:]: nombre, _, resto = l.partition(":") if nombre.strip() == "lo": continue c = resto.split() baja += int(c[0]); sube += int(c[8]) except (OSError, IndexError, ValueError): return None previo, self._red_previo = getattr(self, "_red_previo", None), (baja, sube, time.time()) if not previo: return {"baja": 0, "sube": 0} dt = max(self._red_previo[2] - previo[2], 0.001) return {"baja": max(0, (baja - previo[0]) / dt), "sube": max(0, (sube - previo[1]) / dt)} def _bateria(self): for n in range(3): p = _lee(f"/sys/class/power_supply/BAT{n}/capacity") if p.isdigit(): return int(p) return None def lee(self) -> dict: """Las unidades son las que espera el HUD, no las de /proc. Se convierte aqui y no en JavaScript porque el formato lo fija la pantalla: hud.js escribe "GB" y "TB" literalmente al lado del numero (ponMedidor), asi que mandar megas se ve como "63488/64256 GB" y parece que el panel esta roto cuando lo que esta mal es quien lo llena. """ cpu, nucleos = self._cpu() mtotal, musada, stotal, susada = self._memoria() dusado, dtotal = self._disco() arriba = _lee("/proc/uptime", "0").split()[0] datos = { "cpu": cpu, "nucleos": nucleos, "ram": {"usada": musada / 1024, "total": mtotal / 1024}, # GB "swap": {"usada": susada / 1024, "total": max(stotal / 1024, 0.01)}, "disco": {"usado": dusado / 1000, "total": dtotal / 1000}, # TB "temperatura": self._temperatura(), "carga": [float(x) for x in _lee("/proc/loadavg", "0 0 0").split()[:3]], "uptime": float(arriba) if arriba else 0, # segundos "procesos": self._procesos(), } g = self._grafica() if g: datos["gpu"] = {"uso": g["uso"], "vram_usada": g["usada"] / 1024, # GB "vram_total": g["total"] / 1024} red = self._red() if red: datos["red"] = red bat = self._bateria() if bat is not None: datos["bateria"] = bat return datos class Panel: """El nucleo visto desde la cara.""" def __init__(self, cara, boca, manos, cerebro=None, estado_voz=None, candado=True, centinela=None): self.cara = cara self.boca = boca self.manos = manos self.cerebro = cerebro self.centinela = centinela # quien decide si hay que llamarle self.estado_voz = estado_voz # callable -> "escuchando"|"hablando"|... self.telemetria = Telemetria() self._parar = threading.Event() # Pausa a peticion del usuario (boton PARAR): mientras esta puesta, el # bucle de escucha ignora lo que oiga. Distinto de self._parar, que corta # los HILOS del panel al cerrar. Lo lee jarvis.py en cada frase. self.parado = False # Las tarjetas: una por orden en marcha, con su reloj y su boton de # parar. El prefijo va vacio porque aqui ya estamos en el host; dentro # del flatpak era `flatpak-spawn --host --watch-bus`. from manos.tareas import GestorTareas self.tareas = GestorTareas(al_cambiar=self._tarea_cambio, prefijo_host=[]) # El candado. Tapa el panel y, sobre todo, hace que no obedezca: sin # esto cualquiera que pase por delante del portatil abierto le puede # dar ordenes a una maquina con sudo. # # La contraseña es la del sistema, comprobada con `sudo -S -v`, que NO # ejecuta nada: solo valida. Asi no hay una contraseña propia que # guardar en ningun sitio, que es una cosa menos que se puede filtrar. self.candado = candado self.desbloqueado = not candado self._intentos = 0 # Lo que se ha preguntado y esta esperando respuesta hablada. # (accion, para_que, hasta_cuando). para_que es "argumento" o "confirmar". self.pendiente = None # Root a demanda: las manos avisan aqui cuando una accion necesita sudo # y no hay sesion. En vez de pedir la contraseña al arrancar, se pide # solo entonces, una vez, y se cachea. Ver _pide_root. self.manos.al_pedir_root = self._pide_root # -------------------------------------------------------------- empujar def arranca(self): threading.Thread(target=self._bucle_telemetria, daemon=True).start() if self.candado and not self.desbloqueado: self.cara.manda("bloqueo", {"mensaje": "Introduzca la contraseña"}) self.manda_catalogo() self.manda_controles() self.cara.manda("voz", {"nombre": self.boca.voz, "id": self.boca.voz}) if self.cerebro: self._manda_cerebro(self.cerebro.modelo) # Cuando JARVIS elige otro modelo, el rotulo del boton "cerebro" se # actualiza solo: si no, dice uno y piensa con otro. self.cerebro.al_cambiar_modelo = self._manda_cerebro def _manda_cerebro(self, modelo): corto = modelo.partition(":") corto = corto[0] if corto[2] in ("", "latest") else corto[2] self.cara.manda("cerebro", {"nombre": corto, "id": modelo, "detalle": modelo, "fuera": False}) def _bucle_telemetria(self): tic = 0 while not self._parar.is_set(): # Sin nadie mirando no se gasta un nvidia-smi por segundo. if self.cara.conectados: try: self.cara.manda("telemetria", self.telemetria.lee()) except Exception: pass # Las medidas de JARVIS (latencia, reparto, RAG) cambian mucho # mas despacio que la CPU: se recalculan cada 5 s, no cada uno. # Salen del diario, que es leer un fichero, no del sistema. if tic % 5 == 0: try: from manos.medidas import lee_medidas self.cara.manda("medidas", lee_medidas()) except Exception: pass tic += 1 self._parar.wait(1.0) def manda_catalogo(self): """Lo que sabe hacer, agrupado, para la pestaña de la derecha.""" grupos = {} for a in self.manos.catalogo.acciones: if not a.frases: continue grupos.setdefault(a.grupo or "otras", []).append({ "id": a.id, "frase": a.frases[0], "argumento": a.captura, "confirma": a.confirmar, }) datos = [{"grupo": g, "acciones": v} for g, v in sorted(grupos.items(), key=lambda x: -len(x[1]))] self.cara.manda("catalogo", datos) def _tarea_cambio(self, tarea): """Llamado desde el hilo de la tarea cada vez que hay algo que enseñar.""" try: self.cara.manda("tarea", tarea.resumen()) except Exception: pass def lanza(self, accion, argumento=""): """Ejecuta una accion como TAREA, con tarjeta en el panel. Se usa para lo que tarda. Lo instantaneo —el disco, la hora— no merece una tarjeta: aparece, parpadea y se va antes de que nadie la lea. """ return self.tareas.lanza(accion, argumento) def manda_controles(self): """El estado real de los interruptores, para que el rotulo no mienta. Los botones NO se pintan al pulsarlos: se manda la orden y se espera a que el nucleo conteste con como quedo. Asi el boton nunca dice "despierto" con el centinela dormido. """ self.cara.manda("controles", { # encendido = atiende todo; apagado = solo si le llaman "sueno": not (self.centinela and self.centinela.activo), # encendido = escuchando; apagado = en pausa (boton PARAR). Sin este # campo el HUD no sabe alternar el boton a "reanudar" ni marcar el # documento como "parado", y el boton parece que no hace nada. "parar": not self.parado, }) def estado(self, cual: str): self.cara.manda("estado", cual) def dicho(self, texto: str, atendido=True, motivo=""): """Lo que se ha oido, SIEMPRE, tambien cuando se descarta. Esto es lo que faltaba y por lo que el panel parecia muerto: se oia, se transcribia, y si no llevaba la palabra de activacion se tiraba en silencio. Desde la pantalla eso es indistinguible de un microfono estropeado, y no hay forma de saber si el problema es que no te oye, que te oye mal, o que te oye bien y no te hace caso. El diccionario y no la cadena pelada: el HUD lee datos.texto. """ self.cara.manda("dicho", {"texto": texto, "atendido": atendido, "motivo": motivo}) def respuesta(self, texto: str): self.cara.manda("respuesta", {"texto": texto}) def para(self): self._parar.set() try: self.tareas.para_todas() except Exception: pass def cerrar(self): """Apaga JARVIS entero: la escucha, la voz, whisper-server y el proceso. Cerrar la pestaña o pulsar «cerrar» tiene que soltar de verdad el microfono y la grafica, no dejar el proceso escuchando en segundo plano. Se manda SIGINT al propio proceso: eso dispara el mismo apagado que Ctrl+C —el finally del bucle de escucha para whisper, y atexit lo remata— asi que se reutiliza la limpieza que ya estaba probada, sin duplicarla. Corre desde un hilo del servidor; la señal la atiende el hilo principal. """ self.parado = True # que la escucha no atienda nada mas ya self.para() # corta los hilos y las tareas del panel os.kill(os.getpid(), signal.SIGINT) # --------------------------------------------------------------- pulsar def pulsado(self, canal: str, dato: str): """Lo que llega del panel. Corre en su propio hilo, puede tardar.""" if canal == "listo": self.arranca() return if canal != "accion": return # Desbloquear es lo unico que se atiende con el candado puesto. if dato.startswith("desbloquear:"): self._intenta_desbloquear(dato.split(":", 1)[1]) return # Todo lo demas son ORDENES, y con el candado puesto no hay ordenes. if self.candado and not self.desbloqueado: self.cara.manda("bloqueo", {"mensaje": "Introduzca la contraseña"}) return if dato == "voz:siguiente": nueva = self.boca.siguiente() self.cara.manda("voz", {"nombre": nueva, "id": nueva}) self.boca.di("Buenas noches, señor.") elif dato == "prueba": threading.Thread(target=self._tanda_de_prueba, daemon=True).start() elif dato.startswith("explica:"): self._explica(dato.split(":", 1)[1]) elif dato.startswith("parar_tarea:"): self.tareas.para(dato.split(":", 1)[1]) elif dato.startswith("cerrar_tarea:"): ident = dato.split(":", 1)[1] self.tareas.cierra(ident) self.cara.manda("tarea_fuera", {"id": ident}) elif dato == "sueno": # Cambia si hace falta llamarle por su nombre. # # Por defecto NO hace falta: medido sobre 412 frases reales, el # ruido ambiente no dispara nada, asi que exigir el nombre en cada # frase era peaje sin contrapartida. El interruptor se queda para # cuando haya una tele cerca. if not self.centinela: return self.centinela.activo = not self.centinela.activo if self.centinela.activo: self.centinela.duerme() self.boca.di_si_libre("Llameme por mi nombre, señor.") else: self.boca.di_si_libre("Le escucho, señor.") self.manda_controles() self.estado("escuchando" if not self.centinela.activo else "dormido") elif dato == "cerrar": # Apagar JARVIS del todo. Se responde primero al HUD para que la # ventana sepa cerrarse, y luego se mata el proceso. self.cara.manda("cerrando", True) self.cerrar() return elif dato.startswith("parar"): # El boton alterna PAUSA. No es solo "cortar lo de ahora": el titulo # promete "cierra el microfono y corta la voz; pulsa otra vez para # reanudar", asi que tiene que ALTERNAR y avisar al HUD de como quedo, # que es lo que hacia que pareciera que no funcionaba. self.parado = not self.parado if self.parado: # cortar lo que este corriendo; la escucha la ignora jarvis.py # mirando self.parado en cada frase. Pausa en silencio: si el # usuario pulsa PARAR es para que se calle, no para que hable. for t in self.tareas.todas(): if t.estado == "corriendo": self.tareas.para(t.id) self.estado("parado") else: self.boca.di_si_libre("A sus órdenes, señor.") self.estado("escuchando") # el HUD alterna el boton y marca el documento SOLO con esto self.manda_controles() # Cuanto se espera la respuesta a una pregunta antes de darla por olvidada. # Corto cansa —hay que pensar que fichero era—; largo hace que una frase # suelta media conversacion despues se tome por la respuesta. ESPERA_S = 45 def _explica(self, ident: str): """Pulsar una accion PREGUNTA lo que falta y la hace. No recita la frase que la dispara. "Diga «busca el fichero» y a continuacion lo que busca" es un manual de instrucciones, y uno pulsa justo porque no quiere leerse el manual. espera argumento -> pregunta cual y lo ejecuta con lo que digas no tiene vuelta -> pide confirmacion antes ninguna cosa -> la hace, que es a lo que has pulsado Lo destructivo sigue sin dispararse de un clic: pasa por un si o un no hablado, que es la misma red que tenia antes pero sin obligar a leer. """ accion = next((a for a in self.manos.catalogo.acciones if a.id == ident), None) if accion is None: return if accion.captura: self._pregunta(accion, "argumento", pregunta_por(accion)) elif accion.confirmar: self._pregunta(accion, "confirmar", confirmacion(accion)) else: self.tareas.lanza(accion, "") def _pregunta(self, accion, para_que, texto, argumento=""): self._argumento_pendiente = argumento self.pendiente = (accion, para_que, time.monotonic() + self.ESPERA_S, getattr(self, "_argumento_pendiente", "")) self.cara.manda("preguntando", {"texto": texto, "accion": accion.id}) self.cara.manda("respuesta", {"texto": texto}) self.boca.di(texto) def esperando(self): """La pregunta en el aire, o None si no hay o ya caduco.""" if not self.pendiente: return None if time.monotonic() > self.pendiente[2]: self.olvida() return None return self.pendiente def olvida(self): self.pendiente = None self.cara.manda("preguntando", {"texto": ""}) def responde(self, texto: str) -> bool: """Contesta a lo preguntado. Devuelve si consumio la frase. Se llama desde el bucle de voz ANTES del emparejador: mientras hay una pregunta en el aire, lo que se dice es la respuesta, no una orden nueva. """ pend = self.esperando() if not pend: return False accion, para_que, _, argumento = pend if para_que == "confirmar": respuesta = es_si(texto) if respuesta is None: self.boca.di("No le he entendido. ¿Si o no, señor?") return True self.olvida() if respuesta: self.tareas.lanza(accion, argumento) else: self.boca.di("De acuerdo, lo dejo.") return True # argumento: cualquier cosa que no sea una cancelacion vale if es_si(texto) is False: self.olvida() self.boca.di("De acuerdo, lo dejo.") return True self.olvida() arg = texto.strip() # Si ademas no tiene vuelta atras, se confirma AHORA que ya se sabe # sobre que. Hay siete acciones que piden argumento y confirmacion # —"mata el proceso", "que puertos tiene abiertos"— y preguntando solo # el argumento se saltaban la red entera. Y confirmar antes de saber el # argumento seria pedir un cheque en blanco: "¿mato el proceso?" "¿cual?" if accion.confirmar: self._pregunta(accion, "confirmar", f"«{accion.frases[0]} {arg}». Esto no tiene vuelta " f"atras. ¿Confirma, señor?", argumento=arg) return True self.tareas.lanza(accion, arg) return True def _pide_root(self, comando: str = ""): """Muestra el dialogo de contraseña porque una accion necesita root. No es el candado del arranque —eso pedia la contraseña de primeras y echaba para atras—: esto es a demanda, la PRIMERA vez que hace falta sudo en la sesion. La contraseña que se teclee la valida y la cachea _intenta_desbloquear (por el mismo canal "desbloquear"), asi que a partir de ahi el root ya esta abierto y no se vuelve a pedir hasta que caduque. Corre desde el hilo de las manos, asi que solo manda al HUD. """ que = (comando or "").strip()[:60] self.cara.manda("bloqueo", { "mensaje": f"Contraseña para: {que}" if que else "Contraseña de administrador", "root": True}) def _intenta_desbloquear(self, clave: str): if not clave: return # La misma GestorRoot de las manos: si la contraseña es buena, se queda # ahi y el cerebro puede hacer sudo durante la sesion. Desbloquear el # panel y abrir el root son el mismo gesto, con la misma contraseña. try: valida = self.manos.root.valida(clave) except Exception as e: self.cara.manda("bloqueo", {"mensaje": f"No se pudo comprobar: {e}", "error": True}) return if not valida: self._intentos += 1 # Sin bloqueo por intentos ni espera creciente: la contraseña es la # del sistema y quien la teclea esta sentado delante del portatil. # Contarlos sirve para que el mensaje sea util, no para castigar. self.cara.manda("bloqueo", { "mensaje": "No es esa" + (f" ({self._intentos} intentos)" if self._intentos > 1 else ""), "error": True}) return self._intentos = 0 self.desbloqueado = True self.manos.root.recuerda(clave) # root disponible para la sesion self.cara.manda("desbloqueado", True) self.boca.di_si_libre("A su servicio, señor.") def bloquea(self): """Echa el candado, aunque se arrancara sin el. Que `--sin-candado` sea un arranque comodo no significa que uno no pueda querer cerrarlo al levantarse de la silla. Por eso esto ACTIVA el candado en vez de comprobar si estaba activo: negarse a cerrar porque no estaba puesto seria justo lo contrario de lo que se pide. """ self.candado = True self.desbloqueado = False self._intentos = 0 self.manos.root.olvida() # y se cierra el root con el candado self.cara.manda("bloqueo", {"mensaje": "Introduzca la contraseña"}) # Elegidas para que se vea el ciclo entero: una que acaba al instante, dos # que tardan un poco, y una larga a la que da tiempo a pulsarle el boton de # parar. Todas de solo lectura. # Cada una comprobada contra el catalogo: una que no encaje no lanza # tarjeta y la tanda queda coja sin decir por que. PRUEBA = ["cuanto espacio queda", "cuanta memoria", "que temperatura hay", "que esta consumiendo", "cuantos procesos hay"] def _tanda_de_prueba(self): """Lanza varias a la vez para ver las tarjetas trabajando. Van como TAREAS y no ejecutadas a pelo: el objetivo es justo ver el ciclo entero —aparecer, contar segundos, terminar— y poder pulsarle el boton de parar a la que tarda. """ for orden in self.PRUEBA: accion, argumento = self.manos.busca(orden) if accion is None: continue self.tareas.lanza(accion, argumento) time.sleep(0.25)