Pika H3 Flex: Fast test opens plainly off, and examples can't ship a half-set switch

The Fast test row set the video loader to 1 MP when off, but the workflow ships the loader at 0.41 MP, so the row opened striped 'changed' with neither side picked. Its off side is now 0.41.

workflow_contract now fails any Workflow Switches row whose saved values sit on neither side, are half on and half off, or disagree with whether its nodes run.

CHANGELOG.md

@@ -4,6 +4,16 @@ All notable changes to ComfyUI-AusBoss are documented here.
 
 ## Unreleased
 
+- **New example: Pika H3 Flex.** One MiniMax H3 workflow for pictures,
+  video and sound. Give each picture a job on H3 Prompt Writer, or switch on
+  Video to swap someone into a clip or copy its moves. Keep the clip's sound
+  carries the clip's own words and music through a swap, and lets them steer
+  the mouths. Video swaps stop at 10 seconds and come out at about 480p,
+  where they hold best. Voice and Song use the new Load Audio node.
+- **H3 Prompt Writer swaps two people.** With Swap the person and two person
+  pictures, Picture 1 takes the place of the person on the left and Picture 2
+  the person on the right. Type who goes to change that, one line each, such
+  as `@1 the man in the hat` and `@2 the woman in red`.
 - **New node: Load Audio.** Load a sound file and keep only the part you
   want. The node shows the sound as a wave: drag the IN and OUT handles to
   trim it, and press play to hear just that part. Set Max length to the

nodes/_h3_cast_helpers.py

@@ -37,6 +37,12 @@ NEEDS_A_PERSON = {
     MOVES: "To copy the moves, connect a picture of your person and pick person for it.",
     SWAP: "To swap the person, connect a picture of the new person and pick person for it.",
 }
