hacklab/JARVIS
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
JARVIS/nucleo/cara/panel.py
sito 3a2a1bb111 Root a demanda, lanzador de escritorio, captura nueva y ajustes 
- Root: no se pide la contraseña al arrancar (echaba para atras). Se pide
  solo la primera vez que una accion necesita sudo, y se cachea. --candado
  para el modo seguro. Aviso de root prominente en el README.
- install.sh crea un lanzador de escritorio (jarvis.desktop) con el icono
  naranja: aparece en el menu de aplicaciones para arrancar con un clic.
- Captura del panel reemplazada por una mas limpia (solo la ventana).
- Quitado "Probado en ..." del README (informacion del equipo).
- ventanas.sh: aviso claro segun el compositor en Wayland.
2026-08-16 16:20:27 +02:00

616 lines
25 KiB
Python
Raw Blame History

"""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 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.cara.manda("cerebro", {"nombre": "local", "id": "local",
"detalle": self.cerebro.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
# --------------------------------------------------------------- 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.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)
Reference in a new issue View git blame Copy permalink