【Stable Video Diffusion】ローカルでの使い方や料金体系、商用利用について解説

Stability AI は、非独占的、全世界的、譲渡不可、サブライセンス不可、取り消し可能なロイヤリティフリーの制限付きライセンスをお客様に付与します。
ソフトウェア製品に組み込まれている Stability AI の知的財産または Stability AI が所有するその他の権利に基づいて、
商用または生産使用以外の目的でソフトウェア製品の複製、配布、派生作品を作成することはできません。     

■Google Colab（無料プラン）
GPU：T4

#@title Setup

!nvidia-smi

!git clone https://github.com/Stability-AI/generative-models.git

# install required packages from pypi

# !pip3 install -r generative-models/requirements/pt2.txt

# manually install only necesarry packages for colab

!wget https://gist.githubusercontent.com/mkshing/4ad40699756d996ba6b3f7934e6ca532/raw/3f0094272c7a2bd3eb5f1a0db91bed582c9e8f01/requirements.txt

!pip3 install -r requirements.txt

!pip3 install -e generative-models

!pip3 install -e git+https://github.com/Stability-AI/datapipelines.git@main#egg=sdata

!pip3 install gradio

#@title Colab hack for SVD

# !pip uninstall -y numpy

# !pip install -U numpy

!mkdir -p /content/scripts/util/detection

!ln -s /content/generative-models/scripts/util/detection/p_head_v1.npz /content/scripts/util/detection/p_head_v1.npz

!ln -s /content/generative-models/scripts/util/detection/w_head_v1.npz /content/scripts/util/detection/w_head_v1.npz

#@title Login HuggingFace to download weights
#@markdown Please make sure to fill in the form in the model cards and accept it.
from huggingface_hub import login
login()

# @title Download weights
import os
import subprocess
from huggingface_hub import hf_hub_download
version = "svd-xt" #@param ["svd", "svd-xt", "svd-xt-1-1"]
TYPE2PATH = {
    "svd": ["stabilityai/stable-video-diffusion-img2vid", "svd.safetensors"],
    "svd-xt": ["stabilityai/stable-video-diffusion-img2vid-xt", "svd_xt.safetensors"],
    "svd-xt-1-1": ["stabilityai/stable-video-diffusion-img2vid-xt-1-1", "svd_xt_1_1.safetensors"],
}
repo_id, fname = TYPE2PATH[version]
ckpt_dir = "/content/checkpoints"
ckpt_path = os.path.join(ckpt_dir, fname)
# @markdown This will take several minutes. <br>
# @markdown **Reference:**
# @markdown * `svd`: [stabilityai/stable-video-diffusion-img2vid](https://huggingface.co/stabilityai/stable-video-diffusion-img2vid) for 14 frames generation
# @markdown * `svd-xt`: [stabilityai/stable-video-diffusion-img2vid-xt](https://huggingface.co/stabilityai/stable-video-diffusion-img2vid-xt) for 25 frames generation
# @markdown * `svd-xt-1-1`: [stabilityai/stable-video-diffusion-img2vid-xt-1-1](https://huggingface.co/stabilityai/stable-video-diffusion-img2vid-xt-1-1) for 25 frames generation with fixed conditioning at 6FPS and Motion Bucket Id 127

os.makedirs("checkpoints", exist_ok=True)
if os.path.exists(ckpt_path):
  print("Already downloaded")
else:
  hf_hub_download(
      repo_id=repo_id,
      filename=fname,
      local_dir=ckpt_dir,
  )

#@title Load Model
import sys
from omegaconf import OmegaConf

import torch

sys.path.append("generative-models")
from sgm.util import default, instantiate_from_config
from scripts.util.detection.nsfw_and_watermark_dectection import DeepFloydDataFiltering

def load_model(
    config: str,
    device: str,
    num_frames: int,
    num_steps: int,
    ckpt_path: str = None,
):
    config = OmegaConf.load(config)
    config.model.params.conditioner_config.params.emb_models[
        0
    ].params.open_clip_embedding_config.params.init_device = device
    config.model.params.sampler_config.params.num_steps = num_steps
    config.model.params.sampler_config.params.guider_config.params.num_frames = (
        num_frames
    )
    if ckpt_path is not None:
        config.model.params.ckpt_path = ckpt_path
        print(f"Changed `ckpt_path` to {ckpt_path}")
    with torch.device(device):
        model = instantiate_from_config(config.model).to(device).eval().requires_grad_(False)

    filter = DeepFloydDataFiltering(verbose=False, device=device)
    return model, filter