+# A hat or a shirt with no person picture: the user wants the video's own person to wear it.
+DRESS_THE_VIDEO_PERSON = (
+    "To put this on the person in your video, load a still of them from the video as Picture 1 "
+    "and pick person for it. Give the hat or outfit its own picture, pick outfit or thing, and type "
+    "\"@1 wears @2\" with video job swap the person."
+)
 
 # (what the label is, how it is kept). {picture} is H3's number for the picture.
 JOB_WORDS = {
@@ -88,14 +94,20 @@ MOVES_KEPT = (
     "keep the source performer's face, body, clothes, props, background and colours out of the target.\n"
     "<Video 1>: attribute_transfer - apply only its camera movement to the target scene."
 )
-SWAP_IS = (
-    "<Video 1> supplies the place, the action, the timing, the camera movement and the look.\n"
-    "{replaced} is {who} in <Video 1>, selected for replacement."
-)
-SWAP_KEPT = (
+SWAP_VIDEO = "<Video 1> supplies the place, the action, the timing, the camera movement and the look."
+SWAP_IS = "{replaced} is {who} in <Video 1>, selected for replacement."
+SWAP_VIDEO_KEPT = (
     "<Video 1>: partially_preserved - keep the place, the action, the timing, the camera movement and the look; "
-    "change only the person.\n"
-    "{replaced}: removed - replace {who} with {main} throughout."
+    "change only the {people}."
+)
+SWAP_KEPT = "{replaced}: removed - replace {who} with {main} throughout."
+ONE_GOES = "the original person"
+# Where each person stands when two or more go and the script doesn't say, from left to right like the pictures.
+SPOTS = {2: ("left", "right"), 3: ("left", "centre", "right"), 4: ("far left", "left", "right", "far right")}
+SAYS_WHO = re.compile(
+    r"^(?:is|are|replaces|takes\s+the\s+place\s+of|goes\s+in\s+for|swaps?\s+(?:in\s+)?for|in\s+place\s+of|for|as)\b\s*"
+    r"|^[:=\-\u2013\u2014>]+\s*",
+    re.I,
 )
 SWAPPED = "attribute_transfer - apply the defined appearance to the original performance; the background of {picture} is not transferred."
 KEEP_IS = (
@@ -115,6 +127,11 @@ REVOICE_KEPT = (
 )
 
 
+def lower_first(text):
+    first = text.split(" ", 1)[0]
+    return text if text[:1] == "<" or (len(first) > 1 and first.isupper()) else text[:1].lower() + text[1:]
+
+
 def subject_label(number):
     return f"<Subject {number}>"
 
@@ -135,11 +152,16 @@ class Picture:
     subject: int = 0  # A picture that only marks the last frame has none.
 
 
+def swapped_in(jobs):
+    """How many people a swap puts in: one for each person or sheet picture."""
+    return max(1, sum(job in (PERSON, SHEET) for job in jobs.values()))
+
+
 def number_pictures(jobs, video=""):
     """Connected pictures in order, from {number on the node: job}. In a swap
-    the person being replaced comes first, as <Subject 1>."""
+    the people being replaced come first, as <Subject 1>, <Subject 2> and on."""
     pictures = []
-    subjects = 1 if video == SWAP else 0
+    subjects = swapped_in(jobs) if video == SWAP else 0
     for picture, slot in enumerate(sorted(jobs), 1):
         subjects += jobs[slot] != END
         pictures.append(Picture(slot, jobs[slot], picture, 0 if jobs[slot] == END else subjects))
@@ -154,8 +176,12 @@ class Cast:
         self.video = video
         last = max((picture.subject for picture in pictures), default=0)
         self.moves = subject_label(last + 1) if video == MOVES else ""
+        # In a swap each person or sheet picture takes one person's place in the video, in picture order. The script
+        # can name who goes: "the child in the seat", or "@1 the man on the left. @2 the woman in red".
+        self.swappers = [picture for picture in pictures if picture.job in (PERSON, SHEET)] if video == SWAP else []
+        goes = len(self.swappers)
+        self.whos = [f"the person on the {spot}" for spot in SPOTS[goes]] if goes > 1 else [ONE_GOES]
         self.replaced = subject_label(1) if video == SWAP else ""
-        self.who = "the original person"  # In a swap the script can name who goes: "the child in the seat".
         self.in_video = subject_label(last + 1) if video in (KEEP, REVOICE) else ""
         self.people = [picture for picture in pictures if picture.job in (PERSON, SHEET, START)]
         self.subjects = {picture.slot: subject_label(picture.subject) for picture in pictures if picture.subject}
@@ -163,6 +189,39 @@ class Cast:
     def of(self, *jobs):
         return [picture for picture in self.pictures if picture.job in jobs]
 
+    @property
+    def who(self):
+        return self.whos[0]
+
+    @who.setter
+    def who(self, words):
+        self.whos[0] = words
+
+    @property
+    def swaps(self):
+        """(who goes, the words for them, who comes in) for each person a swap replaces."""
+        coming = [subject_label(picture.subject) for picture in self.swappers] or [self.main]
+        return [(subject_label(number), who, new) for number, (who, new) in enumerate(zip(self.whos, coming), 1)]
+
+    def name_who_goes(self, lines):
+        """Who each new person replaces, from the lines typed in a swap. With one, the words are who goes. With more,
+        each @ names one, or else the lines go in picture order; anything else keeps left to right."""
+        lines = [line.strip() for line in lines if line.strip(" .")]
+        text = " ".join(lines)
+        if len(self.whos) == 1:
+            self.who = lower_first(SAYS_WHO.sub("", TOKEN.sub("", text).strip()).rstrip(". ") or ONE_GOES)
+            return
+        slots = [picture.slot for picture in self.swappers]
+        pieces = re.split(r"@(\d)\b", text)
+        named = {int(slot): words for slot, words in zip(pieces[1::2], pieces[2::2]) if int(slot) in slots}
+        if not named:
+            named = dict(zip(slots, lines)) if len(lines) == len(slots) else {}
+        for index, slot in enumerate(slots):
+            words = SAYS_WHO.sub("", named.get(slot, "").strip(" .,;")).strip(" .,;")
+            words = re.sub(r"\s+and$", "", words)
+            if words:
+                self.whos[index] = lower_first(words)
+
     @property
     def main(self):
         """Who a line is about when it names nobody."""
@@ -188,6 +247,8 @@ class Cast:
         if not self.pictures and not self.video:
             return NOTHING_IN
         if self.video in NEEDS_A_PERSON and not self.people:
+            if self.of(OUTFIT, THING):
+                return DRESS_THE_VIDEO_PERSON
             return NEEDS_A_PERSON[self.video]
         typed = sorted({int(number) for number in TOKEN.findall(text)} - {0})
         slots = {picture.slot: picture for picture in self.pictures}
@@ -222,7 +283,8 @@ class Cast:
         ends = self.ends
         lines = []
         if self.video == SWAP:
-            lines.append(SWAP_IS.format(replaced=self.replaced, who=self.who))
+            lines.append(SWAP_VIDEO)
+            lines += [SWAP_IS.format(replaced=gone, who=who) for gone, who, _ in self.swaps]
         worn = set(wearers.values())
         for picture in self.pictures:
             if picture.job == END:
@@ -276,7 +338,8 @@ class Cast:
 
         lines = []
         if self.video == SWAP:
-            lines.append(SWAP_KEPT.format(replaced=self.replaced, who=self.who, main=self.main))
+            lines.append(SWAP_VIDEO_KEPT.format(people="people" if len(self.swaps) > 1 else "person"))
+            lines += [SWAP_KEPT.format(replaced=gone, who=who, main=new) for gone, who, new in self.swaps]
         for picture in self.pictures:
             if picture.job == END:
                 continue

nodes/_h3_prompt_helpers.py

@@ -13,7 +13,7 @@ import re
 import numpy as np
 
 from ._h3_cast_helpers import KEEP, NEEDS_NO_WORDS, REVOICE, SWAP, Cast, number_pictures
-from ._h3_script_helpers import ACTION_GOES_ON, Moment, add_moments, add_sound, lower_first, read_script, script_parts, speakers
+from ._h3_script_helpers import ACTION_GOES_ON, Moment, add_moments, add_sound, read_script, script_parts, speakers
 
 SECTIONS = (
     "subject_definitions",
@@ -184,7 +184,7 @@ def write_prompt(script, seconds=15.0, camera="", look="", sound="", language="E
         if not scene.moments and video not in NEEDS_NO_WORDS:
             raise ValueError(NO_SCRIPT)
         if video == SWAP and scene.moments:  # Typed words in a swap say which person goes, for a clip with two.
-            cast.who = lower_first(" ".join(moment.text for moment in scene.moments).rstrip(". "))
+            cast.name_who_goes([moment.text for moment in scene.moments])
             scene.moments = []
         if not scene.moments:
             scene.moments.append(Moment(ACTION_GOES_ON if video == KEEP else ""))

nodes/_h3_script_helpers.py

@@ -25,7 +25,10 @@ from ._h3_cast_helpers import (
     SOUND_GOES_ON_KEPT,
     THING,
     TOKEN,
+    ONE_GOES,
+    SPOTS,
     Cast,
+    lower_first,
     number_pictures,
     picture_label,
     subject_label,
@@ -124,7 +127,6 @@ VIDEO_LOOKS = {
     KEEP: "The target video keeps the look, colours and camera of <Video 1>.",
     REVOICE: "The target video keeps the look, light and camera of <Video 1>.",
 }
-SPOTS = {2: ("left", "right"), 3: ("left", "centre", "right"), 4: ("far left", "left", "right", "far right")}
 NAMES_A_SPOT = re.compile(r"\b(?:left|right|behind|in\s+front|foreground|background)\b", re.I)
 OPENS_ON_THEM = re.compile(r"^(?:She|He)\b")
 FOLLOWS_A_COMMA = re.compile(r"^(?:A|An|The|She|He|They|It|Her|His|Their|Its|This|That|These|Those|There|Someone|Everyone|Nobody)\b")
@@ -539,9 +541,6 @@ def stamp(seconds):
     return f"{int(minutes):02d}:{rest:06.3f}"
 
 
-def lower_first(text):
-    first = text.split(" ", 1)[0]
-    return text if text[:1] == "<" or (len(first) > 1 and first.isupper()) else text[:1].lower() + text[1:]
 
 
 def after_comma(text):
@@ -669,7 +668,8 @@ def before_the_action(number, text, cast, wearers):
         words = OPENS_ON if cast.ends else OPENING
         notes.append(words.format(subject=subject_label(opens.subject), picture=picture_label(opens.picture)))
     if cast.video == SWAP:
-        notes += [TAKES_OVER.format(main=cast.main, who=cast.who), ONE_PERSON if cast.who == "the original person" else OTHERS_STAY]
+        notes += [TAKES_OVER.format(main=new, who=who) for _, who, new in cast.swaps]
+        notes.append(ONE_PERSON if cast.whos == [ONE_GOES] else OTHERS_STAY)
     if number == 1 and cast.video == KEEP:
         notes += [CARRIES_ON]
     if number == 1 and cast.video == REVOICE:
@@ -678,7 +678,7 @@ def before_the_action(number, text, cast, wearers):
     if places and not any(place in text for place in places):
         notes.append(IN_THE_PLACE.format(place=places[0]))
     people = people_in(text, cast)
-    if len(people) > 1 and not NAMES_A_SPOT.search(text):
+    if len(people) > 1 and not NAMES_A_SPOT.search(text) and cast.video != SWAP:  # A swap's people stand where the video has them.
         spots = [f"{person} is on the {spot}" for person, spot in zip(people, SPOTS[len(people)])]
         notes.append(sentence(", ".join(spots[:-1]) + f" and {spots[-1]} of the frame"))
     for slot, wearer in wearers.items():
@@ -786,7 +786,8 @@ def task_of(cast):
 def summary_of(script, cast, new_places, soundtrack):
     task = task_of(cast)
     if cast.video == SWAP:
-        return f"[{task}] Replace {cast.replaced} with {cast.main} while preserving the original action, place, camera and look of <Video 1>."
+        swaps = " and ".join(f"{gone} with {new}" for gone, _, new in cast.swaps)
+        return f"[{task}] Replace {swaps} while preserving the original action, place, camera and look of <Video 1>."
     if cast.video == REVOICE:
         return (
             f"[{task}] The target video is an edited version of <Video 1>. The picture, the timing and the lip movement "

scripts/workflow_contract.py

@@ -63,6 +63,58 @@ def named_copy_problems(node: dict) -> list[str]:
     return [f"node {node['id']} has stale named widget values: {', '.join(stale)}"] if stale else []
 
 
+def saved_widget(node: dict, name: str):
+    """A widget's saved value by name, or KeyError when the node keeps no named copy of it."""
+    values = node.get("widgets_values")
+    if isinstance(values, dict) and name in values:
+        return values[name]
+    named = node.get("widgets_values_named")
+    if isinstance(named, dict) and name in named:
+        return named[name]
+    raise KeyError(name)
+
+
+def switch_row_problems(card: dict, by_id: dict) -> list[str]:
+    """A Workflow Switches row opens striped "changed", with neither off nor
+    on picked, when the values it sets are saved half one way and half the
+    other, or when its nodes run one way and its values say the other. Every
+    row of an example has to open plainly off or on."""
+    settings = (card.get("properties") or {}).get("ausboss_workflow_switches") or {}
+    problems = []
+    for row in settings.get("switches") or []:
+        title = row.get("title", row.get("id"))
+        sides = set()
+        on, off = row.get("on") or {}, row.get("off") or {}
+        for node_id, widgets in on.items():
+            node = by_id.get(int(node_id)) if str(node_id).isdigit() else None
+            if node is None or not isinstance(widgets, dict):
+                continue
+            for name, on_value in widgets.items():
+                off_value = (off.get(node_id) or {}).get(name)
+                lora_rows = isinstance(on_value, str) and on_value.startswith("[")
+                if isinstance(on_value, (dict, list)) or lora_rows or on_value == off_value:
+                    continue  # a LoRA stack is matched row by row, and a shared value says nothing
+                try:
+                    saved = saved_widget(node, name)
+                except KeyError:
+                    continue
+                if saved == on_value:
+                    sides.add("on")
+                elif saved == off_value:
+                    sides.add("off")
+                else:
+                    sides.add(f"neither ({node_id} {name} is {saved!r}, not {on_value!r} or {off_value!r})")
+        members = [by_id[i] for i in row.get("nodes") or [] if i in by_id]
+        modes = {"on" if m.get("mode", 0) == 0 else "off" for m in members}
+        if len(modes) > 1:
+            problems.append(f"switch card {card['id']} row '{title}': some of its nodes are on and some off")
+        odd = sorted(side for side in sides if side not in ("on", "off"))
+        if odd or len(sides) > 1 or (sides and modes and sides != modes):
+            detail = "; ".join(odd) if odd else f"values say {'/'.join(sorted(sides))}, nodes say {'/'.join(sorted(modes)) or '-'}"
+            problems.append(f"switch card {card['id']} row '{title}' opens as changed: {detail}")
+    return problems
+
+
 def workflow_problems(data: dict) -> list[str]:
     problems = []
     nodes = data.get("nodes", [])
@@ -112,6 +164,8 @@ def workflow_problems(data: dict) -> list[str]:
         problems.append(f"saved zoom {scale} is below {MIN_ZOOM}; widget text would not draw")
     for n in nodes:
         problems.extend(named_copy_problems(n))
+        if n["type"] == "AUSBOSS_NODES_WorkflowSwitches":
+            problems.extend(switch_row_problems(n, by_id))
         if n["type"] in CORE_LORA_LOADERS:
             problems.append(f"node {n['id']} loads a LoRA with core {n['type']}; examples use LoRA Loader 🆎")
         selected = selected_files(n)

tests/test_h3_prompt_writer.py

@@ -652,6 +652,12 @@ class VideoJobTests(unittest.TestCase):
         cast = Cast(number_pictures({1: "place"}, "copy its moves"), "copy its moves")
         self.assertIn("To copy the moves", cast.problem("She dances."))
 
+    def test_a_hat_with_no_person_picture_says_how_to_put_it_on_the_video_person(self):
+        for job in ("outfit", "thing"):
+            for video in ("swap the person", "copy its moves"):
+                cast = Cast(number_pictures({1: job}, video), video)
+                self.assertIn("load a still of them from the video as Picture 1", cast.problem("He wears @1."))
+
     def test_a_swap_with_nobody_to_put_in_says_so(self):
         with self.assertRaisesRegex(ValueError, "new person"):
             write_prompt("", jobs={}, video="swap the person")
@@ -676,6 +682,28 @@ class VideoJobTests(unittest.TestCase):
         self.assertIn("Everyone else stays", parts["detailed_description"])
         self.assertNotIn("Show one person", parts["detailed_description"])
 
+    def test_a_swap_with_two_people_puts_each_in_one_place(self):
+        parts = parts_of("", jobs={1: "person", 2: "person"}, video="swap the person")
+        self.assertIn("Replace <Subject 1> with <Subject 3> and <Subject 2> with <Subject 4>", parts["summary"])
+        self.assertIn("<Subject 1> is the person on the left in <Video 1>", parts["subject_definitions"])
+        self.assertIn("<Subject 2> is the person on the right in <Video 1>", parts["subject_definitions"])
+        self.assertIn("<Subject 4> is the person in <Picture 2>", parts["subject_definitions"])
+        self.assertIn("replace the person on the right with <Subject 4>", parts["retention_analysis"])
+        self.assertIn("change only the people", parts["retention_analysis"])
+        self.assertIn("<Subject 4> takes the place of the person on the right", parts["detailed_description"])
+        self.assertNotIn("Show one person", parts["detailed_description"])
+
+    def test_a_swap_with_two_people_reads_who_goes_from_the_script(self):
+        for script in ("@2 is the woman in red. @1 replaces the man in the hat", "The man in the hat.\nThe woman in red."):
+            parts = parts_of(script, jobs={1: "person", 2: "character sheet"}, video="swap the person")
+            self.assertIn("replace the man in the hat with <Subject 3>", parts["retention_analysis"])
+            self.assertIn("replace the woman in red with <Subject 4>", parts["retention_analysis"])
+            self.assertNotIn("@", join_sections(parts))
+
+    def test_one_swap_takes_the_at_out_of_who_goes(self):
+        parts = parts_of("@1 replaces the child in the seat", jobs={1: "person"}, video="swap the person")
+        self.assertIn("replace the child in the seat with <Subject 2>", parts["retention_analysis"])
+
     def test_keeping_going_needs_no_words_and_no_picture(self):
         parts = parts_of("", jobs={}, video="keep going")
         self.assertTrue(parts["summary"].startswith("[video continuation]"))

tests/test_workflow_contract.py

@@ -38,6 +38,24 @@ class WorkflowContractTests(unittest.TestCase):
         graph["nodes"][0]["widgets_values_named"] = {"video": "someone_else.mp4", "start_seconds": 0, "upload": ""}
         self.assertIn("stale named widget values: video", " ".join(workflow_problems(graph)))
 
+    def test_a_switch_row_must_open_plainly_off_or_on(self):
+        graph = good_graph()
+        graph["nodes"][0]["widgets_values"] = {"megapixels": 0.41}
+        card = {"id": 4, "type": "AUSBOSS_NODES_WorkflowSwitches", "pos": [0, 300], "size": [100, 40], "properties": {
+            "ausboss_workflow_switches": {"switches": [{"id": 9, "title": "Fast test", "nodes": [],
+                "on": {"1": {"megapixels": 0.26}}, "off": {"1": {"megapixels": 1}}}]}}}
+        graph["nodes"].append(card)
+        graph["last_node_id"] = 4
+        graph["groups"][0]["bounding"] = [-20, 40, 320, 320]
+        self.assertIn("row 'Fast test' opens as changed", " ".join(workflow_problems(graph)))
+        card["properties"]["ausboss_workflow_switches"]["switches"][0]["off"] = {"1": {"megapixels": 0.41}}
+        self.assertEqual(workflow_problems(graph), [])
+        graph["nodes"][1]["widgets_values"] = {"steps": 8}
+        graph["nodes"][1]["mode"] = 4
+        card["properties"]["ausboss_workflow_switches"]["switches"].append(
+            {"id": 10, "title": "Turbo", "nodes": [2], "on": {"2": {"steps": 8}}, "off": {"2": {"steps": 20}}})
+        self.assertIn("row 'Turbo' opens as changed: values say on, nodes say off", " ".join(workflow_problems(graph)))
+
     def test_workflow_id_and_zoom(self):
         graph = good_graph()
         graph["id"] = "3f2b8c1e-5d4a-4e6f-9a7b-1c2d3e4f5a6b"