3.4.0: los BRAW se convierten sin Resolve, y lo que no compensa no se reconvierte
Munka no podía reconvertir archivos que antes sí. Auditado sobre la carpeta de Andrea Galeano: - Los BRAW nunca habían salido: 87 clips (496 GB), tres tandas por el guion de Resolve y ni un clip convertido. Ahora los lee el decodificador de Blackmagic que instala Resolve (braw.py, en un proceso aparte), sin abrir Resolve, y ffmpeg los codifica como cualquier otro clip. Probado con clips del NAS: 15,6 GB → 394 MB con sus 7.004 cuadros, VMAF 96,0 contra el BRAW revelado sin pérdida. Las conversiones de BRAW que fallaron por Resolve pasan solas al lector. - Lo que «antes podía» eran reconversiones de lo ya convertido: ahorraban un 5 % con pérdida. Siguen sin ofrecerse, y lo que ya viene comprimido de cámara solo se ofrece si compensa. - La tarjeta codifica con «uhq»: entre un 19 y un 55 % menos de peso con la misma calidad medida, siempre H.265 de 10 bits. - Una conversión que no aligera (más del 80 % del clip, o reconversión de una conversión) no reemplaza al original: queda «no compensa» y lo que salió va a _Para borrar\descartados. De paso: los MP4 guardan las etiquetas de la cámara; un intento que no pasó la verificación se puede reintentar; un H.264 de más de 4.096 px de lado va por CPU; el motivo de un fallo se lee entero; el guion de Resolve llega con tildes y borra su ficha. Pruebas: prueba_orden.py 33 (una con un BRAW de verdad), probar.py 19, prueba_funcional.py 30, prueba_ventana.py 18 desde el código y 17 con el paquete. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Este commit está contenido en:
Labstream
429
prueba_orden.py
429
prueba_orden.py
@@ -11,6 +11,7 @@ import json
|
||||
import time
|
||||
import shutil
|
||||
import tempfile
|
||||
import threading
|
||||
import unittest
|
||||
|
||||
AQUI = os.path.dirname(os.path.abspath(__file__))
|
||||
@@ -18,7 +19,7 @@ TMP = tempfile.mkdtemp(prefix="organilab-motor-")
|
||||
APP = os.path.join(TMP, "app")
|
||||
os.makedirs(os.path.join(APP, "datos"))
|
||||
for _f in ("organizador.py", "reglas.py", "asistente.py", "actualizador.py", "firma.py", "instalar.py", "ventana.py",
|
||||
"resolve_bridge.py"):
|
||||
"resolve_bridge.py", "braw.py"):
|
||||
shutil.copy(os.path.join(AQUI, _f), APP)
|
||||
# un diccionario como el que dejaba la 3.1.1: con una carpeta de la propia app aprendida como grabación
|
||||
with open(os.path.join(APP, "datos", "diccionario.json"), "w", encoding="utf-8") as _fh:
|
||||
@@ -601,6 +602,432 @@ class Nuevo(unittest.TestCase):
|
||||
O.CONFIG["automatico"] = ajuste
|
||||
|
||||
|
||||
# el lector de BRAW de mentira: habla como braw.py (una línea JSON por evento) sin decodificar nada
|
||||
LECTOR_FALSO = """# -*- coding: utf-8 -*-
|
||||
import os, sys, json, time
|
||||
t = json.load(open(sys.argv[2], encoding="utf-8"))
|
||||
modo = os.environ.get("BRAW_FALSO", "bien")
|
||||
def di(**kw):
|
||||
print(json.dumps(kw), flush=True)
|
||||
di(msg="2160x4096 · 30 fps · 60 cuadros · curva de prueba")
|
||||
if modo == "falla":
|
||||
di(estado="error", log="el decodificador no pudo abrir el clip (prueba)")
|
||||
sys.exit(1)
|
||||
with open(t["tmp"], "wb") as fh:
|
||||
fh.write(b"x" * 200000)
|
||||
if modo == "lento":
|
||||
for i in range(300):
|
||||
di(pct=i / 3.0)
|
||||
if os.path.exists(t["cancelar"]):
|
||||
os.remove(t["tmp"])
|
||||
di(estado="cancelado")
|
||||
sys.exit(1)
|
||||
time.sleep(0.1)
|
||||
di(pct=50.0)
|
||||
di(estado="listo", pct=100, segundos=0.1)
|
||||
"""
|
||||
|
||||
|
||||
class Convertidor(unittest.TestCase):
|
||||
"""El convertidor, medido el 5-oct-2026 sobre la carpeta de Andrea Galeano: los BRAW no habían salido
|
||||
nunca (87 clips, 462 GB, tres tandas fallidas por la vía de Resolve) y lo que ya venía bien comprimido
|
||||
se reconvertía para ahorrar un 5 %, con una generación de pérdida encima."""
|
||||
|
||||
MOTORES = {"ffmpeg": True, "gpu": {"nombre": "", "hevc": False, "h264": False, "chroma422": False, "uhq": False},
|
||||
"cpu": {"libx265": True, "libx264": True, "prores_ks": True}, "formatos": {k: "cpu" for k in O.FORMATOS}, "chroma422": True}
|
||||
|
||||
def setUp(self):
|
||||
antes = (O.decodificador_braw, O.motores, O.medir_braw, O.ffprobe)
|
||||
|
||||
def devolver():
|
||||
O.decodificador_braw, O.motores, O.medir_braw, O.ffprobe = antes
|
||||
O.COLA["cancelar"] = False
|
||||
os.environ.pop("BRAW_FALSO", None)
|
||||
self.de_verdad = antes
|
||||
self.addCleanup(devolver)
|
||||
O.decodificador_braw = lambda: {"ok": True, "carpeta": "X:\\decodificador"}
|
||||
O.motores = lambda: self.MOTORES
|
||||
|
||||
def leidos(self, raiz, datos):
|
||||
"""Apunta en el inventario lo que ffprobe habría leído de cada clip."""
|
||||
con = O.db()
|
||||
with con:
|
||||
for nombre, d in datos.items():
|
||||
d = dict({"perfil": "", "pix_fmt": "yuv420p", "ancho": 3840, "alto": 2160, "fps": 30, "dur": 60, "acodec": "aac"}, **d)
|
||||
n = con.execute("UPDATE archivos SET analizado = 1, codec = ?, perfil = ?, pix_fmt = ?, ancho = ?, alto = ?, fps = ?, dur = ?, "
|
||||
"kbps = ?, acodec = ? WHERE raiz = ? AND nombre = ?",
|
||||
(d["codec"], d["perfil"], d["pix_fmt"], d["ancho"], d["alto"], d["fps"], d["dur"], d["kbps"], d["acodec"],
|
||||
os.path.normpath(raiz), nombre)).rowcount
|
||||
assert n == 1, nombre
|
||||
con.close()
|
||||
|
||||
def trabajo(self, id_):
|
||||
con = O.db()
|
||||
t = dict(con.execute("SELECT * FROM trabajos WHERE id = ?", (id_,)).fetchone())
|
||||
con.close()
|
||||
return t
|
||||
|
||||
def test_lo_ya_comprimido_no_se_reconvierte(self):
|
||||
"""9 clips que ya eran H.265 se reconvirtieron para quedarse en el 95 % de su peso."""
|
||||
raiz = cliente("Comprimidos")
|
||||
g = "2025/06- Junio/00_Grabacion/"
|
||||
for n in ("A014_06170738_C019.mp4", "C0001.MP4", "C0002.MP4", "C0003.MP4", "C0004.MP4", "C0005.MP4"):
|
||||
mk(raiz, g + n, 2025, 6, 17)
|
||||
O.escanear(raiz)
|
||||
self.leidos(raiz, {
|
||||
"A014_06170738_C019.mp4": {"codec": "h264", "perfil": "Main", "ancho": 4608, "alto": 2592, "kbps": 25000},
|
||||
"C0001.MP4": {"codec": "h264", "perfil": "High 4:2:2 Intra", "pix_fmt": "yuv422p10le", "fps": 59.94, "kbps": 596000},
|
||||
"C0002.MP4": {"codec": "hevc", "perfil": "Main 10", "pix_fmt": "yuv420p10le", "fps": 59.94, "kbps": 72000},
|
||||
"C0003.MP4": {"codec": "h264", "perfil": "High", "kbps": 120000},
|
||||
"C0004.MP4": {"codec": "av1", "perfil": "Main", "kbps": 30000},
|
||||
"C0005.MP4": {"codec": "prores", "perfil": "HQ", "pix_fmt": "yuv422p10le", "kbps": 870000},
|
||||
})
|
||||
nr = os.path.normpath(raiz)
|
||||
cs = {os.path.basename(c["ruta"]): c["excluido"] for c in O.candidatos(nr, 0, "equilibrado", 0)}
|
||||
self.assertTrue(cs["A014_06170738_C019.mp4"].startswith("ya viene comprimido (H.264 a 25 Mb/s)"), cs) # por debajo del objetivo (30)
|
||||
self.assertTrue(cs["C0002.MP4"].startswith("ya viene comprimido (H.265 a 72 Mb/s)"), cs)
|
||||
self.assertTrue(cs["C0004.MP4"].startswith("ya viene comprimido (AV1 a 30 Mb/s)"), cs)
|
||||
# lo que guarda cada cuadro entero (XAVC-I, ProRes) y el H.264 muy holgado sí adelgazan de verdad
|
||||
self.assertEqual([cs["C0001.MP4"], cs["C0003.MP4"], cs["C0005.MP4"]], ["", "", ""])
|
||||
# a un formato de edición no se le aplica: ahí no se busca que pese menos
|
||||
cs = {os.path.basename(c["ruta"]): c["excluido"] for c in O.candidatos(nr, 0, "equilibrado", 0, False, "prores_lt")}
|
||||
self.assertEqual(cs["A014_06170738_C019.mp4"], "")
|
||||
# y la lista de acciones cuenta solo lo que se puede aligerar
|
||||
ajustes = dict(O.CONFIG["optimizar"])
|
||||
self.addCleanup(lambda: O.CONFIG["optimizar"].update(ajustes))
|
||||
O.CONFIG["optimizar"].update(min_mb=0, meses_minimos=0)
|
||||
conv = [a for a in O.acciones(raiz)["acciones"] if a["clave"] == "convertir"]
|
||||
self.assertEqual(len(conv), 1)
|
||||
self.assertTrue(conv[0]["cifra"].startswith("3 clips"), conv[0])
|
||||
|
||||
def test_un_nombre_parecido_no_es_una_conversion(self):
|
||||
"""El sufijo solo no basta: «Entrevista_HQ.mp4» en H.264 no lo hizo este programa."""
|
||||
raiz = cliente("Parecidos")
|
||||
g = "2025/06- Junio/00_Grabacion/"
|
||||
for n in ("Entrevista_HQ.mp4", "C0001_10bit.mp4", "C0002_HQ.mov", "C0003_h264.mp4"):
|
||||
mk(raiz, g + n, 2025, 6, 17)
|
||||
O.escanear(raiz)
|
||||
self.leidos(raiz, {"Entrevista_HQ.mp4": {"codec": "h264", "perfil": "High", "kbps": 400000},
|
||||
"C0001_10bit.mp4": {"codec": "hevc", "perfil": "Main 10", "kbps": 400000},
|
||||
"C0002_HQ.mov": {"codec": "prores", "perfil": "HQ", "kbps": 870000},
|
||||
"C0003_h264.mp4": {"codec": "hevc", "perfil": "Main", "kbps": 400000}})
|
||||
cs = {os.path.basename(c["ruta"]): c["excluido"] for c in O.candidatos(os.path.normpath(raiz), 0, "equilibrado", 0, True)}
|
||||
self.assertEqual(cs["Entrevista_HQ.mp4"], "")
|
||||
self.assertEqual(cs["C0003_h264.mp4"], "") # dice _h264 y es H.265: tampoco es nuestro
|
||||
self.assertIn("ya es una conversión", cs["C0001_10bit.mp4"])
|
||||
self.assertIn("ya es una conversión", cs["C0002_HQ.mov"])
|
||||
|
||||
def test_los_braw_van_por_el_lector(self):
|
||||
raiz = cliente("Braws")
|
||||
g = "2025/12- Diciembre/00_Grabacion/URSA_BRAW/"
|
||||
a = mk(raiz, g + "A014_12091814_C001.braw", 2025, 12, 9)
|
||||
b = mk(raiz, g + "A014_12091819_C002.braw", 2025, 12, 9)
|
||||
O.escanear(raiz)
|
||||
como_ffprobe = {"codec": "braw", "pix_fmt": "", "ancho": 4640, "alto": 2624, "kbps": 560000, "acodec": "pcm_s24le"}
|
||||
self.leidos(raiz, {"A014_12091814_C001.braw": como_ffprobe, "A014_12091819_C002.braw": como_ffprobe})
|
||||
nr = os.path.normpath(raiz)
|
||||
cs = O.candidatos(nr, 0, "equilibrado", 0)
|
||||
self.assertEqual([(c["motor"], c["excluido"]) for c in cs], [("braw", ""), ("braw", "")])
|
||||
# con el lector, un BRAW sale a cualquier formato
|
||||
self.assertEqual([c["excluido"] for c in O.candidatos(nr, 0, "equilibrado", 0, False, "prores_lt")], ["", ""])
|
||||
# sin decodificador se intenta por Resolve, y por ahí solo a H.265 de 10 bits
|
||||
O.decodificador_braw = lambda: {"ok": False, "motivo": "no está"}
|
||||
self.assertEqual({c["motor"] for c in O.candidatos(nr, 0, "equilibrado", 0)}, {"resolve"})
|
||||
self.assertIn("Resolve", O.candidatos(nr, 0, "equilibrado", 0, False, "prores_lt")[0]["excluido"])
|
||||
self.assertEqual(O.encolar([a], "equilibrado", "420", "rec709", nr, "prores_lt"), [])
|
||||
viejo = O.encolar([a], "equilibrado", "420", "rec709", nr)
|
||||
self.assertEqual(self.trabajo(viejo[0])["motor"], "resolve")
|
||||
# así quedó la cola de verdad: cada clip fallido dos veces, y uno apuntando a donde ya no está
|
||||
con = O.db()
|
||||
with con:
|
||||
con.execute("UPDATE trabajos SET estado = 'error', log = 'Resolve está en la ventana del Gestor de proyectos' WHERE id = ?", (viejo[0],))
|
||||
for ruta in (a, os.path.join(nr, "Orden", "A014_12091814_C001.braw")):
|
||||
con.execute("INSERT INTO trabajos (ruta, salida, motor, perfil, chroma, color, kbps, estado, bytes_in, creado, raiz, formato) "
|
||||
"VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", (ruta, ruta[:-5] + "_10bit.mp4", "resolve", "equilibrado", "420", "rec709", 31047,
|
||||
"error", 1, time.time(), nr, "h265_10bit"))
|
||||
con.close()
|
||||
self.assertEqual(O.braws_al_lector(), 0) # sin decodificador no se toca nada
|
||||
O.decodificador_braw = lambda: {"ok": True, "carpeta": "X:\\decodificador"}
|
||||
self.assertEqual(O.braws_al_lector(), 1)
|
||||
con = O.db()
|
||||
filas = [tuple(r) for r in con.execute("SELECT ruta, motor, estado FROM trabajos WHERE raiz = ?", (nr,))]
|
||||
con.close()
|
||||
self.assertEqual(filas, [(a, "braw", "error")])
|
||||
# volver a encolar el clip sustituye el intento fallido en vez de dejar otro al lado
|
||||
nuevos = O.encolar([a, b], "equilibrado", "420", "rec709", nr)
|
||||
self.assertEqual(len(nuevos), 2)
|
||||
con = O.db()
|
||||
filas = sorted(tuple(r) for r in con.execute("SELECT ruta, motor, estado FROM trabajos WHERE raiz = ?", (nr,)))
|
||||
con.execute("DELETE FROM trabajos WHERE raiz = ?", (nr,))
|
||||
con.commit()
|
||||
con.close()
|
||||
self.assertEqual(filas, [(a, "braw", "pendiente"), (b, "braw", "pendiente")])
|
||||
|
||||
def test_un_braw_pasa_por_la_cola(self):
|
||||
"""La cola con un lector de mentira: lo que importa aquí es lo que hace el programa con lo que el lector dice."""
|
||||
raiz = cliente("ColaBraw")
|
||||
g = "2025/03- Marzo/00_Grabacion/URSA_BRAW/"
|
||||
clips = [mk(raiz, g + f"A002_0311042{i}_C04{i}.braw", 2025, 3, 11, tam=1_000_000) for i in (4, 5, 6)]
|
||||
mk(raiz, g + "A002_03110424_C044.sidecar", 2025, 3, 11)
|
||||
O.escanear(raiz)
|
||||
self.leidos(raiz, {os.path.basename(c): {"codec": "braw", "pix_fmt": "", "ancho": 4128, "alto": 2176, "kbps": 407000, "dur": 2.0,
|
||||
"acodec": "pcm_s24le"} for c in clips})
|
||||
nr = os.path.normpath(raiz)
|
||||
# el decodificador dice el tamaño de verdad: el clip es vertical, y ffprobe lo leyó tumbado y con relleno
|
||||
O.medir_braw = lambda ruta: {"ok": True, "ancho": 2160, "alto": 4096, "fps": 30.0, "cuadros": 60, "dur": 2.0, "tc": "04:27:26:19",
|
||||
"camara": "Blackmagic Pocket Cinema Camera 4K", "gamma": "Blackmagic Design Film", "gamut": "Blackmagic Design", "gen": 5}
|
||||
lector = os.path.join(APP, "braw.py")
|
||||
shutil.copy(lector, lector + ".de-verdad")
|
||||
self.addCleanup(lambda: os.replace(lector + ".de-verdad", lector))
|
||||
with open(lector, "w", encoding="utf-8") as fh:
|
||||
fh.write(LECTOR_FALSO)
|
||||
ids = O.encolar(clips, "equilibrado", "420", "rec709", nr)
|
||||
self.assertEqual(len(ids), 3)
|
||||
|
||||
# 1) sale bien
|
||||
O.correr_braw(self.trabajo(ids[0]))
|
||||
t = self.trabajo(ids[0])
|
||||
self.assertEqual(t["estado"], "listo", t["log"])
|
||||
self.assertEqual(os.path.getsize(t["salida"]), 200000)
|
||||
self.assertIn("curva de prueba", t["log"])
|
||||
self.assertEqual(t["kbps"], O.kbps_objetivo(2160, 4096, 30.0, "equilibrado")) # la tasa, con el tamaño de verdad
|
||||
con = O.db()
|
||||
f = con.execute("SELECT ancho, alto, perfil FROM archivos WHERE ruta = ?", (clips[0],)).fetchone()
|
||||
con.close()
|
||||
self.assertEqual((f["ancho"], f["alto"], f["perfil"]), (2160, 4096, "Film"))
|
||||
self.assertEqual([x for x in os.listdir(O.DATOS) if x.startswith("braw_")], [])
|
||||
# la verificación compara con ese tamaño: una salida tumbada no pasa
|
||||
bien = {"codec": "hevc", "perfil": "Main 10", "pix_fmt": "yuv420p10le", "ancho": 2160, "alto": 4096, "fps": 30.0, "dur": 2.0, "kbps": 20000,
|
||||
"alfa": 0, "acodec": "aac", "pistas_audio": 1}
|
||||
O.ffprobe = lambda ruta: dict(bien, ancho=4128, alto=2176)
|
||||
v = O.verificar([ids[0]])[0]
|
||||
self.assertFalse(v["ok"])
|
||||
self.assertIn("2160×4096", v["motivo"])
|
||||
O.set_trabajo(ids[0], estado="listo")
|
||||
O.ffprobe = lambda ruta: dict(bien)
|
||||
self.assertTrue(O.verificar([ids[0]])[0]["ok"])
|
||||
# al reemplazar, el .sidecar del BRAW se va con él
|
||||
self.assertTrue(O.reemplazar([ids[0]])[0]["ok"])
|
||||
apartados = os.path.join(nr, R.PAPELERA, "originales", "2025", "03- Marzo", "00_Grabacion", "URSA_BRAW")
|
||||
self.assertEqual(sorted(os.listdir(apartados)), ["A002_03110424_C044.braw", "A002_03110424_C044.sidecar"])
|
||||
self.assertTrue(os.path.isfile(t["salida"]))
|
||||
|
||||
# 2) el lector dice que no puede: el motivo llega tal cual y no queda nada a medias
|
||||
os.environ["BRAW_FALSO"] = "falla"
|
||||
O.correr_braw(self.trabajo(ids[1]))
|
||||
t = self.trabajo(ids[1])
|
||||
self.assertEqual((t["estado"], t["log"]), ("error", "el decodificador no pudo abrir el clip (prueba)"))
|
||||
self.assertFalse(os.path.exists(t["salida"]) or os.path.exists(t["salida"] + ".parcial.mp4"))
|
||||
|
||||
# 3) Detener a mitad de un clip
|
||||
os.environ["BRAW_FALSO"] = "lento"
|
||||
hilo = threading.Thread(target=O.correr_braw, args=(self.trabajo(ids[2]),))
|
||||
hilo.start()
|
||||
fin = time.time() + 20
|
||||
while time.time() < fin and not (self.trabajo(ids[2])["pct"] or 0) > 0:
|
||||
time.sleep(0.1)
|
||||
O.COLA["cancelar"] = True
|
||||
hilo.join(40)
|
||||
self.assertFalse(hilo.is_alive())
|
||||
t = self.trabajo(ids[2])
|
||||
self.assertEqual(t["estado"], "cancelado")
|
||||
self.assertFalse(os.path.exists(t["salida"]) or os.path.exists(t["salida"] + ".parcial.mp4"))
|
||||
self.assertEqual([x for x in os.listdir(O.DATOS) if x.startswith("braw_")], [])
|
||||
|
||||
# 4) si ya hay un archivo con el nombre de la salida, no se pisa
|
||||
O.COLA["cancelar"] = False
|
||||
os.environ["BRAW_FALSO"] = "bien"
|
||||
O.set_trabajo(ids[1], estado="pendiente")
|
||||
with open(self.trabajo(ids[1])["salida"], "wb") as fh:
|
||||
fh.write(b"de otro")
|
||||
O.correr_braw(self.trabajo(ids[1]))
|
||||
t = self.trabajo(ids[1])
|
||||
self.assertEqual(t["estado"], "error")
|
||||
self.assertIn("no se pisa", t["log"])
|
||||
with open(t["salida"], "rb") as fh:
|
||||
self.assertEqual(fh.read(), b"de otro")
|
||||
|
||||
def test_lo_que_no_compensa_no_reemplaza(self):
|
||||
"""8 conversiones esperaban con «Reemplazar 8 verificados»: seis eran copias de una copia, al 85-99 % de su peso."""
|
||||
raiz = cliente("NoCompensa")
|
||||
g = "2025/11- Noviembre/00_Grabacion/SONY/"
|
||||
clips = [mk(raiz, g + n, 2025, 11, 11, tam=300000) for n in ("C0001_10bit.mp4", "C0002.MP4", "C0003.MP4")]
|
||||
O.escanear(raiz)
|
||||
nr = os.path.normpath(raiz)
|
||||
self.leidos(raiz, {"C0001_10bit.mp4": {"codec": "hevc", "perfil": "Main 10", "kbps": 72000},
|
||||
"C0002.MP4": {"codec": "h264", "perfil": "High", "kbps": 120000},
|
||||
"C0003.MP4": {"codec": "h264", "perfil": "High", "kbps": 120000}})
|
||||
ids = []
|
||||
con = O.db()
|
||||
with con:
|
||||
# una reconversión (93 %), un clip que apenas aligera (95 %) y uno que sí (40 %)
|
||||
for ruta, pesa in zip(clips, (280000, 285000, 120000)):
|
||||
salida = os.path.splitext(ruta)[0] + "_10bit.mp4"
|
||||
with open(salida, "wb") as fh:
|
||||
fh.write(b"x" * pesa)
|
||||
ids.append(con.execute("INSERT INTO trabajos (ruta, salida, motor, perfil, chroma, color, kbps, estado, bytes_in, creado, raiz, formato) "
|
||||
"VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", (ruta, salida, "ffmpeg", "equilibrado", "420", "rec709", 21150, "listo", 300000,
|
||||
time.time(), nr, "h265_10bit")).lastrowid)
|
||||
con.close()
|
||||
O.ffprobe = lambda ruta: {"codec": "hevc", "perfil": "Main 10", "pix_fmt": "yuv420p10le", "ancho": 3840, "alto": 2160, "fps": 30.0, "dur": 60.0,
|
||||
"kbps": 20000, "alfa": 0, "acodec": "aac", "pistas_audio": 1}
|
||||
v = O.verificar(ids)
|
||||
self.assertEqual([x["ok"] for x in v], [False, False, True])
|
||||
self.assertIn("93 %", v[0]["motivo"])
|
||||
self.assertIn("ya era una conversión", v[0]["motivo"])
|
||||
self.assertIn("95 %", v[1]["motivo"])
|
||||
self.assertEqual([self.trabajo(i)["estado"] for i in ids], ["descartado", "descartado", "verificado"])
|
||||
# lo descartado ni se reemplaza…
|
||||
r = O.reemplazar(ids)
|
||||
self.assertEqual([x["ok"] for x in r], [False, False, True])
|
||||
self.assertEqual([os.path.isfile(c) for c in clips], [True, True, False])
|
||||
# …ni aunque viniera dado por bueno de una versión anterior: se repasa al arrancar y al reemplazar
|
||||
O.set_trabajo(ids[0], estado="verificado")
|
||||
O.set_trabajo(ids[1], estado="verificado")
|
||||
r = O.reemplazar([ids[1]])[0]
|
||||
self.assertFalse(r["ok"])
|
||||
self.assertTrue(r["motivo"].startswith("no compensa"), r)
|
||||
self.assertEqual(O.repasar_verificados(), 1)
|
||||
self.assertEqual([self.trabajo(i)["estado"] for i in ids[:2]], ["descartado", "descartado"])
|
||||
self.assertTrue(os.path.isfile(clips[0]) and os.path.isfile(clips[1]))
|
||||
# y el clip no se vuelve a ofrecer con la misma calidad; con otra, sí
|
||||
cs = {os.path.basename(c["ruta"]): c["excluido"] for c in O.candidatos(nr, 0, "equilibrado", 0)}
|
||||
self.assertTrue(cs["C0002.MP4"].startswith("ya se probó con esta calidad: pesa el 95 %"), cs)
|
||||
cs = {os.path.basename(c["ruta"]): c["excluido"] for c in O.candidatos(nr, 0, "ahorro", 0)}
|
||||
self.assertEqual(cs["C0002.MP4"], "")
|
||||
# a un formato de edición no se le pide que pese menos
|
||||
self.assertEqual(O._no_compensa({"formato": "prores_lt", "ruta": clips[1], "bytes_in": 100, "bytes_out": 900}), "")
|
||||
|
||||
def test_lo_que_no_compensa_se_aparta(self):
|
||||
"""Lo que salió de una conversión que no compensa no se queda junto al original ocupando su nombre, y un
|
||||
intento que no pasó la verificación se puede rehacer."""
|
||||
raiz = cliente("Apartar")
|
||||
g = "2025/11- Noviembre/00_Grabacion/SONY/"
|
||||
a, b, c = (mk(raiz, g + n, 2025, 11, 11, tam=300000) for n in ("C0001.MP4", "C0002.MP4", "C0003.MP4"))
|
||||
O.escanear(raiz)
|
||||
nr = os.path.normpath(raiz)
|
||||
self.leidos(raiz, {n: {"codec": "h264", "perfil": "High", "kbps": 120000} for n in ("C0001.MP4", "C0002.MP4", "C0003.MP4")})
|
||||
O.ffprobe = lambda ruta: {"codec": "hevc", "perfil": "Main 10", "pix_fmt": "yuv420p10le", "ancho": 3840, "alto": 2160, "fps": 30.0, "dur": 60.0,
|
||||
"kbps": 20000, "alfa": 0, "acodec": "aac", "pistas_audio": 1}
|
||||
ids = O.encolar([a, b, c], "equilibrado", "420", "rec709", nr)
|
||||
self.assertEqual(len(ids), 3)
|
||||
for i in ids: # tres conversiones que salieron casi del mismo peso
|
||||
with open(self.trabajo(i)["salida"], "wb") as fh:
|
||||
fh.write(b"x" * 290000)
|
||||
O.set_trabajo(i, estado="listo", bytes_out=290000)
|
||||
apartado = lambda t: os.path.join(nr, R.PAPELERA, "descartados", os.path.relpath(t["salida"], nr))
|
||||
|
||||
# 1) recién convertida y no compensa: lo que salió se aparta en el acto, con su lote en la bitácora
|
||||
v = O.verificar([ids[0]])[0]
|
||||
self.assertFalse(v["ok"])
|
||||
self.assertIn("descartados", v["motivo"])
|
||||
t0 = self.trabajo(ids[0])
|
||||
self.assertFalse(os.path.exists(t0["salida"]))
|
||||
self.assertTrue(os.path.isfile(apartado(t0)))
|
||||
self.assertTrue(os.path.isfile(a)) # el original, intacto
|
||||
lotes = [e["lote"] for e in O.leer_bitacora() if e.get("op") == "fin" and e.get("raiz") == nr and e["lote"].startswith("descartados-")]
|
||||
self.assertEqual(len(lotes), 1)
|
||||
O.deshacer(lotes[0]) # y «Deshacer» lo devuelve
|
||||
self.assertTrue(os.path.isfile(t0["salida"]))
|
||||
|
||||
# 2) las que venían «verificadas» de antes: al arrancar solo cambian de estado, sin mover nada…
|
||||
O.set_trabajo(ids[1], estado="verificado")
|
||||
O.set_trabajo(ids[2], estado="verificado")
|
||||
self.assertEqual(O.repasar_verificados(), 2)
|
||||
t1, t2 = self.trabajo(ids[1]), self.trabajo(ids[2])
|
||||
self.assertEqual((t1["estado"], t2["estado"]), ("descartado", "descartado"))
|
||||
self.assertTrue(os.path.isfile(t1["salida"]) and os.path.isfile(t2["salida"]))
|
||||
# …volver a convertir el clip con otra calidad aparta lo que salió de aquel y deja libre la ruta…
|
||||
nuevos = O.encolar([b], "ahorro", "420", "rec709", nr)
|
||||
self.assertEqual(len(nuevos), 1)
|
||||
self.assertFalse(os.path.exists(t1["salida"]))
|
||||
self.assertTrue(os.path.isfile(apartado(t1)))
|
||||
# …y «Limpiar terminados» aparta lo de los que quedaran
|
||||
r = O.apartar_descartados([ids[2]])
|
||||
self.assertEqual((r["movidos"], r["errores"], r["quedan"]), (1, [], []))
|
||||
self.assertFalse(os.path.exists(t2["salida"]))
|
||||
self.assertTrue(os.path.isfile(apartado(t2)))
|
||||
|
||||
# 3) un intento que no pasó la verificación se rehace: su propia salida no bloquea la ruta; otra sí
|
||||
tn = self.trabajo(nuevos[0])
|
||||
with open(tn["salida"], "wb") as fh:
|
||||
fh.write(b"y" * 1000)
|
||||
O.set_trabajo(nuevos[0], estado="error", bytes_out=1000, log="verificación: duración 3.00s vs original 60.00s")
|
||||
self.assertFalse(O._salida_ajena(self.trabajo(nuevos[0])))
|
||||
with open(tn["salida"], "wb") as fh:
|
||||
fh.write(b"z" * 999) # alguien dejó otro archivo con ese nombre
|
||||
self.assertTrue(O._salida_ajena(self.trabajo(nuevos[0])))
|
||||
|
||||
def test_el_codificador_que_toca(self):
|
||||
fm = O.FORMATOS["h265_10bit"]
|
||||
t = {"chroma": "420", "perfil": "equilibrado", "kbps": 30000}
|
||||
con_uhq = dict(self.MOTORES, gpu=dict(self.MOTORES["gpu"], hevc=True, h264=True, uhq=True), formatos={k: "gpu" for k in O.FORMATOS})
|
||||
O.motores = lambda: con_uhq
|
||||
args, pix = O._args_video(t, fm, "gpu", 30)
|
||||
self.assertEqual((pix, args[args.index("-tune") + 1], args[args.index("-profile:v") + 1]), ("p010le", "uhq", "main10"))
|
||||
self.assertNotIn("-bf", args) # «uhq» no admite que se le fijen los cuadros B
|
||||
self.assertEqual(O._args_video(dict(t, chroma="422"), fm, "gpu", 30)[1], "p210le")
|
||||
self.assertIn("hq", O._args_video(dict(t, chroma="422"), fm, "gpu", 30)[0]) # solo donde se midió: 4:2:0
|
||||
self.assertIn("hq", O._args_video(t, O.FORMATOS["h264"], "gpu", 30)[0])
|
||||
O.motores = lambda: dict(con_uhq, gpu=dict(con_uhq["gpu"], uhq=False))
|
||||
args, _ = O._args_video(t, fm, "gpu", 30)
|
||||
self.assertEqual(args[args.index("-tune") + 1], "hq")
|
||||
self.assertIn("-bf", args)
|
||||
# la tarjeta tiene un tamaño máximo de imagen por códec: un 4.6K en H.264 falla («No capable devices found»)
|
||||
self.assertEqual(O._motor_de("h264", "420", 4608, 2592)[0], "cpu")
|
||||
self.assertEqual(O._motor_de("h264", "420", 3840, 2160)[0], "gpu")
|
||||
self.assertEqual(O._motor_de("h265_10bit", "420", 4608, 2592)[0], "gpu")
|
||||
self.assertEqual(O._motor_de("h265_10bit", "420", 12288, 6480)[0], "cpu")
|
||||
# el contenedor guarda las etiquetas de la cámara, sin arrastrar las marcas de formato del original
|
||||
cont, audio = O._args_salida(fm, None)
|
||||
self.assertIn("+faststart+use_metadata_tags", cont)
|
||||
self.assertIn("major_brand=", cont)
|
||||
self.assertEqual(audio, ["-c:a", "aac", "-b:a", "320k"])
|
||||
|
||||
def test_el_lector_sin_decodificador(self):
|
||||
"""Lo que braw.py hace sin tocar el decodificador, y cómo responde si lo que encuentra no sirve."""
|
||||
import braw
|
||||
self.assertEqual([braw.fraccion(x) for x in (23.976, 24.0, 25.0, 29.97, 30.0, 59.94, 60.0)],
|
||||
["24000/1001", "24", "25", "30000/1001", "30", "60000/1001", "60"])
|
||||
falso = os.path.join(TMP, "decodificador-falso")
|
||||
os.makedirs(falso)
|
||||
self.assertNotEqual(braw.buscar(falso), falso)
|
||||
with open(os.path.join(falso, braw.NOMBRE), "wb") as fh:
|
||||
fh.write(b"esto no es una DLL")
|
||||
self.assertEqual(braw.buscar(falso), falso)
|
||||
r = braw.sonda(falso)
|
||||
self.assertFalse(r["ok"])
|
||||
self.assertTrue(r["motivo"])
|
||||
r = braw.info(os.path.join(TMP, "no-existe.braw"), falso)
|
||||
self.assertFalse(r["ok"])
|
||||
self.assertTrue(r["error"].startswith("el clip ya no está en esa ruta"), r)
|
||||
|
||||
@unittest.skipUnless(os.environ.get("ORGANILAB_BRAW_PRUEBA"), "sin un .braw de prueba (variable ORGANILAB_BRAW_PRUEBA)")
|
||||
def test_un_braw_de_verdad(self):
|
||||
"""Con el decodificador instalado y un clip real: se mide, se convierte, se verifica."""
|
||||
O.decodificador_braw, O.motores, O.medir_braw, O.ffprobe = self.de_verdad
|
||||
self.assertTrue(O.decodificador_braw()["ok"], O.decodificador_braw())
|
||||
raiz = cliente("BrawReal")
|
||||
destino = os.path.join(raiz, "2025", "03- Marzo", "00_Grabacion", "URSA_BRAW", "A002_03110347_C038.braw")
|
||||
os.makedirs(os.path.dirname(destino))
|
||||
shutil.copy(os.environ["ORGANILAB_BRAW_PRUEBA"], destino)
|
||||
O.escanear(raiz)
|
||||
nr = os.path.normpath(raiz)
|
||||
O.analizar(nr, 0)
|
||||
real = O.medir_braw(destino)
|
||||
self.assertTrue(real["ok"], real)
|
||||
c = O.candidatos(nr, 0, "equilibrado", 0)[0]
|
||||
self.assertEqual((c["motor"], c["excluido"], c["res"]), ("braw", "", f"{real['ancho']}x{real['alto']}"))
|
||||
ids = O.encolar([destino], "equilibrado", "420", "rec709", nr)
|
||||
O.correr_braw(self.trabajo(ids[0]))
|
||||
t = self.trabajo(ids[0])
|
||||
self.assertEqual(t["estado"], "listo", t["log"])
|
||||
v = O.verificar(ids)[0]
|
||||
self.assertTrue(v["ok"], v)
|
||||
i = O.ffprobe(t["salida"])
|
||||
self.assertEqual((i["codec"], i["pix_fmt"], i["ancho"], i["alto"]), ("hevc", "yuv420p10le", real["ancho"], real["alto"]))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
unittest.main(verbosity=2, exit=False)
|
||||
|
||||
Referencia en una nueva incidencia
Block a user