🦝
MotionProで画像からアニメーション作った: 4.WEB実装
前回はCOLABもしくはローカルでコマンドをたたいてMotionProで動画を出すやり方を紹介しました
今回はすでにフロントのHTML5＋Javascriptはできておりかつ、バックでコマンドでモデルを動かし動く動画を作成するとおまでできているので、それらをつなぐバックエンドの開設です。
基本的にいまあるものをそのままつかいます。
APIサーバーにはFlaskをつかってますがFaskAPIとかでもいいかもしれません。Flaskは個人的に柔軟でかつApacheやNginxとWSGIでつなぐ仕組みがあるの、APACHEやNGINXはLet's EncryptのようなSSLをシンプルに導入できたりアプリのポートをProxyで柔軟につないだりとシンプルにサービス展開できる要素があるので私は好んで利用してます。
まずflaskを導入します、今回はAPACHEのPORXYを利用してFLASKのPORT 5001につなぐので、WSGI系のライブラリは導入不要です。
(motionpro)$ pip install flask
今回は、BASE64で画像ファイルを受け取りますが、前回のコマンドをそのまま利用するため、いったんファイルに落としてそれを読ませる形にします。
コマンドの一部を抜粋します。
**mp_client.py
# -------------------- main ---------------------------------
def main(img, mask, traj, out, gifname=None):
ensure_dirname(out)
img_pil = Image.open(img).convert("RGB")
mask_pil = Image.open(mask).convert("RGB")
H,W = img_pil.height, img_pil.width
tracks = json.load(open(traj))
drag_t = build_drag_mask(tracks, H, W, mask_pil).to(DEVICE)
first = transforms.ToTensor()(img_pil)[None].to(DEVICE)*2 - 1
runner = Drag(DEVICE, CKPT_PATH, CFG_PATH, MODEL_LEN)
vid = runner.forward_sample(drag_t, first, BUCKET_ID) # (T,3,H,W)
frames = [tensor2pil(f) for f in vid]
gif_path = os.path.join(out, f"motionpro_{uuid.uuid4().hex}.gif")
data2file(frames, gif_path, printable=False, duration=1/8, override=True)
# gifname が指定されていればリネーム
if gifname:
final_path = os.path.join(out, gifname)
# 親ディレクトリがなければ作成
os.makedirs(os.path.dirname(final_path), exist_ok=True)
os.replace(gif_path, final_path)
else:
final_path = gif_path
print("✔ GIF saved:", final_path)
# CLI -------------------------------------------------------
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("--img", required=True)
parser.add_argument("--mask", required=True)
parser.add_argument("--traj", required=True)
parser.add_argument("--out", default="output_cli")
parser.add_argument("--gifname", default=None, help="（任意）最終GIFのファイル名")
args = parser.parse_args()
main(args.img, args.mask, args.traj, args.out, args.gifname)
Flaskではうけとったbase64をinputに保存してそのパスを引数にmainを呼ぶようにします。
以下のようなapp_flask.pyを用意します.
# app_flask.py
import os, io, json, re, base64, random, sys
from datetime import datetime
from typing import Any, Dict
from PIL import Image
from flask import Flask, request, jsonify
from flask_cors import CORS # 別オリジンから叩くなら有効化
# mp_client.py を import（app_flask.py と同じディレクトリ前提）
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
if BASE_DIR not in sys.path:
sys.path.insert(0, BASE_DIR)
from mp_cli import main as motion_main
# ========= 設定 =========
SAVE_INPUT_DIR = "YOURPATH/MotionPro/input" # 入力の保存先。適宜環境に合わせて変更
SAVE_OUTPUT_DIR = "YOURPATH/MotionPro/output" # 出力GIFの保存先,適宜環境に合わせて変更
AUTO_RESIZE_MASK = True # 画像と同サイズにマスクを合わせる
MAX_CONTENT_MB = 64 # 受信サイズ上限（必要に応じて調整）
# ========================
app = Flask(__name__)
app.config['MAX_CONTENT_LENGTH'] = MAX_CONTENT_MB * 1024 * 1024
CORS(app) # 逆プロキシ（/api）経由で別オリジン扱いになるなら有効化
DATAURL_RE = re.compile(r"^data:image/[^;]+;base64,(.+)$", re.IGNORECASE)
def ensure_dir(path: str) -> None:
os.makedirs(path, exist_ok=True)
def b64_to_bytes(s: str) -> bytes:
"""
data URL（data:image/png;base64,....）でも、生の Base64 文字列でもOK。
"""
s = s.strip()
m = DATAURL_RE.match(s)
payload = m.group(1) if m else s
payload = payload.replace("\n", "").replace("\r", "")
try:
return base64.b64decode(payload, validate=True)
except Exception:
# 改行含みやパディング不足などに寛容にしたい場合
return base64.b64decode(payload)
def decode_image_b64_to_pil(b64: str, mode: str = "RGB") -> Image.Image:
data = b64_to_bytes(b64)
im = Image.open(io.BytesIO(data))
return im.convert(mode) if mode else im
def ts_token() -> str:
"""
ファイル名の先頭に使うトークン:
YYMMDDHHmmss + 4桁ランダム（0埋め）
"""
ts = datetime.now().strftime("%y%m%d%H%M%S")
r4 = f"{random.randint(0, 9999):04d}"
return f"{ts}{r4}"
def save_payload(image_b64: str, mask_b64: str, trajectory: Any) -> Dict[str, Any]:
"""
Base64 image, Base64 mask, JSON/list trajectory を受け取って
input/ に指定の命名規則で保存。
戻り値: 保存パスや画像サイズなど
"""
ensure_dir(SAVE_INPUT_DIR)
# 画像・マスクを PIL に
img_pil = decode_image_b64_to_pil(image_b64, mode="RGB")
# マスクは RGB で保存（mp_client が mask を RGB 前提で読むため）
mask_pil = decode_image_b64_to_pil(mask_b64, mode="RGB")
W, H = img_pil.size
if AUTO_RESIZE_MASK and mask_pil.size != (W, H):
mask_pil = mask_pil.resize((W, H), Image.NEAREST)
# 軌跡は list でも JSON 文字列でも可
if isinstance(trajectory, str):
traj_obj = json.loads(trajectory)
else:
traj_obj = trajectory
token = ts_token() # YYMMDDHHmmssrrrr
img_path = os.path.join(SAVE_INPUT_DIR, f"{token}_image.png")
mask_path = os.path.join(SAVE_INPUT_DIR, f"{token}_mask.png")
traj_path = os.path.join(SAVE_INPUT_DIR, f"{token}_track.json")
# 保存
img_pil.save(img_path) # PNG
mask_pil.save(mask_path) # PNG
with open(traj_path, "w", encoding="utf-8") as f:
json.dump(traj_obj, f, ensure_ascii=False, indent=2)
return {
"ok": True,
"token": token,
"width": W,
"height": H,
"paths": {
"image": img_path,
"mask": mask_path,
"trajectory": traj_path,
}
}
def run_motion_and_encode(token: str, img_path: str, mask_path: str, traj_path: str) -> Dict[str, str]:
"""
mp_cli.main を呼んで GIF を生成し、Base64(data URL) で返す。
生成先: SAVE_OUTPUT_DIR/{token}_motion.gif
戻り: { "gif_path": ..., "gif_b64": "data:image/gif;base64,..." }
"""
ensure_dir(SAVE_OUTPUT_DIR)
gifname = f"{token}_motion.gif"
# 1) 推論実行（mp_cli.main は戻り値を返さないので、gifname を指定して確定パスに落とす）
motion_main(img_path, mask_path, traj_path, SAVE_OUTPUT_DIR, gifname)
# 2) 読み出して base64 化
final_path = os.path.join(SAVE_OUTPUT_DIR, gifname)
with open(final_path, "rb") as f:
gif_bytes = f.read()
gif_b64 = "data:image/gif;base64," + base64.b64encode(gif_bytes).decode("ascii")
return {"gif_path": final_path, "gif_b64": gif_b64}
@app.get("/")
def get_root():
return "OK"
@app.post("/add_motion")
def api_add_motion():
"""
受信 JSON 例:
{
"image_b64": "data:image/png;base64,...", // または生のBase64文字列
"mask_b64": "data:image/png;base64,...",
"trajectory": [[[x,y], [x,y], ...], ...] // list でも JSON文字列でもOK
}
レスポンス:
{
"ok": true,
"token": "YYMMDDHHmmssrrrr",
"width": 512,
"height": 320,
"paths": {
"image": "input/YYMMDDHHmmssrrrr_image.png",
"mask": "input/YYMMDDHHmmssrrrr_mask.png",
"trajectory": "input/YYMMDDHHmmssrrrr_track.json"
},
"gif_path": "output/YYMMDDHHmmssrrrr_motion.gif",
"gif_b64": "data:image/gif;base64,...."
}
"""
data = request.get_json(silent=True)
if not data:
return jsonify({"ok": False, "error": "invalid JSON"}), 400
image_b64 = data.get("image_b64")
mask_b64 = data.get("mask_b64")
traj = data.get("trajectory")
if not image_b64 or not mask_b64 or traj is None:
return jsonify({"ok": False, "error": "missing fields: image_b64/mask_b64/trajectory"}), 400
try:
# 1) 保存
info = save_payload(image_b64, mask_b64, traj)
# 2) 推論→GIF→Base64
gif_info = run_motion_and_encode(
info["token"],
info["paths"]["image"],
info["paths"]["mask"],
info["paths"]["trajectory"]
)
# 3) 結果まとめて返す
resp = {**info, **gif_info}
return jsonify(resp), 200
except Exception as e:
# 例外は文字列化して返す
return jsonify({"ok": False, "error": str(e)}), 500
if __name__ == "__main__":
# 本番では Apache のリバースプロキシや WSGI/Gunicorn を推奨
app.run(host="0.0.0.0", port=5001, debug=False)
これでこれを立ち上げればOKです。私はcronで自動立ち上げにしてます。
@reboot /opt/python3.10.16/MotionPro/run_flask.sh
**run_flask.sh
cd /YOURPATH/MotionPro/
$HOME/.pyenv/versions/3.10.16/envs/motionpro/bin/python /YOURPATH/MotionPro/app_flask.py > /YOURPATH/MotionPro/app_flask.log 2>&1
先ほどのgenerate_motionpro_input.html をapacheのDOCUMENT ROOTにおいておきます。
また133行目のFLASK＿BASEをWEBサーバーのURLに変えます。
const FLASK_BASE = 'https://your-api-site-url';
APACHE2にはLET'S ENCRYPTでSSLをいれたため、HTTPSからHTTPへの非同期POST接続ができなかったためそこはPROSYで回避してます。
**/etc/apache2/conf-enabled/flask-proxy.conf
ProxyPreserveHost On
RequestHeader set X-Forwarded-Proto "https"
# /api/ 以下を Flask のルートに中継
ProxyPass /api/ http://127.0.0.1:5001/
ProxyPassReverse /api/ http://127.0.0.1:5001/
# ルート直下 /api へのアクセスも安全側で拾う
ProxyPass /api http://127.0.0.1:5001/
ProxyPassReverse /api http://127.0.0.1:5001/
# アプリの場所（静的HTML等を置いている場所とは別でOK）
<Directory /YOURPATH/MotionPro>
Require all granted
</Directory>
# ログ
ErrorLog ${APACHE_LOG_DIR}/flask_proxy_error.log
CustomLog ${APACHE_LOG_DIR}/flask_proxy_access.log combined
これでほととり一応サービスにはなっています。先ほどのhtmlをWEBサーバーに置いた後そこにアクセスしてSend Flask to add motion を押します
１－２分で以下のようなアニメーションがかえります。
次回は、いよいよ学習に挑戦しようと思います。おそらく手持ちの環境ではまともあことはできませんがどのように学習を進めるのか参考になれば幸いです。以下が学習実行手順の記事です。
Discussion