{ "nbformat": 4, "nbformat_minor": 5, "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "name": "python" }, "accelerator": "GPU" }, "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# 02 · Movimiento corporal — MimicMotion (experimental)\n", "\n", "**Objetivo:** `chica.png + guia.mp4 → humana_movimiento.mp4`, sin mostrar el guía final.\n", "\n", "Usa un notebook NUEVO con **GPU T4** e **Internet activado**. Añade `chica.png` y un vídeo corto llamado `guia.mp4` con una persona visible, encuadre estable, manos visibles y un gesto suave. Un plano y una postura similares a la referencia facilitan la prueba.\n", "\n", "Este entorno es independiente de MuseTalk: usa Diffusers 0.27 y Hub 0.24.7. Se comienza con guía de 3 segundos, 384 px de lado corto, ventanas de 16 fotogramas y VAE decode_chunk_size=1. No se garantiza una calidad o latencia concreta en T4.\n", "\n", "Puede ser necesario aceptar el acceso al modelo base [SVD](https://huggingface.co/stabilityai/stable-video-diffusion-img2vid-xt-1-1) en Hugging Face y guardar **HF_TOKEN** en Kaggle Secrets. Las licencias del código y de cada peso deben revisarse antes de explotación comercial." ], "id": "cell-00" }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 1. Opciones" ], "id": "cell-01" }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "IMAGE_PATH = ''\n", "GUIDE_PATH = ''\n", "GUIDE_SECONDS = 3\n", "RESOLUTION = 384\n", "STEPS = 15" ], "id": "cell-02" }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 2. Preparar scripts" ], "id": "cell-03" }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from pathlib import Path\n", "import os, json, subprocess, sys\n", "ROOT = Path('/kaggle/working/humana') if Path('/kaggle').exists() else Path.cwd() / 'humana-runtime'\n", "ROOT.mkdir(parents=True, exist_ok=True)\n", "GPU = ROOT / 'gpu'\n", "GPU.mkdir(exist_ok=True)\n", "PAYLOAD = {'doctor.py': '\"\"\"Se ejecuta con el Python aislado, nunca con el del notebook.\"\"\"\\nimport argparse\\nimport importlib.metadata as metadata\\nimport json\\nimport subprocess\\nimport sys\\n\\n\\ndef check(engine=\\'musetalk\\'):\\n import torch\\n import numpy\\n import cv2\\n assert sys.version_info[:2] == (3, 10), \\'No se está usando el Python aislado 3.10.\\'\\n assert torch.cuda.is_available(), \\'Activa la GPU T4 en Settings > Accelerator.\\'\\n assert torch.__version__ == \\'2.0.1+cu118\\', \\'Versión de Torch inesperada.\\'\\n assert numpy.__version__ == \\'1.23.5\\', \\'NumPy incompatible.\\'\\n report = {\\'python\\': sys.version, \\'torch\\': torch.__version__, \\'cuda_runtime\\': torch.version.cuda,\\n \\'gpu\\': torch.cuda.get_device_name(0), \\'vram_gb\\': round(torch.cuda.get_device_properties(0).total_memory / 2**30, 2),\\n \\'numpy\\': numpy.__version__, \\'opencv\\': cv2.__version__}\\n if engine == \\'musetalk\\':\\n import mmcv\\n import mmpose\\n import mmdet\\n from mmcv.ops import nms\\n # Verifica las operaciones compiladas y su CUDA, además de los imports.\\n boxes = torch.tensor([[0., 0., 10., 10.], [1., 1., 9., 9.]], device=\\'cuda\\')\\n scores = torch.tensor([0.9, 0.8], device=\\'cuda\\')\\n _, kept = nms(boxes, scores, 0.5)\\n assert kept.numel() == 1\\n report.update(mmcv=mmcv.__version__, mmpose=mmpose.__version__, mmdet=mmdet.__version__, mmcv_cuda_ops=\\'OK\\')\\n else:\\n import diffusers\\n import onnxruntime\\n report.update(diffusers=diffusers.__version__, onnx_providers=onnxruntime.get_available_providers())\\n report[\\'dependencies\\'] = {name: metadata.version(name) for name in [\\'diffusers\\', \\'transformers\\', \\'huggingface-hub\\']}\\n subprocess.run([sys.executable, \\'-m\\', \\'pip\\', \\'check\\'], check=True)\\n print(json.dumps(report, indent=2, ensure_ascii=False))\\n print(\\'ENTORNO AISLADO OK. El kernel de Kaggle puede seguir en Python 3.12.\\')\\n\\n\\nif __name__ == \\'__main__\\':\\n parser = argparse.ArgumentParser(); parser.add_argument(\\'--engine\\', default=\\'musetalk\\')\\n check(parser.parse_args().engine)\\n', 'download_models.py': \"import argparse\\nimport json\\nimport os\\nfrom pathlib import Path\\nimport urllib.request\\nfrom huggingface_hub import hf_hub_download, snapshot_download\\nfrom model_files import materialize, check_models\\n\\n\\ndef download(vendor, engine):\\n manifest = json.loads(Path(__file__).with_name('models.json').read_text())\\n cache = vendor.parent / 'hf-cache'\\n token = os.environ.get('HF_TOKEN') or None\\n marker = vendor / 'models/.models-manifest.json'\\n previous = json.loads(marker.read_text()) if marker.exists() else None\\n for item in manifest[engine]:\\n destination = vendor / 'models' / item['destination']\\n destination.parent.mkdir(parents=True, exist_ok=True)\\n if previous == manifest[engine] and destination.is_file() and not destination.is_symlink() and destination.stat().st_size > 0:\\n print('Modelo reutilizado:', item['destination'], flush=True)\\n continue\\n print('Modelo:', item['destination'], flush=True)\\n cached = hf_hub_download(item['repo'], item['filename'], revision=item['revision'], cache_dir=str(cache), token=token)\\n materialize(cached, destination)\\n if engine == 'musetalk':\\n s3fd = vendor / 'musetalk/utils/face_detection/detection/sfd/s3fd.pth'\\n if not s3fd.is_file() or s3fd.stat().st_size < 1000000:\\n temporary = s3fd.with_suffix('.part')\\n urllib.request.urlretrieve('https://www.adrianbulat.com/downloads/python-fan/s3fd-619a316812.pth', temporary)\\n if temporary.stat().st_size < 1000000: raise RuntimeError('Descarga incompleta de S3FD.')\\n temporary.replace(s3fd)\\n else:\\n svd = manifest['svd']\\n print('Modelo base SVD: la primera descarga es grande.', flush=True)\\n location = snapshot_download(svd['repo'], revision=svd['revision'], cache_dir=str(cache), token=token,\\n allow_patterns=['model_index.json','scheduler/*','feature_extractor/*','unet/config.json',\\n 'vae/config.json','vae/*fp16.safetensors','image_encoder/config.json','image_encoder/*fp16.safetensors'])\\n (vendor / 'svd-location.txt').write_text(location)\\n check_models(vendor, engine)\\n marker.write_text(json.dumps(manifest[engine], indent=2))\\n print('Modelos descargados con revisiones fijadas.', flush=True)\\n\\n\\nif __name__ == '__main__':\\n parser = argparse.ArgumentParser(); parser.add_argument('--vendor', type=Path, required=True)\\n parser.add_argument('--engine', choices=['musetalk','mimicmotion'], default='musetalk')\\n args = parser.parse_args(); download(args.vendor.resolve(), args.engine)\\n\", 'mimicmotion-requirements.txt': '# Entorno separado: Diffusers 0.27 usa cached_download, eliminado en Hub 0.26.\\nnumpy==1.23.5\\nscipy==1.10.1\\nopencv-python==4.9.0.80\\ndiffusers==0.27.0\\ntransformers==4.32.1\\nhuggingface-hub==0.24.7\\naccelerate==0.28.0\\ndecord==0.6.0\\neinops==0.8.1\\nomegaconf==2.3.0\\nonnxruntime-gpu==1.16.3\\nimageio==2.34.2\\nimageio-ffmpeg==0.5.1\\nPillow==10.4.0\\nrequests==2.32.4\\ntqdm==4.67.1\\nsympy==1.12\\nprotobuf==4.25.3\\n', 'model_files.py': '\"\"\"Archivos reales de modelos y reparación local de enlaces de la caché HF.\"\"\"\\nimport argparse\\nimport json\\nimport os\\nfrom pathlib import Path\\nimport shutil\\nimport tempfile\\n\\n\\ndef manifest(engine=\\'musetalk\\'):\\n return json.loads(Path(__file__).with_name(\\'models.json\\').read_text())[engine]\\n\\n\\ndef materialize(source, destination):\\n # hf_hub_download devuelve normalmente un enlace relativo en snapshots/.\\n # En Linux os.link puede copiar ese enlace, cuya ruta se rompe en models/.\\n source = Path(source).resolve(strict=True)\\n destination = Path(destination)\\n if not source.is_file() or source.stat().st_size == 0:\\n raise FileNotFoundError(f\\'El modelo de origen está vacío o no existe: {source}\\')\\n destination.parent.mkdir(parents=True, exist_ok=True)\\n if destination.is_file() and not destination.is_symlink() and destination.stat().st_size == source.stat().st_size:\\n return\\n with tempfile.TemporaryDirectory(prefix=\\'.model_\\', dir=destination.parent) as folder:\\n temporary = Path(folder) / destination.name\\n try:\\n os.link(source, temporary)\\n except OSError:\\n shutil.copyfile(source, temporary)\\n temporary.replace(destination)\\n\\n\\ndef check_models(vendor, engine=\\'musetalk\\'):\\n vendor = Path(vendor)\\n required = [vendor / \\'models\\' / item[\\'destination\\'] for item in manifest(engine)]\\n if engine == \\'musetalk\\':\\n required.append(vendor / \\'musetalk/utils/face_detection/detection/sfd/s3fd.pth\\')\\n missing = [str(p.relative_to(vendor)) for p in required if not p.is_file() or p.stat().st_size == 0]\\n if missing:\\n raise FileNotFoundError(\\'Faltan modelos o hay enlaces rotos: \\' + \\', \\'.join(missing) +\\n \\'. Ejecuta la reparación de modelos; si falta la caché, repite la celda 5.\\')\\n\\n\\ndef repair_models(vendor, engine=\\'musetalk\\'):\\n vendor = Path(vendor).resolve()\\n cache = vendor.parent / \\'hf-cache\\'\\n missing = []\\n for item in manifest(engine):\\n destination = vendor / \\'models\\' / item[\\'destination\\']\\n if destination.is_file() and not destination.is_symlink() and destination.stat().st_size > 0:\\n continue\\n cached = cache / (\\'models--\\' + item[\\'repo\\'].replace(\\'/\\', \\'--\\')) / \\'snapshots\\' / item[\\'revision\\'] / item[\\'filename\\']\\n if not cached.is_file():\\n missing.append(item[\\'destination\\'])\\n continue\\n materialize(cached, destination)\\n print(\\'Modelo reparado:\\', item[\\'destination\\'], flush=True)\\n if missing:\\n raise FileNotFoundError(\\'No están en la caché local: \\' + \\', \\'.join(missing) + \\'. Repite la celda 5 con el descargador actualizado.\\')\\n check_models(vendor, engine)\\n print(\\'Modelos comprobados. Reparación completada sin reinstalar paquetes ni descargar pesos.\\', flush=True)\\n\\n\\nif __name__ == \\'__main__\\':\\n parser = argparse.ArgumentParser()\\n parser.add_argument(\\'--vendor\\', type=Path, required=True)\\n parser.add_argument(\\'--engine\\', choices=[\\'musetalk\\', \\'mimicmotion\\'], default=\\'musetalk\\')\\n args = parser.parse_args()\\n repair_models(args.vendor, args.engine)\\n', 'models.json': '{\\n \"musetalk\": [\\n {\\n \"repo\": \"TMElyralab/MuseTalk\",\\n \"filename\": \"musetalkV15/musetalk.json\",\\n \"destination\": \"musetalkV15/musetalk.json\",\\n \"revision\": \"2bcb936e2fddb4d86db4c62fd45b387d0c061571\"\\n },\\n {\\n \"repo\": \"TMElyralab/MuseTalk\",\\n \"filename\": \"musetalkV15/unet.pth\",\\n \"destination\": \"musetalkV15/unet.pth\",\\n \"revision\": \"2bcb936e2fddb4d86db4c62fd45b387d0c061571\"\\n },\\n {\\n \"repo\": \"stabilityai/sd-vae-ft-mse\",\\n \"filename\": \"config.json\",\\n \"destination\": \"sd-vae/config.json\",\\n \"revision\": \"31f26fdeee1355a5c34592e401dd41e45d25a493\"\\n },\\n {\\n \"repo\": \"stabilityai/sd-vae-ft-mse\",\\n \"filename\": \"diffusion_pytorch_model.bin\",\\n \"destination\": \"sd-vae/diffusion_pytorch_model.bin\",\\n \"revision\": \"31f26fdeee1355a5c34592e401dd41e45d25a493\"\\n },\\n {\\n \"repo\": \"openai/whisper-tiny\",\\n \"filename\": \"config.json\",\\n \"destination\": \"whisper/config.json\",\\n \"revision\": \"169d4a4341b33bc18d8881c4b69c2e104e1cc0af\"\\n },\\n {\\n \"repo\": \"openai/whisper-tiny\",\\n \"filename\": \"pytorch_model.bin\",\\n \"destination\": \"whisper/pytorch_model.bin\",\\n \"revision\": \"169d4a4341b33bc18d8881c4b69c2e104e1cc0af\"\\n },\\n {\\n \"repo\": \"openai/whisper-tiny\",\\n \"filename\": \"preprocessor_config.json\",\\n \"destination\": \"whisper/preprocessor_config.json\",\\n \"revision\": \"169d4a4341b33bc18d8881c4b69c2e104e1cc0af\"\\n },\\n {\\n \"repo\": \"yzd-v/DWPose\",\\n \"filename\": \"dw-ll_ucoco_384.pth\",\\n \"destination\": \"dwpose/dw-ll_ucoco_384.pth\",\\n \"revision\": \"1a7144101628d69ee7a3768d1ee3a094070dc388\"\\n },\\n {\\n \"repo\": \"ManyOtherFunctions/face-parse-bisent\",\\n \"filename\": \"79999_iter.pth\",\\n \"destination\": \"face-parse-bisent/79999_iter.pth\",\\n \"revision\": \"0073b233a5a3c4b1377d4dbf49245017938a72b5\"\\n },\\n {\\n \"repo\": \"ManyOtherFunctions/face-parse-bisent\",\\n \"filename\": \"resnet18-5c106cde.pth\",\\n \"destination\": \"face-parse-bisent/resnet18-5c106cde.pth\",\\n \"revision\": \"0073b233a5a3c4b1377d4dbf49245017938a72b5\"\\n }\\n ],\\n \"mimicmotion\": [\\n {\\n \"repo\": \"yzd-v/DWPose\",\\n \"filename\": \"yolox_l.onnx\",\\n \"destination\": \"DWPose/yolox_l.onnx\",\\n \"revision\": \"1a7144101628d69ee7a3768d1ee3a094070dc388\"\\n },\\n {\\n \"repo\": \"yzd-v/DWPose\",\\n \"filename\": \"dw-ll_ucoco_384.onnx\",\\n \"destination\": \"DWPose/dw-ll_ucoco_384.onnx\",\\n \"revision\": \"1a7144101628d69ee7a3768d1ee3a094070dc388\"\\n },\\n {\\n \"repo\": \"tencent/MimicMotion\",\\n \"filename\": \"MimicMotion_1-1.pth\",\\n \"destination\": \"MimicMotion_1-1.pth\",\\n \"revision\": \"db35a2c5d0a12db9155060f14688e72dedbfb370\"\\n }\\n ],\\n \"svd\": {\\n \"repo\": \"stabilityai/stable-video-diffusion-img2vid-xt-1-1\",\\n \"revision\": \"043843887ccd51926e3efed36270444a838e7861\"\\n }\\n}\\n', 'motion.py': '\"\"\"Prueba experimental de movimiento: vídeo guía -> poses -> identidad de chica.png.\"\"\"\\nimport argparse\\nimport os\\nfrom pathlib import Path\\nimport subprocess\\nimport sys\\n\\n\\ndef animate(vendor, image, guide, output, seconds=3, resolution=384, steps=15):\\n vendor, image, guide, output = [Path(p).resolve() for p in (vendor, image, guide, output)]\\n sys.path.insert(0, str(vendor))\\n os.chdir(vendor)\\n import torch\\n import numpy as np\\n from omegaconf import OmegaConf\\n from torchvision.transforms.functional import to_pil_image\\n from inference import preprocess\\n from mimicmotion.utils.loader import create_pipeline\\n from mimicmotion.utils.utils import save_to_mp4\\n if not torch.cuda.is_available(): raise RuntimeError(\\'Activa una GPU T4.\\')\\n output.parent.mkdir(parents=True, exist_ok=True)\\n normalized = output.parent / \\'guide-normalized.mp4\\'\\n # El modelo trabaja sobre todos los fotogramas; num_frames es el tamaño de\\n # ventana, no un límite total. Recortamos el guía ANTES de extraer las poses.\\n subprocess.run([\\'ffmpeg\\',\\'-y\\',\\'-v\\',\\'error\\',\\'-i\\',str(guide),\\'-t\\',str(seconds),\\'-an\\',\\'-r\\',\\'12\\',\\'-c:v\\',\\'libx264\\',str(normalized)], check=True)\\n torch.set_default_dtype(torch.float16)\\n cfg = OmegaConf.create({\\'base_model_path\\': (vendor/\\'svd-location.txt\\').read_text().strip(),\\n \\'ckpt_path\\': str(vendor/\\'models/MimicMotion_1-1.pth\\')})\\n pipeline = create_pipeline(cfg, torch.device(\\'cuda:0\\'))\\n pipeline.enable_attention_slicing()\\n pose, pixels = preprocess(str(normalized), str(image), resolution=resolution, sample_stride=1)\\n reference = [to_pil_image(img.to(torch.uint8)) for img in (pixels + 1) * 127.5]\\n with torch.no_grad():\\n frames = pipeline(reference, image_pose=pose, num_frames=pose.size(0), tile_size=16, tile_overlap=4,\\n height=pose.shape[-2], width=pose.shape[-1], fps=7, noise_aug_strength=0,\\n num_inference_steps=steps, generator=torch.Generator(device=\\'cuda\\').manual_seed(42),\\n min_guidance_scale=2.0, max_guidance_scale=2.0, decode_chunk_size=1,\\n output_type=\\'pt\\', device=torch.device(\\'cuda:0\\')).frames.cpu()\\n save_to_mp4((frames[0, 1:] * 255).to(torch.uint8), str(output), fps=12)\\n subprocess.run([\\'ffprobe\\', \\'-v\\', \\'error\\', \\'-show_entries\\', \\'format=duration\\', str(output)], check=True)\\n print(\\'Movimiento experimental generado:\\', output)\\n print(\\'Revisa identidad, manos, ojos y continuidad antes de usarlo en el directo.\\')\\n\\n\\nif __name__ == \\'__main__\\':\\n parser = argparse.ArgumentParser()\\n for name in [\\'vendor\\',\\'image\\',\\'guide\\',\\'output\\']: parser.add_argument(\\'--\\'+name,type=Path,required=True)\\n parser.add_argument(\\'--seconds\\',type=float,default=3)\\n parser.add_argument(\\'--resolution\\',type=int,default=384)\\n parser.add_argument(\\'--steps\\',type=int,default=15)\\n animate(**vars(parser.parse_args()))\\n', 'musetalk-requirements.txt': '# Solo inferencia. TensorFlow, Gradio y los paquetes de entrenamiento no son necesarios.\\nnumpy==1.23.5\\nscipy==1.10.1\\nopencv-python==4.9.0.80\\ndiffusers==0.30.2\\naccelerate==0.28.0\\ntransformers==4.39.2\\nhuggingface-hub==0.30.2\\nlibrosa==0.11.0\\nnumba==0.60.0\\nllvmlite==0.43.0\\nsoundfile==0.12.1\\neinops==0.8.1\\nomegaconf==2.3.0\\nffmpeg-python==0.2.0\\nimageio==2.34.2\\nimageio-ffmpeg==0.5.1\\nPillow==10.4.0\\nmatplotlib==3.7.5\\nmmengine==0.10.7\\nmmdet==3.1.0\\nmmpose==1.1.0\\nxtcocotools==1.14.3\\njson-tricks==3.17.3\\nmunkres==1.1.4\\nrequests==2.32.4\\ntqdm==4.67.1\\nsympy==1.12\\nprotobuf==4.25.3\\npycocotools==2.0.7\\nscikit-learn==1.3.2\\nyapf==0.40.1\\n', 'render.py': '\"\"\"Adaptador de MuseTalk 1.5: prepara entradas y verifica que exista vídeo con audio.\"\"\"\\nimport argparse\\nimport json\\nimport os\\nfrom pathlib import Path\\nimport subprocess\\nimport sys\\nimport tempfile\\nfrom model_files import check_models\\n\\n\\ndef probe(path):\\n result = subprocess.check_output([\\'ffprobe\\', \\'-v\\', \\'error\\', \\'-show_streams\\', \\'-show_format\\', \\'-of\\', \\'json\\', str(path)], text=True)\\n return json.loads(result)\\n\\n\\ndef render(vendor, image, audio, output, source=None, max_seconds=5, batch_size=2):\\n vendor, image, audio, output = [Path(p).resolve() for p in (vendor, image, audio, output)]\\n if not image.is_file() or not audio.is_file():\\n raise FileNotFoundError(\\'Falta la imagen o el audio de entrada.\\')\\n check_models(vendor)\\n if any(c.isspace() for c in str(vendor)):\\n raise ValueError(\\'La carpeta MuseTalk debe tener una ruta sin espacios.\\')\\n output.parent.mkdir(parents=True, exist_ok=True)\\n env = os.environ.copy()\\n env.pop(\\'PYTHONHOME\\', None)\\n env[\\'PYTHONNOUSERSITE\\'] = \\'1\\'\\n env[\\'PYTHONPATH\\'] = os.pathsep.join([str(vendor), str(vendor / \\'musetalk/utils\\')])\\n env[\\'CUDA_VISIBLE_DEVICES\\'] = \\'0\\'\\n env[\\'PYTORCH_CUDA_ALLOC_CONF\\'] = \\'max_split_size_mb:128\\'\\n with tempfile.TemporaryDirectory(prefix=\\'humana_\\', dir=vendor.parent) as tmp:\\n work = Path(tmp)\\n wav = work / \\'audio.wav\\'\\n audio_args = [\\'ffmpeg\\', \\'-y\\', \\'-v\\', \\'error\\', \\'-i\\', str(audio), \\'-vn\\', \\'-ac\\', \\'1\\', \\'-ar\\', \\'16000\\']\\n if max_seconds > 0: audio_args += [\\'-t\\', str(max_seconds)]\\n subprocess.run(audio_args + [str(wav)], check=True)\\n duration = float(probe(wav)[\\'format\\'][\\'duration\\'])\\n if duration < 0.3: raise ValueError(\\'El audio debe durar al menos 0,3 segundos.\\')\\n base = work / \\'source.mp4\\'\\n if source:\\n source = Path(source).resolve()\\n source_duration = float(probe(source)[\\'format\\'][\\'duration\\'])\\n if source_duration + 0.04 < duration:\\n raise ValueError(\\'El movimiento corporal es más corto que la voz. Genera un movimiento más largo para evitar repetirlo hacia atrás.\\')\\n subprocess.run([\\'ffmpeg\\', \\'-y\\', \\'-v\\', \\'error\\', \\'-i\\', str(source), \\'-t\\', str(duration), \\'-an\\', \\'-r\\', \\'25\\',\\n \\'-vf\\', \\'scale=trunc(iw/2)*2:trunc(ih/2)*2\\', \\'-c:v\\', \\'libx264\\', \\'-pix_fmt\\', \\'yuv420p\\', str(base)], check=True)\\n else:\\n # Una imagen se convierte en vídeo de un fotograma. Evita el fallo de limpieza\\n # save_dir_full del script oficial al pasar imágenes directamente.\\n subprocess.run([\\'ffmpeg\\', \\'-y\\', \\'-v\\', \\'error\\', \\'-loop\\', \\'1\\', \\'-i\\', str(image), \\'-frames:v\\', \\'1\\', \\'-r\\', \\'25\\',\\n \\'-vf\\', \\'scale=trunc(iw/2)*2:trunc(ih/2)*2\\', \\'-c:v\\', \\'libx264\\', \\'-pix_fmt\\', \\'yuv420p\\', str(base)], check=True)\\n config = work / \\'inference.json\\'\\n config.write_text(json.dumps({\\'humana\\': {\\'video_path\\': str(base), \\'audio_path\\': str(wav), \\'result_name\\': \\'result.mp4\\'}}))\\n result_dir = work / \\'result\\'\\n command = [sys.executable, \\'-m\\', \\'scripts.inference\\', \\'--inference_config\\', str(config),\\n \\'--result_dir\\', str(result_dir), \\'--unet_model_path\\', \\'models/musetalkV15/unet.pth\\',\\n \\'--unet_config\\', \\'models/musetalkV15/musetalk.json\\', \\'--whisper_dir\\', \\'models/whisper\\',\\n \\'--version\\', \\'v15\\', \\'--fps\\', \\'25\\', \\'--batch_size\\', str(batch_size), \\'--use_float16\\']\\n subprocess.run(command, cwd=vendor, env=env, check=True)\\n result = result_dir / \\'v15/result.mp4\\'\\n if not result.is_file() or result.stat().st_size < 1000:\\n raise RuntimeError(\\'MuseTalk no creó el vídeo. El script oficial puede devolver código 0 incluso al fallar; revisa el error anterior.\\')\\n # Reempaqueta con duración exacta, audio AAC y moov al inicio para navegador/Safari.\\n subprocess.run([\\'ffmpeg\\', \\'-y\\', \\'-v\\', \\'error\\', \\'-i\\', str(result), \\'-i\\', str(wav), \\'-map\\', \\'0:v:0\\', \\'-map\\', \\'1:a:0\\',\\n \\'-c:v\\', \\'copy\\', \\'-c:a\\', \\'aac\\', \\'-b:a\\', \\'128k\\', \\'-shortest\\', \\'-movflags\\', \\'+faststart\\', str(output)], check=True)\\n info = probe(output)\\n kinds = {s[\\'codec_type\\'] for s in info[\\'streams\\']}\\n if not {\\'video\\', \\'audio\\'} <= kinds: raise RuntimeError(\\'El resultado no contiene vídeo y audio.\\')\\n if abs(float(info[\\'format\\'][\\'duration\\']) - duration) > 0.35:\\n raise RuntimeError(\\'La duración generada no coincide con el audio.\\')\\n video_stream = next(s for s in info[\\'streams\\'] if s[\\'codec_type\\'] == \\'video\\')\\n if int(video_stream.get(\\'nb_frames\\', 0)) < max(1, int(duration * 25) - 4):\\n raise RuntimeError(\\'Faltan fotogramas en el vídeo generado.\\')\\n print(json.dumps({\\'output\\': str(output), \\'seconds\\': duration, \\'animationMode\\': \\'body-lips\\' if source else \\'lip-sync-only\\'}, ensure_ascii=False))\\n return output\\n\\n\\nif __name__ == \\'__main__\\':\\n parser = argparse.ArgumentParser()\\n for field in [\\'vendor\\', \\'image\\', \\'audio\\', \\'output\\']: parser.add_argument(\\'--\\' + field, type=Path, required=True)\\n parser.add_argument(\\'--source\\', type=Path)\\n parser.add_argument(\\'--max-seconds\\', type=float, default=5)\\n parser.add_argument(\\'--batch-size\\', type=int, default=2)\\n args = parser.parse_args(); render(**vars(args))\\n', 'setup_runtime.py': '\"\"\"Instala un Python independiente. No instala paquetes en el kernel de Kaggle.\"\"\"\\nimport argparse\\nimport hashlib\\nimport io\\nimport os\\nfrom pathlib import Path\\nimport shutil\\nimport subprocess\\nimport tarfile\\nimport urllib.request\\n\\nUV_VERSION = \\'0.7.13\\'\\nUV_SHA256 = \\'909278eb197c5ed0e9b5f16317d1255270d1f9ea4196e7179ce934d48c4c2545\\'\\nREVISIONS = {\\n \\'musetalk\\': (\\'TMElyralab/MuseTalk\\', \\'0a89dec45a0192b824e3cf4daf96c239440c5ed8\\', \\'MuseTalk\\'),\\n \\'mimicmotion\\': (\\'Tencent/MimicMotion\\', \\'6907bdcc259a6a048d41a365e840d22274f9256c\\', \\'MimicMotion\\'),\\n}\\nMMCV_WHEEL = \\'https://download.openmmlab.com/mmcv/dist/cu118/torch2.0.0/mmcv-2.0.1-cp310-cp310-manylinux1_x86_64.whl\\'\\n\\n\\ndef clean_env(root):\\n env = os.environ.copy()\\n for key in [\\'PYTHONPATH\\', \\'PYTHONHOME\\', \\'VIRTUAL_ENV\\', \\'CONDA_PREFIX\\']:\\n env.pop(key, None)\\n env.update(PYTHONNOUSERSITE=\\'1\\', UV_PYTHON_INSTALL_DIR=str(root / \\'python\\'),\\n UV_CACHE_DIR=str(root / \\'uv-cache\\'), CUDA_VISIBLE_DEVICES=\\'0\\')\\n return env\\n\\n\\ndef run(args, env, **kwargs):\\n subprocess.run([str(a) for a in args], check=True, env=env, **kwargs)\\n\\n\\ndef setup(root, engine):\\n root = root.resolve()\\n if any(c.isspace() for c in str(root)):\\n raise RuntimeError(\\'Usa una carpeta de trabajo sin espacios, por ejemplo /kaggle/working/humana.\\')\\n root.mkdir(parents=True, exist_ok=True)\\n env = clean_env(root)\\n uv = root / \\'bin/uv\\'\\n if not uv.exists():\\n print(\\'Descargando uv verificado...\\', flush=True)\\n url = f\\'https://github.com/astral-sh/uv/releases/download/{UV_VERSION}/uv-x86_64-unknown-linux-gnu.tar.gz\\'\\n raw = urllib.request.urlopen(url, timeout=90).read()\\n if hashlib.sha256(raw).hexdigest() != UV_SHA256:\\n raise RuntimeError(\\'La descarga de uv no coincide con su SHA256.\\')\\n with tarfile.open(fileobj=io.BytesIO(raw), mode=\\'r:gz\\') as tar:\\n member = next(m for m in tar.getmembers() if m.isfile() and Path(m.name).name == \\'uv\\')\\n uv.parent.mkdir(exist_ok=True)\\n uv.write_bytes(tar.extractfile(member).read())\\n uv.chmod(0o700)\\n python = root / f\\'env-{engine}/bin/python\\'\\n if not python.exists():\\n run([uv, \\'python\\', \\'install\\', \\'3.10.16\\'], env)\\n run([uv, \\'venv\\', \\'--seed\\', \\'--python\\', \\'3.10.16\\', python.parent.parent], env)\\n marker = python.parent.parent / \\'.installed-v1\\'\\n if not marker.exists():\\n run([python, \\'-m\\', \\'pip\\', \\'install\\', \\'pip==24.0\\', \\'setuptools==69.5.1\\', \\'wheel==0.43.0\\'], env)\\n run([python, \\'-m\\', \\'pip\\', \\'install\\', \\'torch==2.0.1+cu118\\', \\'torchvision==0.15.2+cu118\\',\\n \\'torchaudio==2.0.2+cu118\\', \\'--index-url\\', \\'https://download.pytorch.org/whl/cu118\\'], env)\\n requirements = Path(__file__).with_name(f\\'{engine}-requirements.txt\\')\\n if engine == \\'musetalk\\':\\n # Chumpy importa pip durante setup. Se construye con pip/setuptools ya presentes.\\n run([python, \\'-m\\', \\'pip\\', \\'install\\', \\'numpy==1.23.5\\', \\'scipy==1.10.1\\'], env)\\n run([python, \\'-m\\', \\'pip\\', \\'install\\', \\'--no-build-isolation\\', \\'chumpy==0.70\\'], env)\\n # Descarga obligatoria de wheel: evita compilar MMCV contra el CUDA de Kaggle.\\n run([python, \\'-m\\', \\'pip\\', \\'install\\', \\'--only-binary=:all:\\', \\'--no-deps\\', MMCV_WHEEL], env)\\n run([python, \\'-m\\', \\'pip\\', \\'install\\', \\'-r\\', requirements], env)\\n run([python, \\'-m\\', \\'pip\\', \\'check\\'], env)\\n marker.write_text(\\'Python 3.10.16 / Torch 2.0.1+cu118\\\\n\\')\\n repo, revision, folder = REVISIONS[engine]\\n vendor = root / folder\\n if not vendor.exists():\\n run([\\'git\\', \\'clone\\', f\\'https://github.com/{repo}.git\\', vendor], env)\\n actual = subprocess.check_output([\\'git\\', \\'-C\\', str(vendor), \\'rev-parse\\', \\'HEAD\\'], text=True).strip()\\n if actual != revision:\\n run([\\'git\\', \\'-C\\', vendor, \\'checkout\\', \\'--detach\\', revision], env)\\n if not shutil.which(\\'ffmpeg\\') or not shutil.which(\\'ffprobe\\'):\\n raise RuntimeError(\\'Falta FFmpeg/ffprobe. En Kaggle ejecuta !apt-get update -qq y !apt-get install -y -qq ffmpeg; después repite esta celda.\\')\\n print(f\\'Entorno listo: {python}\\\\nCódigo fijado: {repo}@{revision}\\', flush=True)\\n return python, vendor\\n\\n\\nif __name__ == \\'__main__\\':\\n parser = argparse.ArgumentParser()\\n parser.add_argument(\\'--root\\', type=Path, required=True)\\n parser.add_argument(\\'--engine\\', choices=list(REVISIONS), default=\\'musetalk\\')\\n args = parser.parse_args()\\n setup(args.root, args.engine)\\n', 'worker.py': '\"\"\"Cliente GPU por polling saliente. La GPU no publica ningún servidor o túnel.\"\"\"\\nimport argparse\\nimport json\\nimport os\\nfrom pathlib import Path\\nimport threading\\nimport time\\nimport urllib.parse\\nimport requests\\nfrom render import render\\nfrom model_files import check_models\\n\\n\\ndef connect(base_url, vendor, image, minutes=30, motion_catalog=None):\\n token = os.environ.get(\\'HUMANA_WORKER_TOKEN\\', \\'\\')\\n if not token: raise RuntimeError(\\'Configura HUMANA_WORKER_TOKEN en Kaggle Secrets.\\')\\n parsed = urllib.parse.urlparse(base_url)\\n if parsed.scheme != \\'https\\' or not parsed.netloc or parsed.username or parsed.password or parsed.query:\\n raise ValueError(\\'La URL debe ser la raíz HTTPS de tu web Cloudflare.\\')\\n base_url = base_url.rstrip(\\'/\\')\\n vendor, image = Path(vendor).resolve(), Path(image).resolve()\\n check_models(vendor)\\n if not image.is_file(): raise FileNotFoundError(\\'Falta la imagen de identidad de la GPU.\\')\\n import torch\\n if not torch.cuda.is_available(): raise RuntimeError(\\'Activa la GPU antes de conectar.\\')\\n device = torch.cuda.get_device_name(0)\\n deadline = time.monotonic() + minutes * 60\\n headers = {\\'Authorization\\': \\'Bearer \\' + token}\\n motions = json.loads(Path(motion_catalog).read_text()) if motion_catalog else {}\\n work = vendor.parent / \\'cloudflare-jobs\\'; work.mkdir(exist_ok=True)\\n\\n def api(path, method=\\'POST\\', **kwargs):\\n r = requests.request(method, base_url + \\'/api/\\' + path, headers={**headers, **kwargs.pop(\\'headers\\', {})}, timeout=60, **kwargs)\\n if not r.ok:\\n try: message = r.json().get(\\'error\\', \\'Error API\\')\\n except ValueError: message = f\\'HTTP {r.status_code}\\'\\n raise RuntimeError(message)\\n return r\\n\\n print(f\\'Sesión GPU: {device}. Conectada durante unos {minutes} minutos.\\', flush=True)\\n while time.monotonic() < deadline:\\n try: job = api(\\'worker/claim\\', json={\\'device\\': device}).json()[\\'job\\']\\n except requests.RequestException:\\n print(\\'Conexión interrumpida; reintentando...\\', flush=True); time.sleep(10); continue\\n if not job: time.sleep(5); continue\\n job_id, lease = job[\\'id\\'], job[\\'leaseId\\']\\n stop = threading.Event(); lost = threading.Event()\\n\\n def renew():\\n while not stop.wait(25):\\n try: api(\\'worker/heartbeat\\', json={\\'device\\': device, \\'jobId\\': job_id, \\'leaseId\\': lease})\\n except Exception as error:\\n print(\\'No se pudo renovar la sesión:\\', type(error).__name__, flush=True)\\n lost.set()\\n heartbeat = threading.Thread(target=renew, daemon=True); heartbeat.start()\\n try:\\n folder = work / job_id; folder.mkdir(exist_ok=True)\\n audio = folder / \\'speech.mp3\\'\\n audio.write_bytes(api(f\\'worker/jobs/{job_id}/audio\\', \\'GET\\', headers={\\'X-Lease-Id\\': lease}).content)\\n gesture = job[\\'plan\\'][\\'gesture\\']\\n source = Path(motions[gesture]).resolve() if gesture in motions else None\\n result = render(vendor, image, audio, folder / \\'response.mp4\\', source=source, max_seconds=0)\\n if lost.is_set(): raise RuntimeError(\\'La GPU perdió la conexión durante el trabajo; no se sube un resultado con sesión incierta.\\')\\n if result.stat().st_size > 50 * 1024 * 1024: raise RuntimeError(\\'El resultado supera 50 MB. Reduce la resolución o la longitud de las respuestas.\\')\\n with result.open(\\'rb\\') as stream:\\n api(f\\'worker/jobs/{job_id}/result\\', \\'PUT\\', data=stream,\\n headers={\\'X-Lease-Id\\': lease, \\'Content-Type\\': \\'video/mp4\\', \\'X-Animation-Mode\\': \\'body-lips\\' if source else \\'lip-sync-only\\'})\\n print(\\'Vídeo entregado:\\', job_id, flush=True)\\n except Exception as error:\\n print(\\'Trabajo fallido:\\', type(error).__name__, str(error), flush=True)\\n try: api(f\\'worker/jobs/{job_id}/fail\\', headers={\\'X-Lease-Id\\': lease}, json={})\\n except Exception: pass\\n finally:\\n stop.set(); heartbeat.join(timeout=2)\\n print(\\'Sesión terminada. La cola y los vídeos siguen guardados en Cloudflare.\\', flush=True)\\n\\n\\nif __name__ == \\'__main__\\':\\n parser = argparse.ArgumentParser()\\n parser.add_argument(\\'--base-url\\', required=True)\\n parser.add_argument(\\'--vendor\\', type=Path, required=True)\\n parser.add_argument(\\'--image\\', type=Path, required=True)\\n parser.add_argument(\\'--minutes\\', type=int, default=30)\\n parser.add_argument(\\'--motion-catalog\\', type=Path)\\n connect(**vars(parser.parse_args()))\\n'}\n", "for name, content in PAYLOAD.items():\n", " (GPU / name).write_text(content, encoding='utf-8')\n", "ENV = os.environ.copy()\n", "for key in ['PYTHONPATH', 'PYTHONHOME', 'VIRTUAL_ENV', 'CONDA_PREFIX']:\n", " ENV.pop(key, None)\n", "ENV['PYTHONNOUSERSITE'] = '1'\n", "ENV['CUDA_VISIBLE_DEVICES'] = '0'\n", "ENV['PYTORCH_CUDA_ALLOC_CONF'] = 'max_split_size_mb:128'\n", "def run(args, **kwargs):\n", " subprocess.run([str(a) for a in args], check=True, env=ENV, **kwargs)\n", "print('Kernel:', sys.version.split()[0], '· Se mantiene sin modificar.')\n", "print('Carpeta:', ROOT)\n", "ENGINE = 'mimicmotion'\n", "VENDOR = ROOT / 'MimicMotion'\n", "PYTHON = ROOT / f'env-{ENGINE}/bin/python'" ], "id": "cell-04" }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 3. Instalar en un Python aislado" ], "id": "cell-05" }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "run([sys.executable, GPU / 'setup_runtime.py', '--root', ROOT, '--engine', ENGINE])\n", "# Los modelos y ONNX usan las bibliotecas CUDA del Torch aislado.\n", "torch_lib = PYTHON.parent.parent / 'lib/python3.10/site-packages/torch/lib'\n", "ENV['LD_LIBRARY_PATH'] = str(torch_lib) + ':' + ENV.get('LD_LIBRARY_PATH', '')\n", "run([PYTHON, GPU / 'doctor.py', '--engine', ENGINE])" ], "id": "cell-06" }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 4. Descargar los pesos\n", "La descarga de SVD es grande. Si requiere autenticación, activa USE_HF_SECRET después de configurar tu acceso al modelo." ], "id": "cell-07" }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "USE_HF_SECRET = False\n", "if USE_HF_SECRET:\n", " from kaggle_secrets import UserSecretsClient\n", " ENV['HF_TOKEN'] = UserSecretsClient().get_secret('HF_TOKEN')\n", "try:\n", " run([PYTHON, GPU / 'download_models.py', '--vendor', VENDOR, '--engine', ENGINE])\n", "finally:\n", " ENV.pop('HF_TOKEN', None)" ], "id": "cell-08" }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 5. Imagen y vídeo guía" ], "id": "cell-09" }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import shutil\n", "SEARCH_ROOTS = [Path('/kaggle/input'), Path('/kaggle/working'), Path.cwd()]\n", "def find_exact(filename, specified=''):\n", " if specified:\n", " candidate = Path(specified)\n", " if not candidate.is_file(): raise FileNotFoundError(f'No existe: {candidate}')\n", " return candidate\n", " matches = set()\n", " for root in SEARCH_ROOTS:\n", " if not root.exists(): continue\n", " for p in root.rglob(filename):\n", " if p.is_file() and ROOT not in p.parents:\n", " matches.add(p.resolve())\n", " if len(matches) != 1:\n", " raise FileNotFoundError(f'{filename}: encontrados {len(matches)}. Indica su ruta exacta en la celda de opciones.')\n", " return next(iter(matches))\n", "IMAGE = find_exact('chica.png', IMAGE_PATH)\n", "GUIDE = find_exact('guia.mp4', GUIDE_PATH)\n", "print('Imagen:', IMAGE, '\\nGuía:', GUIDE)" ], "id": "cell-10" }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 6. Generar movimiento" ], "id": "cell-11" }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "OUTPUT = ROOT / 'humana_movimiento.mp4'\n", "run([PYTHON, GPU / 'motion.py', '--vendor', VENDOR, '--image', IMAGE, '--guide', GUIDE,\n", " '--output', OUTPUT, '--seconds', GUIDE_SECONDS, '--resolution', RESOLUTION, '--steps', STEPS])" ], "id": "cell-12" }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 7. Revisar resultado" ], "id": "cell-13" }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from IPython.display import Video, display, FileLink\n", "display(Video(str(OUTPUT), embed=True))\n", "display(FileLink(str(OUTPUT)))" ], "id": "cell-14" }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 8. Combinar con labios (siguiente prueba)\n", "\n", "Descarga `humana_movimiento.mp4` y añádelo como entrada al notebook 01. Pon su ruta exacta en la opción `SOURCE_VIDEO_PATH` y vuelve a ejecutar la celda de generación. La voz debe durar como máximo lo mismo que el movimiento; ajusta `MAX_SECONDS` si estás probando un guía de 3 segundos. Se rechaza una fuente demasiado corta para evitar repetir el gesto hacia atrás.\n", "\n", "Revisa identidad, manos, mirada y postura. MuseTalk controla la zona facial del habla; la mirada, parpadeos y expresiones deben existir en la animación fuente o requerirán un módulo posterior. La unión de modelos no garantiza automáticamente expresiones semánticas ni transiciones invisibles.\n", "\n", "La inferencia de cuerpo es por lotes. Para uso continuo, genera primero una biblioteca de estados que empiecen y terminen en una postura neutral común. Fuente: [MimicMotion oficial](https://github.com/Tencent/MimicMotion)." ], "id": "cell-15" } ] }