TouchDesignerでCLIPを使って映像解析する

この投稿は「PY - PARTY TECH STUDIO Advent Calendar 2025」の9記事目です。

はじめに

TouchDesigner上で再生している動画をCLIPで解析して、その結果をほぼリアルタイムで可視化できないか試してみました。

CLIPは、テキスト指示だけで画像を分類し、「この画像と、このテキストはどれくらい関連しているか」という類似度の計算ができるAIモデルです。

引用：Zero-Shot Image Classification

この類似度スコアを使って、動画内の「人の多さ」「車の交通量」「バスやタクシーの存在」を検出して、TouchDesigner上で表示します。

準備

今回はMacで動作確認しています。
TouchDesignerからCLIPを扱うために、こんな構成にしました。

やることはシンプルで、

• CLIPをローカルサーバーとして起動する
• TDで動画のフレームを画像として保存し、画像パスとテキストをサーバーに送信する
• サーバー側でCLIP推論を行い、テキストごとの類似度スコアをJSONで返す
• TDで受け取ったスコアを使い、数値やUIとして可視化する

という流れです。

1. Python環境のセットアップ

必要なライブラリをインストールします。

pip install fastapi uvicorn transformers torch pillow

2. CLIPサーバーの実装

/scoreに画像パスと文章の配列を渡すと、文章ごとの類似度を返すだけのシンプルな作りです。

今回は日本語のCLIPモデルを使いました。

clip_service.py

from fastapi import FastAPI, Query, HTTPException
from PIL import Image
import torch
from transformers import AutoModel, AutoTokenizer, AutoImageProcessor
import uvicorn
from pathlib import Path

MODEL_ID = "line-corporation/clip-japanese-base"
HOST = "127.0.0.1"
PORT = 8010

# Macならmps、それ以外はcpu
device = "mps" if torch.backends.mps.is_available() else "cpu"

app = FastAPI()

model = None
tokenizer = None
image_processor = None


def lazy_load():
    global model, tokenizer, image_processor
    if model is not None:
        return

    tokenizer = AutoTokenizer.from_pretrained(MODEL_ID, trust_remote_code=True)
    image_processor = AutoImageProcessor.from_pretrained(MODEL_ID, trust_remote_code=True)
    model = AutoModel.from_pretrained(
        MODEL_ID,
        trust_remote_code=True,
        low_cpu_mem_usage=False,
    ).to(device)
    model.eval()


@app.get("/health")
def health():
    lazy_load()
    return {"ok": True, "device": device, "model": MODEL_ID}


@app.get("/score")
def score(in_path: str, texts: list[str] = Query(...)):
    lazy_load()

    # textsは配列でも、"a|b|c" の1本文字列でも受ける
    labels = [t.strip() for t in texts if t and t.strip()]
    if len(labels) == 1 and "|" in labels[0]:
        labels = [t.strip() for t in labels[0].split("|") if t.strip()]

    if not labels:
        return {"ok": False, "error": "no labels"}

    p = Path(in_path).expanduser()
    if not p.exists():
        raise HTTPException(status_code=400, detail=f"file not found: {p}")

    try:
        img = Image.open(p).convert("RGB")
    except Exception as e:
        raise HTTPException(status_code=400, detail=f"invalid image: {e}")

    image_inputs = image_processor(images=img, return_tensors="pt")
    text_inputs = tokenizer(labels, padding=True, truncation=True)

    image_inputs = {k: v.to(device) for k, v in image_inputs.items()}
    text_inputs = {k: torch.as_tensor(v, dtype=torch.long, device=device) for k, v in text_inputs.items()}

    with torch.no_grad():
        image_features = model.get_image_features(**image_inputs)
        text_features = model.get_text_features(**text_inputs)

        # 類似度が安定しやすいように正規化してから内積
        image_features = image_features / image_features.norm(dim=-1, keepdim=True)
        text_features = text_features / text_features.norm(dim=-1, keepdim=True)

        sims = (image_features @ text_features.T)[0]

    # TD側でsoftmaxしてprobを作るので、ここではsimsだけ返す
    return {"ok": True, "labels": labels, "sims": sims.detach().cpu().tolist()}