if version == "svd":
    num_frames = 14
    num_steps = 25
    # output_folder = default(output_folder, "outputs/simple_video_sample/svd/")
    model_config = "generative-models/scripts/sampling/configs/svd.yaml"
elif "svd-xt" in version:
    num_frames = 25
    num_steps = 30
    # output_folder = default(output_folder, "outputs/simple_video_sample/svd_xt/")
    model_config = "generative-models/scripts/sampling/configs/svd_xt.yaml"
else:
    raise ValueError(f"Version {version} does not exist.")

device = "cuda" if torch.cuda.is_available() else "cpu"
model, filter = load_model(
    model_config,
    device,
    num_frames,
    num_steps,
    ckpt_path,
)
# move models expect unet to cpu
model.conditioner.cpu()
model.first_stage_model.cpu()
# change the dtype of unet
model.model.to(dtype=torch.float16)
torch.cuda.empty_cache()
model = model.requires_grad_(False)

# @title Sampling function
import math
import os
from glob import glob
from pathlib import Path
from typing import Optional

import cv2
import numpy as np
import torch
from einops import rearrange, repeat
from fire import Fire

from PIL import Image
from torchvision.transforms import ToTensor
from torchvision.transforms import functional as TF

from sgm.inference.helpers import embed_watermark
from sgm.util import default, instantiate_from_config


def get_unique_embedder_keys_from_conditioner(conditioner):
    return list(set([x.input_key for x in conditioner.embedders]))


def get_batch(keys, value_dict, N, T, device, dtype=None):
    batch = {}
    batch_uc = {}

    for key in keys:
        if key == "fps_id":
            batch[key] = (
                torch.tensor([value_dict["fps_id"]])
                .to(device, dtype=dtype)
                .repeat(int(math.prod(N)))
            )
        elif key == "motion_bucket_id":
            batch[key] = (
                torch.tensor([value_dict["motion_bucket_id"]])
                .to(device, dtype=dtype)
                .repeat(int(math.prod(N)))
            )
        elif key == "cond_aug":
            batch[key] = repeat(
                torch.tensor([value_dict["cond_aug"]]).to(device, dtype=dtype),
                "1 -> b",
                b=math.prod(N),
            )
        elif key == "cond_frames":
            batch[key] = repeat(value_dict["cond_frames"], "1 ... -> b ...", b=N[0])
        elif key == "cond_frames_without_noise":
            batch[key] = repeat(
                value_dict["cond_frames_without_noise"], "1 ... -> b ...", b=N[0]
            )
        else:
            batch[key] = value_dict[key]

    if T is not None:
        batch["num_video_frames"] = T

    for key in batch.keys():
        if key not in batch_uc and isinstance(batch[key], torch.Tensor):
            batch_uc[key] = torch.clone(batch[key])
    return batch, batch_uc



