Labstream/labstream-organizador
1
0
Fork 0
Código Incidencias Pull Requests Paquetes Proyectos Lanzamientos Wiki Actividad

3.9.0: la cola más rápida: la tarjeta descomprime, dos clips a la vez, y las conversiones sueltas se cierran

Explorar el Código 
Medido en la RTX 5080 del estudio con la cola de Vía Sumapaz (1.920 clips, unas 26 horas): el codificador iba al
46-74 % y el descompresor al 0 %. Con NVDEC un H.264 4K pasa de 0,54× a 0,86× tiempo real; dos a la vez salen a
1,29× entre los dos. Además: al arrancar se cierran las conversiones que quedaron sin su programa (una de las
10:04 seguía colgada a las 17:00), «Detener» cierra el ffmpeg del lector de BRAW, la cola se pide entera por
carpeta (antes, los 500 últimos de todas), y lo encolado que ya no compensa con la regla del 30 % no se convierte.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Este commit está contenido en:
Labstream
2026-10-14 17:32:01 -05:00
padre c91473aea4
commit 00b84b4aa7
Se han modificado 4 ficheros con 505 adiciones y 126 borrados
Descargar archivo de parche Descargar archivo de diferencias Expandir todos los archivos Contraer todos los archivos

152
prueba_orden.py
Ver fichero

