SpriteDX - Stage 4 Pipeline JSON Integration (Part 2)

Today, I volunteered for a conference in SF in the morning. After standing up greeting speakers from 6 to 11, I was rather burned out and didn’t really want to attend any talks. I guess it is little bit of different vibe when you are the one serving. So, I drove back early and working on SpriteDX.

We are 97 days til Alpha, and yesterday we started integrating Stage 4 and hit a blocker.

The blocker is that now I have three clips—clip_greet.webp, clip_idle.webp and clip_run.webp—and we need to run the same pipeline on each of these intermediate outputs.

Current pipeline JSON schema does not include such semantics so we will have to invent one.

So, we can’t simply run the pipeline multiple times if there are multiple input files because in some situations, we would want the whole input to go into the same pipeline instance. So, we will need to explicitly state in the JSON how we want to “map“ the iteration.

First Draft

{
  "id": "postProcessing",
  "name": "Post-Processing",
  "type": "runner",
  "runner": "ComfyUI",
  "workflowRef": "pipelines/character/v1/workflows/sprite-dx-v1-stage-4.json",
  "statusMessage": "Post Processing…",
  "inputs": {
    "clips": {
      "type": "array",
      "items": { "type": "image" },
      "source": "..editing.clips"
    },
  },
  "mapOn": { 
    "type": "image",
    "source": "clips", 
    "as": "clip", 
    "mapToFile": "clip.webp" 
  },
  "outputs": {}
}

In here, we are defining “array of images” pulled from previous stage, editing.

Then we set "mapOn": { "input": "clips", … } to make the stage runner iterate over the clips input array. There are some issues with this format.

First of all, this only allows for iterating on single input. Just mapping on single input will not work because we need to be able to produce different output files.

If all iterations produced the same output file name, they will be overwritten.

Likely though, we can support zipping though multiple input values by using computed properties. Let’s try this out.

Second Draft

{
  "id": "postProcessing",
  "name": "Post-Processing",
  "type": "runner",
  "runner": "ComfyUI",
  "workflowRef": "pipelines/character/v1/workflows/sprite-dx-v1-stage-4.json",
  "statusMessage": "Post Processing…",
  "inputs": {
    "clip": {
      "type": "image",
      "source": "..editing.clips.*"
    },
    "clipId": {
      "type": "string",
      "computed": {
        "sandbox": "quickjs",
        "with": { "clip": "clip" },
        "script": "clip.split('/').pop().replace(/\\.webp$/, '')"
      }
    },
    …other params…
  },
  "outputs": {}
}

In this version, we have folded mapOn semantic into the stage input definition by using .* source mapping. This .* will basically indicate that we should iterate through the set of clips and get its index.

The iterative behavior is not explicit here but it should get the job done. It allows for computed properties that depend on the iteration target (i.e. clip), and also allows for iterating through 2 or more input arrays.

Let’s see what happens if we make this more explicit.

Third Draft

{
  "id": "postProcessing",
  "name": "Post-Processing",
  "type": "runner",
  "runner": "ComfyUI",
  "workflowRef": "pipelines/character/v1/workflows/sprite-dx-v1-stage-4.json",
  "statusMessage": "Post Processing…",
  "inputs": {
    "clip": {
      "type": "image",
      "forEach": "..editing.clips"
    },
    "clipId": {
      "type": "string",
      "computed": {
        "sandbox": "quickjs",
        "with": { "clip": "clip" },
        "script": "clip.split('/').pop().replace(/\\.webp$/, '')"
      }
    },
    …other params…
  },
  "outputs": {}
}

In this version, we are renaming source to forEach. This makes it more explicit and perhaps little easier to make sense when reading through the JSON with naked eye.

What would this mean when we have to do this opposite. What we are doing above is basically clips.map(clip => …). That means we also need to support something like clips.reduce(…) where we combine branched out results back into a single one.

{
  "id": "postProcessing",
  "name": "Post-Processing",
  "type": "runner",
  "runner": "ComfyUI",
  "workflowRef": "pipelines/character/v1/workflows/sprite-dx-v1-stage-4.json",
  "statusMessage": "Post Processing…",
  "inputs": {
    "clip": {
      "type": "image",
      "forEach": "..editing.clips"
    },
    "clipId": {
      "type": "string",
      "computed": {
        "sandbox": "quickjs",
        "with": { "clip": "clip" },
        "script": "clip.split('/').pop().replace(/\\.webp$/, '')"
      }
    },
    …other params…
  },
  "outputs": {
    "processedClips": {
      "type": "array",
      "items": {
        "type": "image"
      },
      "mapToFile": "processed_*.webp",
      "saveAs": {
        "match": "^(.*)_(\\d+)_\\.webp$",
        "replace": "$1.webp",
        "conflict": "skip"
      }
    }
  }
}

Because we are collecting the outputs using a glob pattern, at the end of the stage, we are essentially back to single branch.

This probably just works because we currently only support image outputs and won't work if we expect the stages to produce overlapping datapoints.

Fourth Draft

So, alternatively, we can let the outputs to be per run of the pipeline.

{
  "id": "postProcessing",
  "name": "Post-Processing",
  "type": "runner",
  "runner": "ComfyUI",
  "workflowRef": "pipelines/character/v1/workflows/sprite-dx-v1-stage-4.json",
  "statusMessage": "Post Processing…",
  "inputs": {
    "clip": {
      "type": "image",
      "forEach": "..editing.clips"
    },
    "clipId": {
      "type": "string",
      "computed": {
        "sandbox": "quickjs",
        "with": { "clip": "clip" },
        "script": "clip.split('/').pop().replace(/\\.webp$/, '')"
      }
    },
    …other params…
  },
  "outputs": {
    "processedClip": {
      "type": "image",
      "mapToFile": "processed_*.webp",
      "saveAs": {
        "match": "^(.*)_(\\d+)_\\.webp$",
        "replace": "$1.webp",
        "conflict": "skip"
      }
    }
  }
}

Then, when subsequent stages need to refer to the outputs, they will just have to do something like:

{
  "id": "someSubseqeuntStage"
  "inputs": {
    "processedClips": {
      "type": "array",
      "items": { "type": "image" },
      "collectAll": "..postProcessing.processedClip"
    },
  }
  … 
}

where collectAll key is used when needing to collect branched out results.

So, to summarize what we have so far:

• Use forEach when branching.

• Use collectAll when merging.

This is rather complicated because once you have multiple level of branching, collectAll will need to be collecting from multiple branching axis without even knowing which axis we are branching from.

So, I think we need the concept of branching out and merging to be built into stage-level semantics rather than input-level semantics.

Fifth Draft

{
  "id": "postProcessing",
  "type": "forEach",
  "forEach": { "source": "clips", "as": "clip" },
  "inputs": {
    "clips": { 
      "type": "array", "items": { "type": "image" } "source": "..editing.clips"
    }
  },
  "stages": [{
    "type": "runner",
    "inputs": {
      "clip": { "type": "image", "source": "..clip" }
    },
    "outputs": { … } // branched out results.
  }],
  "outputs": { … } // merged results.
}

In this version, we are creating a wrapper stage that only runs “foreach“ semantics. Then hands it off to the child stage to perform the individual iterations.

Here the benefit is that we are separating the concern of the iteration with the runner logic so it is conceptually much more segregated and easier to reason about.

It also guarantees that the top level is always the single thread and only the nested child is ever multi-threaded.

In similar manner, we can implement logic branches and this seems to be the most extendable and flexible format.

---

Let’s close out today with above as the plan of record, but tomorrow, let’s validate that the format indeed works for our needs.

—Sprited Dev 🌱