leefer/aigc-home
1
0
Fork 0
Code Issues Pull Requests 11 Actions Packages Projects Releases Wiki Activity

refactor: remove FirstMotion/NextMotion latent chain

Browse Source
This commit is contained in:
Benjiyaya
2026-08-29 03:34:41 +08:00
parent 4d1d38a542
commit 59188a5b00
10 changed files with 2 additions and 590 deletions
Download Patch File Download Diff File Expand all files Collapse all files
calliope-backend/src/calliope/agent/video_agent.py
+1 -77
View File
@@ -11,10 +11,6 @@ from calliope.agent.prompts import (
minimax_h3_ref_fallback,
scene_video_prompt,
)
from calliope.comfyui.motion_context import (
classify_motion_role,
continued_duration,
)
from calliope.comfyui.parser import parse_dynamic_inputs
from calliope.comfyui.smart_fill import ref_image_slots, smart_fill_inputs
from calliope.config import settings
@@ -90,20 +86,6 @@ def _workflow_json(workflow: dict[str, Any] | None) -> dict[str, Any]:
return raw or {}
def _motion_pair(
workflows: list[dict[str, Any]],
) -> tuple[dict[str, Any] | None, dict[str, Any] | None]:
first: dict[str, Any] | None = None
nxt: dict[str, Any] | None = None
for wf in workflows:
role = classify_motion_role(_workflow_json(wf))
if role == "first" and first is None:
first = wf
elif role == "next" and nxt is None:
nxt = wf
return first, nxt
def _get_workflow(workflow_id: int | None = None) -> dict[str, Any] | None:
conn = get_db(settings.db_path)
try:
@@ -130,7 +112,6 @@ async def enqueue_video_jobs(
scene_ids: list[int] | None = None,
workflow_id: int | None = None,
input_values_override: dict[str, Any] | None = None,
continue_motion: bool | None = None,
) -> list[dict[str, Any]]:
await event_bus.publish(
"agent.thinking", {"message": "Queuing video jobs…", "project_id": project_id}
@@ -146,21 +127,6 @@ async def enqueue_video_jobs(
params.extend(scene_ids)
query += " ORDER BY order_index"
scenes = [row_to_dict(r) for r in conn.execute(query, params).fetchall()]
all_scenes = [
row_to_dict(r)
for r in conn.execute(
"SELECT * FROM scenes WHERE project_id = ? ORDER BY order_index",
(project_id,),
).fetchall()
]
video_wfs = [
row_to_dict(r)
for r in conn.execute(
"SELECT * FROM workflows WHERE is_enabled = 1 AND kind = 'video' ORDER BY id ASC"
).fetchall()
]
first_wf, next_wf = _motion_pair(video_wfs)
batch_ids = {s["id"] for s in scenes}
for scene in scenes:
# Supersede leftover pending jobs so a re-Generate always starts fresh
@@ -182,49 +148,12 @@ async def enqueue_video_jobs(
# self-deadlocks (sqlite3.OperationalError: database is locked).
conn.commit()
earlier = [s for s in all_scenes if s["order_index"] < scene["order_index"]]
prev = earlier[-1] if earlier else None
prev_ready = False
if prev:
if prev.get("video_path") or prev["id"] in batch_ids:
prev_ready = True
else:
pending = conn.execute(
"""
SELECT id FROM jobs
WHERE scene_id = ? AND kind = 'video'
AND status IN ('pending', 'running')
""",
(prev["id"],),
).fetchone()
prev_ready = bool(pending)
want_continue = bool(prev) and (continue_motion if continue_motion is not None else prev_ready)
if want_continue and prev and not prev_ready:
raise ValueError(
"Continue motion needs the previous clip "
f"(scene {prev.get('order_index')}) generated first."
)
wf_id = workflow_id or scene.get("workflow_id")
explicit = _get_workflow(wf_id) if wf_id else None
explicit_role = classify_motion_role(_workflow_json(explicit)) if explicit else None
if first_wf and next_wf and (explicit is None or explicit_role in ("first", "next")):
# Pair swap: clip 1 / continue-off → First; clips 2+ → Next.
workflow = next_wf if want_continue else first_wf
else:
workflow = explicit or _get_workflow(wf_id)
workflow = _get_workflow(wf_id)
workflow_json = _workflow_json(workflow)
inputs = parse_dynamic_inputs(workflow_json) if workflow_json else []
motion_role = classify_motion_role(workflow_json)
is_next = motion_role == "next" and want_continue
duration = scene.get("duration_sec")
if is_next:
duration = continued_duration(duration)
save_index = max(int(scene.get("order_index") or 1), 1)
load_index = max(int(prev["order_index"]), 1) if prev and want_continue else None
char_rows = conn.execute(
"""
@@ -268,8 +197,6 @@ async def enqueue_video_jobs(
prompt=prompt,
ref_images=ref_paths,
duration=duration,
clip_index_load=load_index,
clip_index_save=save_index,
extra=input_values_override,
)
else:
@@ -280,8 +207,6 @@ async def enqueue_video_jobs(
character_image=char_image,
location_image=loc_image,
duration=duration,
clip_index_load=load_index,
clip_index_save=save_index,
extra=input_values_override,
)
payload: dict[str, Any] = {"input_values": values, "prompt": prompt}
@@ -293,7 +218,6 @@ async def enqueue_video_jobs(
# The value the location/first-image ref slot holds right now, so the
# worker can find and replace that exact slot whatever the layout.
payload["chain_replace_value"] = loc_image
payload["continue_motion"] = bool(want_continue and motion_role == "next")
job = queue_manager.enqueue(
project_id=project_id,
kind="video",
calliope-backend/src/calliope/comfyui/motion_context.py
-78
View File
@@ -1,78 +0,0 @@
"""H3 Motion Context pairing: FirstMotion vs NextMotion, prefix isolation."""
from __future__ import annotations
from typing import Any, Literal
MotionRole = Literal["first", "next"]
CLASS_SAVE = "MiniMaxH3MotionContextSaveLatent"
CLASS_LOAD = "MiniMaxH3MotionContextLoadLatent"
CLASS_CONTEXT = "MiniMaxH3MotionContext"
CLASS_TRIM = "MiniMaxH3MotionContextTrim"
# H3 native rate. context_length 22 frames ≈ 0.92s removed by Trim on NextMotion.
CONTEXT_LENGTH = 22
CONTEXT_FPS = 24.0
def _class_types(workflow: dict[str, Any]) -> set[str]:
return {
str(node.get("class_type") or "")
for node in workflow.values()
if isinstance(node, dict)
}
def classify_motion_role(workflow: dict[str, Any]) -> MotionRole | None:
"""``first`` / ``next`` from live H3 Motion Context class_types, else None."""
classes = _class_types(workflow)
has_save = CLASS_SAVE in classes
has_load = CLASS_LOAD in classes
has_ctx = CLASS_CONTEXT in classes
if has_load and has_ctx:
return "next"
if has_save and not has_load:
return "first"
return None
def motion_prefix(project_id: int) -> tuple[str, str]:
"""(save filename_prefix, load latent_path folder) under Comfy output/."""
folder = f"calliope/p{int(project_id)}"
return f"{folder}/clip", folder
def apply_motion_context(
workflow: dict[str, Any],
*,
project_id: int,
continue_motion: bool,
) -> dict[str, Any]:
"""Rewrite Save/Load paths onto a per-project folder.
When ``continue_motion`` is false, drop ``context_latent`` so a NextMotion
graph can run as a chain start (Load is then unreachable).
"""
save_prefix, load_folder = motion_prefix(project_id)
for node in workflow.values():
if not isinstance(node, dict):
continue
class_type = node.get("class_type") or ""
inputs = dict(node.get("inputs") or {})
if class_type == CLASS_SAVE:
inputs["filename_prefix"] = save_prefix
node["inputs"] = inputs
elif class_type == CLASS_LOAD:
inputs["latent_path"] = load_folder
node["inputs"] = inputs
elif class_type == CLASS_CONTEXT and not continue_motion:
inputs.pop("context_latent", None)
node["inputs"] = inputs
return workflow
def continued_duration(duration_sec: int | float | None) -> float | None:
"""Add the Trim head so delivered length matches the scene duration."""
if duration_sec is None:
return None
return float(duration_sec) + (CONTEXT_LENGTH / CONTEXT_FPS)
calliope-backend/src/calliope/comfyui/roles.py
-1
View File
@@ -21,7 +21,6 @@ INPUT_ROLE_ALIASES: dict[str, frozenset[str]] = {
"image": frozenset({"image", "img"}),
"video": frozenset({"video", "vid"}),
"audio": frozenset({"audio", "sound", "sfx"}),
"clipindex": frozenset({"clipindex", "clip_index"}),
"seed": frozenset({"seed"}),
"duration": frozenset({"duration", "dur", "length", "seconds"}),
}
calliope-backend/src/calliope/comfyui/smart_fill.py
+1 -34
View File
@@ -43,34 +43,6 @@ def ref_audio_slots(inputs: list[dict[str, Any]]) -> list[dict[str, Any]]:
return _find_all_by_role(inputs, "audio")
def _clipindex_kind(inp: dict[str, Any]) -> str | None:
"""``load`` or ``save`` from the stripped title label. None if unclear."""
label = (inp.get("label") or "").lower()
if "load" in label:
return "load"
if "save" in label:
return "save"
return None
def _fill_clipindex(
inputs: list[dict[str, Any]],
values: dict[str, Any],
clip_index_load: int | None,
clip_index_save: int | None,
) -> None:
"""Write every ``(Input:clipindex)`` primitive. Timeline owns these values."""
for inp in _find_all_by_role(inputs, "clipindex"):
kind = _clipindex_kind(inp)
if kind == "load" and clip_index_load is not None:
values[str(inp["nodeId"])] = max(int(clip_index_load), 1)
elif kind == "save" and clip_index_save is not None:
values[str(inp["nodeId"])] = max(int(clip_index_save), 1)
elif kind is None and clip_index_save is not None:
# Single untagged clipindex (FirstMotion Save-only) → this clip.
values[str(inp["nodeId"])] = max(int(clip_index_save), 1)
def _legacy_prompt_node(inputs: list[dict[str, Any]]) -> dict[str, Any] | None:
"""Deprecated: untagged (Input) text nodes — prefer (Input:prompt)."""
for inp in inputs:
@@ -104,8 +76,6 @@ def smart_fill_inputs(
ref_videos: list[str] | None = None,
ref_audios: list[str] | None = None,
duration: int | float | None = None,
clip_index_load: int | None = None,
clip_index_save: int | None = None,
extra: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Build valuesByNodeId from workflow dynamic inputs + context.
@@ -201,7 +171,7 @@ def smart_fill_inputs(
reserved = {
str(inp["nodeId"])
for inp in inputs
if normalize_input_role(inp.get("role")) in ("duration", "clipindex")
if normalize_input_role(inp.get("role")) == "duration"
}
for k, v in extra.items():
if v is None:
@@ -212,7 +182,4 @@ def smart_fill_inputs(
continue
values[str(k)] = v
# Timeline-owned — always overwrite form defaults / extras.
_fill_clipindex(inputs, values, clip_index_load, clip_index_save)
return values
calliope-backend/src/calliope/models/schemas.py
-1
View File
@@ -201,4 +201,3 @@ class GenerateVideosRequest(BaseModel):
scene_ids: list[int] | None = None
workflow_id: int | None = None
input_values: dict[str, Any] | None = None
continue_motion: bool | None = None
calliope-backend/src/calliope/queue/worker.py
-120
View File
@@ -9,7 +9,6 @@ from typing import Any
from calliope import config
from calliope.comfyui.client import ComfyUIClient
from calliope.comfyui.dry_run import write_placeholder_mp4, write_placeholder_png
from calliope.comfyui.motion_context import apply_motion_context
from calliope.comfyui.patcher import patch_workflow
from calliope.db import get_db
from calliope.events.bus import event_bus
@@ -129,16 +128,7 @@ class QueueWorker:
raise RuntimeError("No workflow found for job")
input_values = payload.get("input_values") or {}
if payload.get("chain_from_prev") and kind == "video":
input_values = await self._apply_chain_from_prev(
job, payload, workflow, dict(input_values)
)
patched = patch_workflow(workflow, input_values)
patched = apply_motion_context(
patched,
project_id=project_id,
continue_motion=bool(payload.get("continue_motion")),
)
patched = await client.prepare_media_inputs(patched)
prompt_id = await client.queue_prompt(patched)
@@ -171,116 +161,6 @@ class QueueWorker:
finally:
await client.close()
async def _apply_chain_from_prev(
self,
job: dict[str, Any],
payload: dict[str, Any],
workflow: dict[str, Any],
input_values: dict[str, Any],
) -> dict[str, Any]:
"""Start this clip from the previous scene's last frame.
Runs at job time — in a batch the previous scene has rendered by now even
though it had not at enqueue time. Extracts the nearest earlier scene's
last frame and swaps it into the slot that held the location/first-image
reference (found by value; falls back to the location-role node). If no
previous clip exists, the job proceeds with its original refs.
"""
import subprocess
from calliope.comfyui.parser import parse_dynamic_inputs
from calliope.comfyui.roles import normalize_input_role
project_id = job["project_id"]
order_index = payload.get("scene_order_index")
prev_clip: str | None = None
conn = get_db(config.settings.db_path)
try:
rows = conn.execute(
"""
SELECT video_path FROM scenes
WHERE project_id = ? AND order_index < ? AND video_path IS NOT NULL
ORDER BY order_index DESC
""",
(project_id, order_index if order_index is not None else 0),
).fetchall()
for r in rows:
from pathlib import Path as _P
if r["video_path"] and _P(r["video_path"]).exists():
prev_clip = r["video_path"]
break
finally:
conn.close()
if not prev_clip:
await event_bus.publish(
"job.progress",
{
"job_id": job["id"],
"project_id": project_id,
"message": "Chain-from-previous: no earlier clip yet — using the scene's own refs",
},
)
return input_values
from pathlib import Path
frame_dir = config.settings.assets_dir / str(project_id) / "chain"
frame_dir.mkdir(parents=True, exist_ok=True)
frame = frame_dir / f"scene{job.get('scene_id')}_from_prev.png"
proc = subprocess.run(
["ffmpeg", "-y", "-v", "error", "-sseof", "-0.1", "-i", prev_clip,
"-update", "1", "-vframes", "1", str(frame)],
capture_output=True, text=True, timeout=60,
)
if proc.returncode != 0 or not frame.exists():
await event_bus.publish(
"job.progress",
{
"job_id": job["id"],
"project_id": project_id,
"message": f"Chain-from-previous: frame extraction failed ({proc.stderr[:120]}) — using original refs",
},
)
return input_values
# Find the slot to replace: the node holding the enqueue-time location value,
# else the workflow's location-role input (may be empty when no location is set).
replace_value = payload.get("chain_replace_value")
target_node: str | None = None
if replace_value:
for node_id, value in input_values.items():
if value == replace_value:
target_node = str(node_id)
break
if target_node is None:
for inp in parse_dynamic_inputs(workflow):
if normalize_input_role(inp.get("role")) == "location":
target_node = str(inp["nodeId"])
break
if target_node is None:
await event_bus.publish(
"job.progress",
{
"job_id": job["id"],
"project_id": project_id,
"message": "Chain-from-previous: no ref slot found to carry the frame — using original refs",
},
)
return input_values
input_values[target_node] = str(frame)
await event_bus.publish(
"job.progress",
{
"job_id": job["id"],
"project_id": project_id,
"message": "Chain-from-previous: starting from the previous clip's last frame",
},
)
return input_values
async def _dry_run(self, job: dict[str, Any], payload: dict[str, Any]) -> list[str]:
project_id = job["project_id"]
kind = job["kind"]
calliope-backend/src/calliope/routers/jobs.py
-1
View File
@@ -63,7 +63,6 @@ async def generate_videos(
scene_ids=body.scene_ids,
workflow_id=body.workflow_id,
input_values_override=body.input_values,
continue_motion=body.continue_motion,
)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
calliope-backend/src/calliope/routers/workflows.py
-3
View File
@@ -5,7 +5,6 @@ from typing import Any
from fastapi import APIRouter, HTTPException
from calliope.comfyui.motion_context import classify_motion_role
from calliope.comfyui.parser import parse_dynamic_inputs, parse_dynamic_outputs
from calliope.comfyui.profiles import detect_prompt_profile
from calliope.config import settings
@@ -21,7 +20,6 @@ def _serialize_workflow(row: Any) -> dict[str, Any]:
data["input_schema"] = json.loads(data["input_schema"]) if data.get("input_schema") else []
data["output_schema"] = json.loads(data["output_schema"]) if data.get("output_schema") else []
data["is_enabled"] = bool(data.get("is_enabled"))
data["motion_role"] = classify_motion_role(data["workflow_json"])
return data
@@ -33,7 +31,6 @@ async def analyze_workflow(payload: WorkflowAnalyze) -> dict[str, Any]:
"inputs": inputs,
"outputs": outputs,
"suggested_profile": detect_prompt_profile(payload.workflow_json),
"motion_role": classify_motion_role(payload.workflow_json),
}
calliope-backend/tests/test_motion_context.py
-238
View File
@@ -1,238 +0,0 @@
"""H3 Motion Context pair detect, prefix rewrite, duration bump."""
from __future__ import annotations
from calliope.comfyui.motion_context import (
apply_motion_context,
classify_motion_role,
continued_duration,
motion_prefix,
)
from calliope.comfyui.registry import class_to_input_kind, class_to_patch_field
def test_load_video_kind_and_patch_field():
assert class_to_input_kind("LoadVideo") == "video"
assert class_to_patch_field("LoadVideo") == "file"
assert class_to_patch_field("VHS_LoadVideo") == "video"
def test_classify_first_vs_next():
first = {
"274": {"class_type": "MiniMaxH3MotionContextSaveLatent", "inputs": {}},
"272": {"class_type": "MiniMaxH3ReferenceToVideo", "inputs": {}},
}
nxt = {
"274": {"class_type": "MiniMaxH3MotionContextSaveLatent", "inputs": {}},
"275": {"class_type": "MiniMaxH3MotionContext", "inputs": {"context_latent": ["277", 0]}},
"277": {"class_type": "MiniMaxH3MotionContextLoadLatent", "inputs": {}},
"276": {"class_type": "MiniMaxH3MotionContextTrim", "inputs": {}},
}
plain = {"3": {"class_type": "SaveVideo", "inputs": {}}}
assert classify_motion_role(first) == "first"
assert classify_motion_role(nxt) == "next"
assert classify_motion_role(plain) is None
def test_apply_prefix_and_unwire():
wf = {
"274": {
"class_type": "MiniMaxH3MotionContextSaveLatent",
"inputs": {"filename_prefix": "h3_context/clip", "clip_index": 2},
},
"277": {
"class_type": "MiniMaxH3MotionContextLoadLatent",
"inputs": {"latent_path": "h3_context", "clip_index": 1},
},
"275": {
"class_type": "MiniMaxH3MotionContext",
"inputs": {"context_latent": ["277", 0], "conditioning": ["272", 0]},
},
}
save_prefix, load_folder = motion_prefix(12)
started = apply_motion_context(wf, project_id=12, continue_motion=False)
assert started["274"]["inputs"]["filename_prefix"] == save_prefix
assert started["277"]["inputs"]["latent_path"] == load_folder
assert "context_latent" not in started["275"]["inputs"]
chained = apply_motion_context(
{
"275": {
"class_type": "MiniMaxH3MotionContext",
"inputs": {"context_latent": ["277", 0]},
},
},
project_id=12,
continue_motion=True,
)
assert chained["275"]["inputs"]["context_latent"] == ["277", 0]
def test_continued_duration():
assert continued_duration(5) == 5 + 22 / 24
assert continued_duration(None) is None
def test_prepare_media_inputs_uploads_video(tmp_path):
import asyncio
from unittest.mock import AsyncMock
from calliope.comfyui.client import ComfyUIClient
video = tmp_path / "walk.mp4"
video.write_bytes(b"fake-mp4")
client = ComfyUIClient(base_url="http://comfy.test")
client.upload_video = AsyncMock(return_value="calliope/walk.mp4")
wf = {
"10": {
"class_type": "LoadVideo",
"inputs": {"file": str(video)},
}
}
try:
out = asyncio.run(client.prepare_media_inputs(wf))
assert out["10"]["inputs"]["file"] == "calliope/walk.mp4"
client.upload_video.assert_awaited_once()
finally:
asyncio.run(client.close())
FIRST_MOTION = {
"274": {
"class_type": "MiniMaxH3MotionContextSaveLatent",
"inputs": {"filename_prefix": "h3_context/clip"},
},
"209": {
"class_type": "PrimitiveStringMultiline",
"inputs": {"value": ""},
"_meta": {"title": "Prompt (Input:prompt)"},
},
"5": {
"class_type": "PrimitiveFloat",
"inputs": {"value": 5},
"_meta": {"title": "(Input:duration) Duration"},
},
"282": {
"class_type": "PrimitiveInt",
"inputs": {"value": 1},
"_meta": {"title": "(input:clipindex) Save Motion Context Clip Index"},
},
}
NEXT_MOTION = {
**FIRST_MOTION,
"275": {
"class_type": "MiniMaxH3MotionContext",
"inputs": {"context_latent": ["277", 0]},
},
"277": {"class_type": "MiniMaxH3MotionContextLoadLatent", "inputs": {}},
"276": {"class_type": "MiniMaxH3MotionContextTrim", "inputs": {}},
"281": {
"class_type": "PrimitiveInt",
"inputs": {"value": 1},
"_meta": {"title": "(input:clipindex) Load Motion Context Clip Index"},
},
}
def test_analyze_and_enqueue_picks_first_then_next(client):
first_a = client.post("/api/workflows/analyze", json={"workflow_json": FIRST_MOTION})
next_a = client.post("/api/workflows/analyze", json={"workflow_json": NEXT_MOTION})
assert first_a.status_code == 200
assert first_a.json()["motion_role"] == "first"
assert next_a.status_code == 200
assert next_a.json()["motion_role"] == "next"
first = client.post(
"/api/workflows",
json={
"name": "FirstMotion",
"kind": "video",
"workflow_json": FIRST_MOTION,
"prompt_profile": "prose",
},
).json()
nxt = client.post(
"/api/workflows",
json={
"name": "NextMotion",
"kind": "video",
"workflow_json": NEXT_MOTION,
"prompt_profile": "prose",
},
).json()
assert first["motion_role"] == "first"
assert nxt["motion_role"] == "next"
pid = client.post("/api/projects", json={"title": "Hallway"}).json()["id"]
s1 = client.post(
f"/api/projects/{pid}/scenes",
json={"heading": "Walk", "order_index": 1, "duration_sec": 5},
).json()
s2 = client.post(
f"/api/projects/{pid}/scenes",
json={"heading": "Door", "order_index": 2, "duration_sec": 5},
).json()
client.post("/api/jobs/pause")
try:
r = client.post(
f"/api/jobs/projects/{pid}/generate-videos",
json={"scene_ids": [s1["id"], s2["id"]]},
)
assert r.status_code == 200
jobs = r.json()["jobs"]
assert len(jobs) == 2
j1, j2 = jobs
assert j1["workflow_id"] == first["id"]
assert j1["payload"]["continue_motion"] is False
assert j1["payload"]["input_values"]["282"] == 1
assert j1["payload"]["input_values"]["5"] == 5
assert j2["workflow_id"] == nxt["id"]
assert j2["payload"]["continue_motion"] is True
assert j2["payload"]["input_values"]["281"] == 1
assert j2["payload"]["input_values"]["282"] == 2
assert j2["payload"]["input_values"]["5"] == 5 + 22 / 24
finally:
client.post("/api/jobs/resume")
def test_continue_motion_fails_without_previous_clip(client):
client.post(
"/api/workflows",
json={
"name": "FirstMotion",
"kind": "video",
"workflow_json": FIRST_MOTION,
"prompt_profile": "prose",
},
)
client.post(
"/api/workflows",
json={
"name": "NextMotion",
"kind": "video",
"workflow_json": NEXT_MOTION,
"prompt_profile": "prose",
},
)
pid = client.post("/api/projects", json={"title": "No prev"}).json()["id"]
client.post(
f"/api/projects/{pid}/scenes",
json={"heading": "One", "order_index": 1, "duration_sec": 5},
)
s2 = client.post(
f"/api/projects/{pid}/scenes",
json={"heading": "Two", "order_index": 2, "duration_sec": 5},
).json()
client.post("/api/jobs/pause")
try:
r = client.post(
f"/api/jobs/projects/{pid}/generate-videos",
json={"scene_ids": [s2["id"]], "continue_motion": True},
)
assert r.status_code == 400
assert "previous clip" in r.json()["detail"].lower()
finally:
client.post("/api/jobs/resume")
calliope-backend/tests/test_roles.py
-37
View File
@@ -128,40 +128,3 @@ def test_video_and_audio_roles():
)
assert values["10"] == "/tmp/walk.mp4"
assert values["11"] == "/tmp/bed.mp3"
def test_clipindex_load_vs_save_by_label():
workflow = {
"281": {
"inputs": {"value": 1},
"class_type": "PrimitiveInt",
"_meta": {"title": "(input:clipindex) Load Motion Context Clip Index"},
},
"282": {
"inputs": {"value": 2},
"class_type": "PrimitiveInt",
"_meta": {"title": "(input:clipindex) Save Motion Context Clip Index"},
},
}
inputs = parse_dynamic_inputs(workflow)
values = smart_fill_inputs(
inputs,
clip_index_load=3,
clip_index_save=4,
extra={"281": 1, "282": 2},
)
assert values["281"] == 3
assert values["282"] == 4
def test_clipindex_never_zero():
workflow = {
"282": {
"inputs": {"value": 2},
"class_type": "PrimitiveInt",
"_meta": {"title": "(input:clipindex) Save Motion Context Clip Index"},
},
}
inputs = parse_dynamic_inputs(workflow)
values = smart_fill_inputs(inputs, clip_index_save=0)
assert values["282"] == 1