if __name__ == "__main__":
    uvicorn.run(app, host=HOST, port=PORT, log_level="info")

3. サーバーを起動

python clip_service.py

正常に動いているか確認できたら、TouchDesigner側の実装に進みます。

TouchDesignerでの実装

全体像はこんな感じです。

• 左エリア：サーバー通信とスコア加工
• 中央エリア：UI表示
• 右エリア：動画読み込みとUIの合成

1. 動画素材の準備

夜の渋谷スクランブル交差点で人と車が行き交う動画を使いました。
MovieFileIn TOPで読み込んで、これがCLIPに投げる動画ソースになります。

2. CLIPサーバーとの通信

スコア取得部分の構成は次のようにしました。
fetch DATでプロンプト生成 → HTTPリクエスト → clip_scores DATに格納する流れです。

fetch DAT

import threading, urllib.parse, urllib.request, json, math

in_top  = op('/project1/movie_for_clip')
in_path = '/tmp/td_in.jpg'

OUT_DAT = 'clip_scores'
BASE_URL = "http://127.0.0.1:8010/score"

TEMPLATE = "夜の交差点に、{}"
CLASSES = {
    "many_people": {"text": "多くの人が歩いて混雑している"},
    "many_cars":   {"text": "車が走っている"},
    "bus":         {"text": "バスが走っている"},
    "taxi":        {"text": "タクシーが走っている"},
}

SCALE = 80.0
TIMEOUT = 10
SKIP_IF_BUSY = True

_busy = False

def softmax(xs):
    if not xs:
        return []
    m = max(xs)
    exps = [math.exp(x - m) for x in xs]
    s = sum(exps) if exps else 1.0
    return [e / s for e in exps]

def main():
    global _busy
    if in_top is None:
        print("movie_top not found")
        return

    if SKIP_IF_BUSY and _busy:
        return

    _busy = True

    # 1フレームを画像として保存 → パスをAPIに渡す
    try:
        in_top.save(in_path, quality=0.8)
    except Exception as e:
        _busy = False
        print("save error:", e)
        return

    class_keys = list(CLASSES.keys())
    prompts = [TEMPLATE.format(CLASSES[k]["text"]) for k in class_keys]
    texts_param = "|".join(prompts)

    def http_job():
        global _busy
        try:
            url = BASE_URL + "?" + urllib.parse.urlencode({
                "in_path": in_path,
                "texts": texts_param,
            })

            raw = urllib.request.urlopen(url, timeout=TIMEOUT).read().decode('utf-8')
            data = json.loads(raw)

            if not data.get("ok", False):
                raise RuntimeError("clip api error: " + repr(data))

            sims = data.get("sims", [])
            if not sims or len(sims) != len(class_keys):
                raise RuntimeError("invalid sims length: " + repr(len(sims)))

            # TD側で調整したいので、sims → probはここで作る
            probs = softmax([float(s) * SCALE for s in sims])

            def update_dat():
                t = op(OUT_DAT)
                if t is None:
                    print("Table DAT not found:", OUT_DAT)
                    return

                t.clear()
                t.appendRow(["class_key", "prob", "sim", "prompt"])
                for k, prob, sim, prompt in zip(class_keys, probs, sims, prompts):
                    t.appendRow([k, float(prob), float(sim), prompt])

            run(update_dat, delayFrames=1, fromOP=me)

        except Exception as e:
            run("print(" + repr("clip error: " + repr(e)) + ")", delayFrames=1, fromOP=me)
        finally:
            _busy = False

    threading.Thread(target=http_job, daemon=True).start()

main()

プロンプトについて（余談）

当初は、openai/clip-vit-base-patch32モデルを使って、一般的な a photo of a {label} 形式を試しましたが、夜間映像ではスコアが全体的に低く出てしまいました。

日本語のCLIPモデルに変更し、自然な日本語表現に調整したところスコアが安定するようになりました。

実際に使用したプロンプト

夜の交差点に、多くの人が歩いて混雑している
夜の交差点に、車が走っている
夜の交差点に、バスが走っている
夜の交差点に、タクシーが走っている

3. スコアの加工

生のスコアをそのまま使うと、待機中の人も検出されてしまい people_score が常に高く出る傾向がありました。

