Implementing a MiniMax-H3 Multimodal Video and Audio Generation Pipeline with ComfyUI APIs

class H3Graph:
   def __init__(self, schema, unet, te, lora=None):
       self.s, self.g, self._id = schema, {}, 0
       self.unet, self.te, self.lora = unet, te, lora
   def node(self, cls, **inputs):
       self.s.check(cls, inputs)
       self._id += 1
       nid = str(self._id)
       self.g[nid] = {"class_type": cls, "inputs": inputs}
       return nid
  
   def _backbone(self):
       model = self.node("UNETLoader", unet_name=self.unet, weight_dtype="default")
       if self.lora:
           model = self.node("LoraLoaderModelOnly", model=[model, 0],
                             lora_name=self.lora, strength_model=1.0)
       if CFG["SIGMA_SHIFT"]:
           sv, sa = CFG["SIGMA_SHIFT"]
           model = self.node("MiniMaxH3SigmaShift", model=[model, 0],
                             shift_video=float(sv), shift_audio=float(sa))
       clip = self.node("CLIPLoader", clip_name=self.te, type="minimax", device="default")
       vvae = self.node("VAELoader", vae_name=VAE_VIDEO)
       avae = self.node("VAELoader", vae_name=VAE_AUDIO)
       return model, clip, vvae, avae
   def _tail(self, model, cond, latent, vvae, avae):
       turbo = bool(self.lora)
       steps = CFG["TURBO_STEPS"] if turbo else CFG["STEPS"]
       sampler_name = CFG["TURBO_SAMPLER"] if turbo else CFG["SAMPLER"]
       sched = CFG["TURBO_SCHEDULER"] if turbo else CFG["SCHEDULER"]
       noise = self.node("RandomNoise", noise_seed=int(CFG["SEED"]))
       samp = self.node("KSamplerSelect", sampler_name=sampler_name)
       sig = self.node("BasicScheduler", model=[model, 0], scheduler=sched,
                       steps=steps, denoise=1.0)
       guider = self.node("BasicGuider", model=[model, 0], conditioning=[cond[0], cond[1]])
       out = self.node("SamplerCustomAdvanced", noise=[noise, 0], guider=[guider, 0],
                       sampler=[samp, 0], sigmas=[sig, 0], latent_image=[latent[0], latent[1]])
      
       frames = self.node("VAEDecode", samples=[out, 0], vae=[vvae, 0])
       audio = self.node("VAEDecodeAudio", samples=[out, 0], vae=[avae, 0])
       vid = self.node("CreateVideo", images=[frames, 0], audio=[audio, 0], fps=24)
       self.node("SaveVideo", video=[vid, 0], filename_prefix="MiniMaxH3/h3",
                 format="auto", codec="auto")
       print(f"  sampling: {steps} steps, {sampler_name}/{sched}")
       return self.g
   def _load_image(self, uploaded_name):
       return self.node("LoadImage", image=uploaded_name, upload="image")
  
   def t2v_or_flf2v(self, w, h, length, first=None, last=None):
       self.s.require("MiniMaxH3ImageToVideo", "SamplerCustomAdvanced", "SaveVideo")
       model, clip, vvae, avae = self._backbone()
       kw = {}
       if first:
           kw["first_frame"] = [self._load_image(first), 0]  
       if last:
           kw["last_frame"] = [self._load_image(last), 0]    
       n = self.node("MiniMaxH3ImageToVideo", clip=[clip, 0], vae=[vvae, 0],
                     prompt=CFG["PROMPT"], width=w, height=h, length=length, **kw)
       return self._tail(model, (n, 0), (n, 1), vvae, avae)
   def r2v(self, w, h, length, ref_names):
       self.s.require("MiniMaxH3ReferenceToVideo")
       model, clip, vvae, avae = self._backbone()
       slots = self.s.autogrow("MiniMaxH3ReferenceToVideo", "ref_image_", len(ref_names))
       refs = {slot: [self._load_image(nm), 0] for slot, nm in zip(slots, ref_names)}
       print(f"  reference slots: {list(refs)}")
       n = self.node("MiniMaxH3ReferenceToVideo", clip=[clip, 0], vae=[vvae, 0],
                     audio_vae=[avae, 0], prompt=CFG["PROMPT"], width=w, height=h,
                     length=length, ref_image_size=CFG["REF_IMAGE_SIZE"], **refs)
       return self._tail(model, (n, 0), (n, 1), vvae, avae)