fix #152 make all option optional

src/ffmpeg/nodes/base.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod, abstractproperty
+from functools import cached_property
 from typing import Any, Iterable, Mapping, Sequence
 
 from pydantic import BaseModel, model_validator
@@ -38,6 +39,13 @@ class HashableBaseModel(BaseModel):
     def __hash__(self) -> int:
         return hash(self.model_dump_json())
 
+    @cached_property
+    def hex(self) -> str:
+        return hex(abs(hash(self)))[2:]
+
+    def __repr__(self) -> str:
+        return f"{self.__class__.__name__}({self.hex})"
+
 
 class Stream(HashableBaseModel):
     node: Node

src/ffmpeg/nodes/nodes.py

@@ -20,6 +20,9 @@ class FilterNode(Node):
     input_typings: list[StreamType] | None = None
     output_typings: list[StreamType] | None = None
 
+    def __repr__(self) -> str:
+        return f"{self.__class__.__name__}:{self.name}({self.hex})"
+
     @property
     def incoming_streams(self) -> Sequence[Stream]:
         return self.inputs

src/ffmpeg/tests/__snapshots__/test_base/test_audio_video_pipeline.json

@@ -5,7 +5,7 @@
   "-i",
   "in2.mp4",
   "-filter_complex",
-  "[0:v]hflip[s1];[1:a]areverse[s2];[1:v]reverse[s0];[s0]hue=s=0[s3];[s1][0:a][s3][s4]concat=v=1:a=1[s5#0][s5#1];[s2]aphaser[s4];[s5#1]volume=volume=0.8[s6]",
+  "[0:v]hflip[s0];[1:v]reverse[s1];[1:a]areverse[s2];[s1]hue=s=0[s3];[s2]aphaser[s4];[s0][0:a][s3][s4]concat=v=1:a=1[s5#0][s5#1];[s5#1]volume=volume=0.8[s6]",
   "-map",
   "[s5#0]",
   "-map",

src/ffmpeg/tests/__snapshots__/test_base/test_mono_to_stereo_with_offsets_and_video.json

@@ -2,17 +2,17 @@
   "ffmpeg",
   "-y",
   "-i",
+  "audio-left.wav",
+  "-i",
   "audio-right.wav",
   "-i",
   "input-video.mp4",
-  "-i",
-  "audio-left.wav",
   "-filter_complex",
-  "[0]atrim=start=10[s0];[2]atrim=start=5[s1];[s0]asetpts=expr=PTS-STARTPTS[s3];[s1]asetpts=expr=PTS-STARTPTS[s2];[s2][s3]join=inputs=2:channel_layout=stereo[s4]",
+  "[0]atrim=start=5[s0];[1]atrim=start=10[s1];[s0]asetpts=expr=PTS-STARTPTS[s2];[s1]asetpts=expr=PTS-STARTPTS[s3];[s2][s3]join=inputs=2:channel_layout=stereo[s4]",
   "-map",
   "[s4]",
   "-map",
-  "[1:v]",
+  "[2:v]",
   "-shortest",
   "-vcodec",
   "copy",

src/ffmpeg/utils/compile.py

@@ -1,19 +1,33 @@
 from __future__ import annotations
 
-from typing import Iterable
+from typing import Iterable, TypeVar
 
 from ..nodes.base import Node, Stream, _DAGContext
 from ..nodes.nodes import FilterNode, GlobalNode, InputNode, OutputNode
 
+T = TypeVar("T")
 
-def _collect(node: Node) -> tuple[set[Node], set[Stream]]:
+
+def _remove_duplicates(seq: list[T]) -> list[T]:
+    seen = set()
+    output = []
+
+    for x in seq:
+        if x not in seen:
+            output.append(x)
+            seen.add(x)
+
+    return output
+
+
+def _collect(node: Node) -> tuple[list[Node], list[Stream]]:
     """Collect all nodes and streams that are upstreamed to the given node"""
-    nodes, streams = {node}, {*node.incoming_streams}
+    nodes, streams = [node], [*node.incoming_streams]
 
     for stream in node.incoming_streams:
         _nodes, _streams = _collect(stream.node)
-        nodes |= _nodes
-        streams |= _streams
+        nodes += _nodes
+        streams += _streams
 
     return nodes, streams
 
@@ -25,11 +39,11 @@ def _calculate_node_max_depth(node: Node, outgoing_streams: dict[Node, list[Stre
     return max(_calculate_node_max_depth(stream.node, outgoing_streams) for stream in node.incoming_streams) + 1
 
 
-def _node_max_depth(all_nodes: set[Node], outgoing_streams: dict[Node, list[Stream]]) -> dict[Node, int]:
+def _node_max_depth(all_nodes: list[Node], outgoing_streams: dict[Node, list[Stream]]) -> dict[Node, int]:
     return {node: _calculate_node_max_depth(node, outgoing_streams) for node in all_nodes}
 
 
-def _build_outgoing_streams(nodes: set[Node], streams: set[Stream]) -> dict[Node, list[Stream]]:
+def _build_outgoing_streams(nodes: list[Node], streams: list[Stream]) -> dict[Node, list[Stream]]:
     outgoing_streams: dict[Node, list[Stream]] = {}
 
     for node in nodes:
@@ -41,7 +55,7 @@ def _build_outgoing_streams(nodes: set[Node], streams: set[Stream]) -> dict[Node
     return outgoing_streams
 
 
-def _build_node_labels(nodes: set[Node], outgoing_streams: dict[Node, list[Stream]]) -> dict[Node, str]:
+def _build_node_labels(nodes: list[Node], outgoing_streams: dict[Node, list[Stream]]) -> dict[Node, str]:
     node_max_depth = _node_max_depth(nodes, outgoing_streams)
     input_node_index = 0
     filter_node_index = 0
@@ -73,11 +87,14 @@ class DAGContext(_DAGContext):
     def build(cls, node: Node) -> DAGContext:
         """create a DAG context based on the given node"""
         nodes, streams = _collect(node)
+        nodes = _remove_duplicates(nodes)
+        streams = _remove_duplicates(streams)
+
         outgoing_streams = _build_outgoing_streams(nodes, streams)
         node_labels = _build_node_labels(nodes, outgoing_streams)
 
         all_nodes = sorted(nodes, key=lambda node: node_labels[node])
-        all_streams = sorted(streams, key=lambda stream: node_labels[stream.node])
+        all_streams = sorted(streams, key=lambda stream: (node_labels[stream.node], stream.index))
 
         return cls(
             node=node,
@@ -119,11 +136,10 @@ def compile(node: Node) -> list[str]:
     vf_commands = []
     filter_nodes = [node for node in context.all_nodes if isinstance(node, FilterNode)]
 
-    for node in filter_nodes:
+    for node in sorted(filter_nodes, key=lambda node: context.node_labels[node]):
         vf_commands += ["".join(node.get_args(context))]
 
     if vf_commands:
-        vf_commands.sort()
         commands += ["-filter_complex", ";".join(vf_commands)]
 
     # compile the output nodes