FlashDreams Post-Processing

Use this skill when adding a video post-processor or changing the runner post-processing stream. The reference implementation is integrations_v2/flashvsr/impl/postprocess.py.

Mental Model

A post-processor is usually three classes, not one class inheriting everything:

• VideoPostProcessorConfig: serializable config and CLI surface. It sets

target to the processor factory and declares fields, outputspec(), requiresallranks(), and validate_execution().

• VideoPostProcessor: lightweight factory created from config. Its job is

start(spec) -> VideoPostProcessorSession.

• VideoPostProcessorSession: mutable per-stream runtime. It owns buffers,

caches, lazy model instances, counters, and process() / flush().

Keep stream state in the session. Do not store per-rollout mutable state on the config or processor factory.

Implementation Steps

1. Pick a home:

- Generic reusable post-processing belongs under flashdreams/flashdreams/infra/postprocess/. - Model-specific processors belong in their integration, for example integrations/<name>/<pkg>/postprocess.py.

1. Define a config subclass:

```python @dataclass(kwonly=True) class MyPostProcessorConfig(VideoPostProcessorConfig): target: type["MyPostProcessor"] = field( default_factory=lambda: MyPostProcessor )

scale: int = 2

def outputspec(self, inputspec: VideoSpec) -> VideoSpec: return VideoSpec( height=inputspec.height self.scale, width=inputspec.width self.scale, fps=inputspec.fps, channels=inputspec.channels, ) ```

Override: - outputspec() when spatial size, channels, or timing changes. - requiresallranks() when the processor must run on nonzero ranks under torchrun. - validateexecution() to reject unsupported distributed or shape modes early.

1. Define the processor factory:

``python class MyPostProcessor(VideoPostProcessor[MyPostProcessorConfig]): def start(self, spec: VideoSpec) -> VideoPostProcessorSession: return _MyPostProcessorSession(self.config, spec) ``

1. Define the session:

```python class MyPostProcessorSession(VideoPostProcessorSession): def init(self, config: MyPostProcessorConfig, spec: VideoSpec) -> None: self.config = config self.spec = spec self.buffer: Tensor | None = None

def process(self, chunk: VideoChunk) -> list[VideoChunk]: ...

def flush(self) -> list[VideoChunk]: ... ```

process() is synchronous but may return []: that means it consumed the input chunk and is buffering frames until a later chunk or flush() can complete an output window.

1. Handle layouts at the boundary:

- Accept VideoChunk.tensor in chunk.layout. - Use to_bvtchw() only as a generic boundary helper. - Convert once into the processor's native layout, make it contiguous if the model kernels require that, and keep internal buffers in that native layout. - Document any forced .contiguous() because it can copy.

1. Return VideoChunks:

- Preserve [-1, 1] value range unless the API is intentionally changed. - Set the correct layout. - Carry metadata only if it helps downstream processors or provenance.

1. Register presets when users should select it from CLI:

``toml [project.entry-points."flashdreams.postprocesspresets"] "my-postprocessor-v1" = "mypkg.postprocess:POSTPROCESSPRESETMY_V1" ``

The exported object must be a VideoPostProcessorConfig, for example:

``python POSTPROCESSPRESETMY_V1 = MyPostProcessorConfig(...) ``

Users select it with --postprocess.preset my-postprocessor-v1.

Runner Interaction

Runners create a VideoPostprocessStream through createrunnerpostprocess_stream(). The stream:

• creates one chain session for whole-stream processing, or one session per

view when postprocessperview=True;

• calls session.process(VideoChunk(...)) for each AR output;
• turns [] into a zero-frame tensor so process() remains tensor-only;
• skips collecting zero-time tensors in appendif_nonempty();
• calls flush() once at end-of-stream and appends any tail output.

Use postprocessoutputlayout to describe the runner's decoded output layout. Use postprocessperview=True for bvtchw outputs when each camera/view needs an independent processor session.

Tests

Add CPU-safe tests unless the behavior genuinely requires a GPU:

• Config/preset discovery: flashdreams/tests/testpostprocesspresets.py.
• Stream contract and buffering: flashdreams/tests/testpostprocessstream.py.
• Processor-specific CPU fakes: integrations/<name>/tests/test_postprocess.py.
• Runner distributed skip/all-rank behavior:

flashdreams/tests/testrunnerpostprocess.py.

Every pytest test must use exactly one marker: cicpu, cigpu, or manual. Prefer fake processor builders for CPU tests instead of loading checkpoints.

Useful focused validation:

uv run pytest flashdreams/tests/test_runner_postprocess.py \
  flashdreams/tests/test_postprocess_stream.py \
  flashdreams/tests/test_postprocess_presets.py \
  integrations/<name>/tests/test_postprocess.py