def sample(
    input_path: str = "assets/test_image.png",  # Can either be image file or folder with image files
    resize_image: bool = False,
    num_frames: Optional[int] = None,
    num_steps: Optional[int] = None,
    fps_id: int = 6,
    motion_bucket_id: int = 127,
    cond_aug: float = 0.02,
    seed: int = 23,
    decoding_t: int = 14,  # Number of frames decoded at a time! This eats most VRAM. Reduce if necessary.
    device: str = "cuda",
    output_folder: Optional[str] = "/content/outputs",
    skip_filter: bool = False,
):
    """
    Simple script to generate a single sample conditioned on an image `input_path` or multiple images, one for each
    image file in folder `input_path`. If you run out of VRAM, try decreasing `decoding_t`.
    """
    torch.manual_seed(seed)

    path = Path(input_path)
    all_img_paths = []
    if path.is_file():
        if any([input_path.endswith(x) for x in ["jpg", "jpeg", "png"]]):
            all_img_paths = [input_path]
        else:
            raise ValueError("Path is not valid image file.")
    elif path.is_dir():
        all_img_paths = sorted(
            [
                f
                for f in path.iterdir()
                if f.is_file() and f.suffix.lower() in [".jpg", ".jpeg", ".png"]
            ]
        )
        if len(all_img_paths) == 0:
            raise ValueError("Folder does not contain any images.")
    else:
        raise ValueError
    all_out_paths = []
    for input_img_path in all_img_paths:
        with Image.open(input_img_path) as image:
            if image.mode == "RGBA":
                image = image.convert("RGB")
            if resize_image and image.size != (1024, 576):
                print(f"Resizing {image.size} to (1024, 576)")
                image = TF.resize(TF.resize(image, 1024), (576, 1024))
            w, h = image.size

            if h % 64 != 0 or w % 64 != 0:
                width, height = map(lambda x: x - x % 64, (w, h))
                image = image.resize((width, height))
                print(
                    f"WARNING: Your image is of size {h}x{w} which is not divisible by 64. We are resizing to {height}x{width}!"
                )

            image = ToTensor()(image)
            image = image * 2.0 - 1.0

        image = image.unsqueeze(0).to(device)
        H, W = image.shape[2:]
        assert image.shape[1] == 3
        F = 8
        C = 4
        shape = (num_frames, C, H // F, W // F)
        if (H, W) != (576, 1024):
            print(
                "WARNING: The conditioning frame you provided is not 576x1024. This leads to suboptimal performance as model was only trained on 576x1024. Consider increasing `cond_aug`."
            )
        if motion_bucket_id > 255:
            print(
                "WARNING: High motion bucket! This may lead to suboptimal performance."
            )

        if fps_id < 5:
            print("WARNING: Small fps value! This may lead to suboptimal performance.")

        if fps_id > 30:
            print("WARNING: Large fps value! This may lead to suboptimal performance.")

        value_dict = {}
        value_dict["motion_bucket_id"] = motion_bucket_id
        value_dict["fps_id"] = fps_id
        value_dict["cond_aug"] = cond_aug
        value_dict["cond_frames_without_noise"] = image
        value_dict["cond_frames"] = image + cond_aug * torch.randn_like(image)
        value_dict["cond_aug"] = cond_aug
        # low vram mode
        model.conditioner.cpu()
        model.first_stage_model.cpu()
        torch.cuda.empty_cache()
        model.sampler.verbose = True

        with torch.no_grad():
            with torch.autocast(device):
                model.conditioner.to(device)
                batch, batch_uc = get_batch(
                    get_unique_embedder_keys_from_conditioner(model.conditioner),
                    value_dict,
                    [1, num_frames],
                    T=num_frames,
                    device=device,
                )
                c, uc = model.conditioner.get_unconditional_conditioning(
                    batch,
                    batch_uc=batch_uc,
                    force_uc_zero_embeddings=[
                        "cond_frames",
                        "cond_frames_without_noise",
                    ],
                )
                model.conditioner.cpu()
                torch.cuda.empty_cache()

                # from here, dtype is fp16
                for k in ["crossattn", "concat"]:
                    uc[k] = repeat(uc[k], "b ... -> b t ...", t=num_frames)
                    uc[k] = rearrange(uc[k], "b t ... -> (b t) ...", t=num_frames)
                    c[k] = repeat(c[k], "b ... -> b t ...", t=num_frames)
                    c[k] = rearrange(c[k], "b t ... -> (b t) ...", t=num_frames)
                for k in uc.keys():
                    uc[k] = uc[k].to(dtype=torch.float16)
                    c[k] = c[k].to(dtype=torch.float16)

                randn = torch.randn(shape, device=device, dtype=torch.float16)

                additional_model_inputs = {}
                additional_model_inputs["image_only_indicator"] = torch.zeros(
                    2, num_frames
                ).to(device, )
                additional_model_inputs["num_video_frames"] = batch["num_video_frames"]

                for k in additional_model_inputs:
                    if isinstance(additional_model_inputs[k], torch.Tensor):
                        additional_model_inputs[k] = additional_model_inputs[k].to(dtype=torch.float16)

                def denoiser(input, sigma, c):
                    return model.denoiser(
                        model.model, input, sigma, c, **additional_model_inputs
                    )

                samples_z = model.sampler(denoiser, randn, cond=c, uc=uc)
                samples_z.to(dtype=model.first_stage_model.dtype)
                ##

                model.en_and_decode_n_samples_a_time = decoding_t
                model.first_stage_model.to(device)
                samples_x = model.decode_first_stage(samples_z)
                samples = torch.clamp((samples_x + 1.0) / 2.0, min=0.0, max=1.0)
                model.first_stage_model.cpu()
                torch.cuda.empty_cache()

                os.makedirs(output_folder, exist_ok=True)
                base_count = len(glob(os.path.join(output_folder, "*.mp4")))
                video_path = os.path.join(output_folder, f"{base_count:06d}.mp4")
                writer = cv2.VideoWriter(
                    video_path,
                    cv2.VideoWriter_fourcc(*"MP4V"),
                    fps_id + 1,
                    (samples.shape[-1], samples.shape[-2]),
                )

                samples = embed_watermark(samples)
                if not skip_filter:
                    samples = filter(samples)
                else:
                    print("WARNING: You have disabled the NSFW/Watermark filter. Please do not expose unfiltered results in services or applications open to the public.")
                vid = (
                    (rearrange(samples, "t c h w -> t h w c") * 255)
                    .cpu()
                    .numpy()
                    .astype(np.uint8)
                )
                for frame in vid:
                    frame = cv2.cvtColor(frame, cv2.COLOR_RGB2BGR)
                    writer.write(frame)
                writer.release()
                all_out_paths.append(video_path)
    return all_out_paths

# @title Do the Run!
# @markdown Generation takes about 10 mins for 25 frames on T4 (Colab free plan). Please be patient...
# @markdown (V100 takes about 3 mins.)
import gradio as gr
import random


def infer(input_path: str, resize_image: bool, n_frames: int, n_steps: int, seed: str, decoding_t: int, fps_id: int, motion_bucket_id: int, cond_aug: float, skip_filter: bool = False) -> str:
  if seed == "random":
    seed = random.randint(0, 2**32)
  if version == "svd-xt-1-1":
    if fps_id != 6:
      print("[WARNING] svd-xt-1-1 was fine-tuned in fixed conditioning (`fps_id=6`, `motion_bucket_id=127`)! The performance may vary compared to SVD 1.0.")
    if motion_bucket_id != 127:
      print("[WARNING] svd-xt-1-1 was fine-tuned in fixed conditioning (`fps_id=6`, `motion_bucket_id=127`)! The performance may vary compared to SVD 1.0.")
  seed = int(seed)
  output_paths = sample(
    input_path=input_path,
    resize_image=resize_image,
    num_frames=n_frames,
    num_steps=n_steps,
    fps_id=fps_id,
    motion_bucket_id=motion_bucket_id,
    cond_aug=cond_aug,
    seed=seed,
    decoding_t=decoding_t,  # Number of frames decoded at a time! This eats most VRAM. Reduce if necessary.
    device=device,
    skip_filter=skip_filter,
  )
  return output_paths[0]


with gr.Blocks() as demo:
  with gr.Column():
    image = gr.Image(label="input image", type="filepath")
    resize_image = gr.Checkbox(label="resize to optimal size", value=True)
    btn = gr.Button("Run")
    with gr.Accordion(label="Advanced options", open=False):
      n_frames = gr.Number(precision=0, label="number of frames", value=num_frames)
      n_steps = gr.Number(precision=0, label="number of steps", value=num_steps)
      seed = gr.Text(value="random", label="seed (integer or 'random')",)
      decoding_t = gr.Number(precision=0, label="number of frames decoded at a time", value=2)
      fps_id = gr.Number(precision=0, label="frames per second", value=6)
      motion_bucket_id = gr.Number(precision=0, value=127, label="motion bucket id")
      cond_aug = gr.Number(label="condition augmentation factor", value=0.02)
      skip_filter = gr.Checkbox(value=False, label="skip nsfw/watermark filter")
  with gr.Column():
    video_out = gr.Video(label="generated video")
  examples = [
      ["https://user-images.githubusercontent.com/33302880/284758167-367a25d8-8d7b-42d3-8391-6d82813c7b0f.png"],
  ]
  inputs = [image, resize_image, n_frames, n_steps, seed, decoding_t, fps_id, motion_bucket_id, cond_aug, skip_filter]
  outputs = [video_out]
  btn.click(infer, inputs=inputs, outputs=outputs)
  gr.Examples(examples=examples, inputs=inputs, outputs=outputs, fn=infer)
  demo.queue().launch(debug=True, share=True, show_error=True)