@@ -12,6 +12,7 @@ import time
import shutil
import tempfile
import threading
import subprocess
import unittest
AQUI = os.path.dirname(os.path.abspath(__file__))
@@ -1315,6 +1316,157 @@ class Convertidor(unittest.TestCase):
self.assertTrue(r["ok"], r)
self.assertTrue(os.path.isfile(os.path.join(nueva, os.path.basename(tb["salida"]))))
def _cola_falsa(self, durante=0.4):
"""Sustituye la conversión por una que tarda `durante` segundos y cuenta cuántas van a la vez."""
cuenta = {"vivos": 0, "maximo": 0, "vistos": []}
candado = threading.Lock()
def falso(t):
with candado:
cuenta["vivos"] += 1
cuenta["maximo"] = max(cuenta["maximo"], cuenta["vivos"])
cuenta["vistos"].append(t["id"])
fin = time.time() + durante
while time.time() < fin and not O.COLA["cancelar"]:
time.sleep(0.02)
with candado:
cuenta["vivos"] -= 1
if O.COLA["cancelar"]:
O.set_trabajo(t["id"], estado="cancelado", pct=0)
return
with open(t["salida"], "wb") as fh:
fh.write(b"x" * 1000)
O.set_trabajo(t["id"], estado="listo", pct=100, bytes_out=1000, fin=time.time())
for k, v in {"correr_ffmpeg": falso, "_verificar_listos": lambda ids: None, "editando": lambda: "",
"_ya_no_compensa": lambda t: ""}.items():
self.addCleanup(setattr, O, k, getattr(O, k))
setattr(O, k, v)
self.addCleanup(O.CONFIG["optimizar"].pop, "simultaneas", None)
return cuenta
def test_dos_a_la_vez(self):
"""RTX 5080 del estudio, 14-oct-2026: con un clip a la vez el codificador iba al 46-74 %; dos a la vez salen
a 1,29× tiempo real entre los dos, frente a 0,77×. Nunca van más de los pedidos, ningún clip se convierte
dos veces, y «Detener» para a todos."""
raiz = cliente("ALaVez")
nr = os.path.normpath(raiz)
g = "2025/11- Noviembre/00_Grabacion/SONY/"
clips = [mk(raiz, g + f"C00{i:02d}.MP4", 2025, 11, 11, tam=300000 + i) for i in range(8)]
O.escanear(raiz)
self.leidos(raiz, {os.path.basename(c): {"codec": "h264", "perfil": "High", "kbps": 120000} for c in clips})
cuenta = self._cola_falsa()
O.CONFIG["optimizar"]["simultaneas"] = 2
# lo que otras pruebas dejaran esperando no entra en esta tanda (y se devuelve al acabar)
con = O.db()
otros = [r[0] for r in con.execute("SELECT id FROM trabajos WHERE estado = 'pendiente'")]
with con:
con.execute("UPDATE trabajos SET estado = 'cancelado' WHERE estado = 'pendiente'")
con.close()
def devolver():
c = O.db()
with c:
c.executemany("UPDATE trabajos SET estado = 'pendiente' WHERE id = ?", [(i,) for i in otros])
c.close()
self.addCleanup(devolver)
ids = O.encolar(clips[:5], "equilibrado", "420", "rec709", nr)
self.assertEqual(len(ids), 5)
self.assertTrue(O.iniciar_cola())
time.sleep(0.2)
e = O.eta_cola()
self.assertEqual(e["paralelo"], 2)
self.assertEqual(len(e["actuales"]), 2)
O.COLA["hilo"].join(30)
self.assertEqual(cuenta["maximo"], 2)
self.assertEqual(sorted(cuenta["vistos"]), sorted(ids))
self.assertEqual({self.trabajo(i)["estado"] for i in ids}, {"listo"})
self.assertEqual((O.COLA["actuales"], O.COLA["activa"], O.COLA["actual"]), ({}, False, None))
# Detener: los dos que van se cortan, los que esperan se quedan esperando
cuenta.update(maximo=0, vistos=[])
ids = O.encolar(clips[5:], "equilibrado", "420", "rec709", nr)
self.assertTrue(O.iniciar_cola())
time.sleep(0.2)
O.COLA["cancelar"], O.COLA["activa"] = True, False # lo que hace «Detener»
O.COLA["hilo"].join(30)
estados = sorted(self.trabajo(i)["estado"] for i in ids)
self.assertEqual(estados, ["cancelado", "cancelado", "pendiente"])
self.assertFalse(O.COLA["cancelar"])
con = O.db()
with con:
con.execute(f"DELETE FROM trabajos WHERE id IN ({','.join('?' * len(ids))})", ids)
con.close()
def test_conversiones_sueltas(self):
"""14-oct-2026: el lector de BRAW y el ffmpeg de una conversión de antes de actualizar seguían colgados siete
horas después. Al arrancar se cierran los que no tienen quien los espere, y no los de un programa vivo."""
base = r"C:\Programas\OrganiLAB"
P = lambda pid, ppid, nombre, cmd: {"ProcessId": pid, "ParentProcessId": ppid, "Name": nombre, "CommandLine": cmd}
ff = base + r"\ffmpeg\ffmpeg.exe"
procesos = [
P(100, 50, "OrganizadorLabstream.exe", f'"{base}\\OrganizadorLabstream.exe"'), # este programa
P(101, 100, "OrganizadorLabstream.exe", f"{base}\\OrganizadorLabstream.exe {base}\\braw.py convertir x.json"), # su lector
P(102, 101, "ffmpeg.exe", f'{ff} -i pipe:0 "V:\\G\\a_10bit.mp4.parcial.mp4"'), # el ffmpeg del lector
P(103, 100, "ffmpeg.exe", f"{ff} -i a.mp4 b.parcial.mp4"), # uno suyo
P(200, 65492, "OrganizadorLabstream.exe", f"{base}\\OrganizadorLabstream.exe {base}\\braw.py convertir y.json"), # lector suelto
P(201, 200, "ffmpeg.exe", f"{ff} -i pipe:0 V:\\G\\b.parcial.mp4"), # y su ffmpeg
P(300, 60, "OrganizadorLabstream.exe", f'"{base}\\OrganizadorLabstream.exe" --automatico'), # la pasada de la noche
P(301, 300, "ffmpeg.exe", f"{ff} -i c.mp4 c.parcial.mp4"), # y su ffmpeg
P(401, 4000, "ffmpeg.exe", f"{ff} -i d.mp4 d.parcial.mp4"), # su programa ya no está
P(500, 1, "ffmpeg.exe", r"C:\otro\ffmpeg.exe -i e.mp4 e.mp4"), # de otro programa
]
self.assertEqual(sorted(p["ProcessId"] for p in O._conversores_sueltos(procesos, base, 100)), [200, 401])
def test_la_tarjeta_descomprime(self):
"""Con la cola en marcha, el descompresor de la tarjeta estaba al 0 %: un H.264 4K pasa de 0,54× a 0,86× tiempo
real si lo descomprime ella. Lo que ella no lee (ProRes) sigue por el procesador, y si con ella falla, se repite."""
real = O.motores
self.addCleanup(setattr, O, "motores", real)
O.motores = lambda: {"gpu": {"nvdec": True, "hevc": True}}
self.assertEqual(O._decodificar_con_tarjeta({"codec": "h264"}, "gpu"), ["-hwaccel", "cuda"])
self.assertEqual(O._decodificar_con_tarjeta({"codec": "hevc"}, "gpu"), ["-hwaccel", "cuda"])
self.assertEqual(O._decodificar_con_tarjeta({"codec": "prores"}, "gpu"), [])
self.assertEqual(O._decodificar_con_tarjeta({"codec": "h264"}, "cpu"), [])
O.motores = lambda: {"gpu": {"nvdec": False}}
self.assertEqual(O._decodificar_con_tarjeta({"codec": "h264"}, "gpu"), [])
O.motores = real
ff = shutil.which("ffmpeg")
if not ff:
self.skipTest("sin ffmpeg")
raiz = cliente("Descomprime")
nr = os.path.normpath(raiz)
clip = os.path.join(nr, "2025", "11- Noviembre", "00_Grabacion", "SONY", "C0001.MP4")
os.makedirs(os.path.dirname(clip))
subprocess.run([ff, "-hide_banner", "-loglevel", "error", "-f", "lavfi", "-i", "testsrc=size=320x180:rate=30:duration=1",
"-c:v", "libx264", "-pix_fmt", "yuv420p", clip], check=True)
O.escanear(raiz)
self.leidos(raiz, {"C0001.MP4": {"codec": "h264", "perfil": "High", "ancho": 320, "alto": 180, "fps": 30, "dur": 1, "kbps": 900}})
id_ = O.encolar([clip], "equilibrado", "420", "rec709", nr)[0]
# una descompresión que no arranca: se repite con el procesador y sale igual
self.addCleanup(setattr, O, "_decodificar_con_tarjeta", O._decodificar_con_tarjeta)
O._decodificar_con_tarjeta = lambda f, motor: ["-hwaccel", "no_existe"]
O.correr_ffmpeg(self.trabajo(id_))
t = self.trabajo(id_)
self.assertEqual(t["estado"], "listo", t["log"])
self.assertTrue(os.path.isfile(t["salida"]))
self.assertNotIn("-hwaccel", t["log"])
def test_lo_que_ya_no_compensa_no_se_convierte(self):
"""Se encolaron con la regla de antes: un H.265 de cámara que se quedaría en el 93 % no se convierte."""
raiz = cliente("YaNoCompensa")
nr = os.path.normpath(raiz)
g = "2025/11- Noviembre/00_Grabacion/SONY/"
a, b = mk(raiz, g + "C0001.MP4", 2025, 11, 11, tam=300000), mk(raiz, g + "C0002.MP4", 2025, 11, 11, tam=300001)
O.escanear(raiz)
self.leidos(raiz, {"C0001.MP4": {"codec": "hevc", "perfil": "Main 10", "pix_fmt": "yuv420p10le", "fps": 59.94, "kbps": 72000},
"C0002.MP4": {"codec": "h264", "perfil": "High", "kbps": 120000}})
ia, ib = O.encolar([a, b], "equilibrado", "420", "rec709", nr)
self.assertTrue(O._ya_no_compensa(self.trabajo(ia)).startswith("convertido ahorraría solo un 7 %"))
self.assertEqual(O._ya_no_compensa(self.trabajo(ib)), "")
con = O.db()
with con:
con.execute("DELETE FROM trabajos WHERE id IN (?, ?)", (ia, ib))
con.close()
def test_el_codificador_que_toca(self):
fm = O.FORMATOS["h265_10bit"]
t = {"chroma": "420", "perfil": "equilibrado", "kbps": 30000}