見た目で分かりやすくするために、Script CHOPでスコアを加工しています。

script1 CHOP - callback

RATIO_GAMMA  = 0.6   # people/trafficの出方を調整
CONF_TH      = 0.4   # people+trafficが小さいときは全体を弱める
BUS_SIM_TH   = 0.2   # バス検出の閾値
TAXI_SIM_TH  = 0.3   # タクシー検出の閾値
SIM_MARGIN   = 0.01

DAT_NAME = "clip_scores"

def onSetupParameters(scriptOp):
    page = scriptOp.appendCustomPage('Custom')
    page.appendFloat('Dummy', label='Dummy')

def onCook(scriptOp):
    scriptOp.clear()

    dat = op(DAT_NAME)
    if dat is None or dat.numRows < 2:
        _write(scriptOp, people_score=0.0, traffic_score=0.0, bus_flag=0.0, taxi_flag=0.0)
        return

    # people/trafficは相対値でバー表示に使う
    people  = _get(dat, "many_people", col_name="prob")
    traffic = _get(dat, "many_cars",   col_name="prob")

    # bus/taxiは相対値だと下がりやすいのでsimを使う
    bus_sim  = _get(dat, "bus",  col_name="sim")
    taxi_sim = _get(dat, "taxi", col_name="sim")

    people_score, traffic_score = _people_traffic(people, traffic)

    bus_flag  = 1.0 if (bus_sim  > BUS_SIM_TH  and bus_sim  > taxi_sim + SIM_MARGIN) else 0.0
    taxi_flag = 1.0 if (taxi_sim > TAXI_SIM_TH and taxi_sim > bus_sim  + SIM_MARGIN) else 0.0

    _write(
        scriptOp,
        people_score=people_score,
        traffic_score=traffic_score,
        bus_flag=bus_flag,
        taxi_flag=taxi_flag,
    )

def _get(dat, key, col_name="prob"):
    col = _col_index(dat, col_name)
    if col < 0:
        return 0.0
    for r in range(1, dat.numRows):
        try:
            if str(dat[r, 0]) == key:
                return float(dat[r, col])
        except:
            pass
    return 0.0

def _col_index(dat, name):
    try:
        for c in range(dat.numCols):
            if str(dat[0, c]) == name:
                return c
    except:
        pass
    return -1

def _people_traffic(people, traffic):
    total = people + traffic
    if total < 1e-6:
        return 0.0, 0.0

    conf = min(1.0, total / CONF_TH)

    p = people / total
    t = traffic / total

    people_ratio  = (p ** RATIO_GAMMA)
    traffic_ratio = (t ** RATIO_GAMMA)

    return people_ratio * conf, traffic_ratio * conf

def _write(scriptOp, **vals):
    for name, v in vals.items():
        scriptOp.appendChan(name)[0] = float(v)

def onGetCookLevel(scriptOp):
    return CookLevel.ON_CHANGE

4. UI表示

加工したスコアとフラグをUI要素にバインドして表示します。

UI要素	使用している値	バインド先
Peopleバー	people_score	scaleX
Trafficバー	traffic_score	scaleX
Bus表示	bus_flag	opacity
Taxi表示	taxi_flag	opacity

各カテゴリごとにこんな構造を作って、
最初に読み込んだ動画とComposite TOPで合成しています。

5. 定期実行の設定

ほぼリアルタイムで解析するために、Timer CHOPからfetchを定期実行しています。

TOPをjpg保存 → FastAPI → CLIP推論 → JSON → DAT更新 という処理が走るので、
間隔を短くしすぎると処理が詰まって不安定になります。

今回は1秒に1回にしてますが、
サーバー側のログやTouchDesignerの出力を見ながら調整するといいと思います。

アウトプット

実際に動かしてみた様子がこちらです！

まとめ

プロンプト次第で結果が変わるので、
そのあたりの調整はまだまだ試行錯誤が必要そうです。

とりあえず動くものは作れたので、スコアと音を連動させたり、映像の演出に使ったり、色々試してみると面白いかもしれません。

公開されている学習済みモデルを、技術検証として試しました。
AIモデルの利用は著作権などの点もあるので、扱いには配慮する必要があります。