`anim/animate_wan.py`

31 lines
import base64,json,os,time,urllib.request
tok=[l.split('=',1)[1].strip().strip('"\'') for l in open(os.path.expanduser('~/Projects/secrets-manager/.env')) if l.startswith('REPLICATE_API_TOKEN=')][0]
def api(url,data=None):
    r=urllib.request.Request(url,data=json.dumps(data).encode() if data else None,headers={"Authorization":f"Bearer {tok}","Content-Type":"application/json"})
    return json.load(urllib.request.urlopen(r,timeout=300))
HERE=os.path.dirname(os.path.abspath(__file__))  # paths resolve from this file, not the cwd
def P(name): return os.path.join(HERE,name)
def poll(pr,interval,deadline_s):
    # Replicate bills at create time: never lose the id, retry transient poll errors, bound the wait.
    end=time.time()+deadline_s
    while pr["status"] not in("succeeded","failed","canceled"):
        if time.time()>end: raise SystemExit(f"TIMEOUT after {deadline_s}s — recover later: GET {pr['urls']['get']}")
        time.sleep(interval)
        for a in range(5):
            try: pr=api(pr["urls"]["get"]); break
            except Exception as e:
                print(f"poll error (attempt {a+1}/5): {e}"); time.sleep(interval*(a+1))
        else: raise SystemExit(f"poll failed 5x — recover later: GET {pr['urls']['get']}")
    return pr
img="data:image/jpeg;base64,"+base64.b64encode(open(P('edited.jpg'),'rb').read()).decode()
p=("Animate this black-and-white pen-and-ink editorial cartoon, keep the hand-drawn ink style throughout, static camera. "
   "The deadpan politician keeps reading aloud from the huge telephone directory, lips moving in a monotone, expression never changing; he licks a finger and turns a page. "
   "Loose pages tear free and flutter slowly through the air. The melted wall clock keeps oozing, its drips sliding slowly down the wall into the spreading puddle on the floor. "
   "The legislators in the background slowly nod off, heads drooping onto their chests, one yawns. Slow, dry, absurd, deadpan comedy timing.")
pr=api("https://api.replicate.com/v1/models/wan-video/wan-2.2-i2v-fast/predictions",{"input":{"prompt":p,"image":img,"num_frames":121,"frames_per_second":12,"resolution":"720p","go_fast":True}})
print("id",pr["id"],pr["status"])
pr=poll(pr,10,1800)
print(pr["status"],pr.get("error"),pr.get("metrics"))
if pr["status"]=="succeeded": urllib.request.urlretrieve(pr["output"],P('raw_wan.mp4')); print('saved')
else: raise SystemExit(1)