Massive refactor to break nodes into streams+nodes

ffmpeg/__init__.py

@@ -3,4 +3,5 @@ from . import _filters, _ffmpeg, _run
 from ._filters import *
 from ._ffmpeg import *
 from ._run import *
-__all__ = _filters.__all__ + _ffmpeg.__all__ + _run.__all__
+from ._view import *
+__all__ = _filters.__all__ + _ffmpeg.__all__ + _run.__all__ + _view.__all__

ffmpeg/_ffmpeg.py

@@ -1,10 +1,11 @@
 from __future__ import unicode_literals
 from .nodes import (
-    FilterNode,
+    filter_operator,
     GlobalNode,
     InputNode,
-    operator,
+    MergeOutputsNode,
     OutputNode,
+    output_operator,
 )
 
 
@@ -19,27 +20,27 @@ def input(filename, **kwargs):
         if 'format' in kwargs:
             raise ValueError("Can't specify both `format` and `f` kwargs")
         kwargs['format'] = fmt
-    return InputNode(input.__name__, **kwargs)
+    return InputNode(input.__name__, kwargs=kwargs).stream()
 
 
-@operator(node_classes={OutputNode, GlobalNode})
-def overwrite_output(parent_node):
+@output_operator()
+def overwrite_output(stream):
     """Overwrite output files without asking (ffmpeg ``-y`` option)
 
     Official documentation: `Main options <https://ffmpeg.org/ffmpeg.html#Main-options>`__
     """
-    return GlobalNode(parent_node, overwrite_output.__name__)
+    return GlobalNode(stream, overwrite_output.__name__).stream()
 
 
-@operator(node_classes={OutputNode})
-def merge_outputs(*parent_nodes):
+@output_operator()
+def merge_outputs(*streams):
     """Include all given outputs in one ffmpeg command line
     """
-    return OutputNode(parent_nodes, merge_outputs.__name__)
+    return MergeOutputsNode(streams, merge_outputs.__name__).stream()
 
 
-@operator(node_classes={InputNode, FilterNode})
-def output(parent_node, filename, **kwargs):
+@filter_operator()
+def output(stream, filename, **kwargs):
     """Output file URL
 
     Official documentation: `Synopsis <https://ffmpeg.org/ffmpeg.html#Synopsis>`__
@@ -50,7 +51,7 @@ def output(parent_node, filename, **kwargs):
         if 'format' in kwargs:
             raise ValueError("Can't specify both `format` and `f` kwargs")
         kwargs['format'] = fmt
-    return OutputNode([parent_node], output.__name__, **kwargs)
+    return OutputNode(stream, output.__name__, kwargs=kwargs).stream()
 
 
 

ffmpeg/_filters.py

@@ -1,64 +1,55 @@
 from __future__ import unicode_literals
 from .nodes import (
     FilterNode,
-    operator,
+    filter_operator,
 )
 
 
-@operator()
-def filter_(parent_node, filter_name, *args, **kwargs):
-    """Apply custom single-source filter.
+@filter_operator()
+def filter_multi_output(stream_spec, filter_name, *args, **kwargs):
+    """Apply custom filter with one or more outputs.
+
+    This is the same as ``filter_`` except that the filter can produce more than one output.
+
+    To reference an output stream, use either the ``.stream`` operator or bracket shorthand:
+
+    Example:
+
+        ```
+        split = ffmpeg.input('in.mp4').filter_multi_output('split')
+        split0 = split.stream(0)
+        split1 = split[1]
+        ffmpeg.concat(split0, split1).output('out.mp4').run()
+        ```
+    """
+    return FilterNode(stream_spec, filter_name, args=args, kwargs=kwargs, max_inputs=None)
+
+
+@filter_operator()
+def filter_(stream_spec, filter_name, *args, **kwargs):
+    """Apply custom filter.
 
     ``filter_`` is normally used by higher-level filter functions such as ``hflip``, but if a filter implementation
     is missing from ``fmpeg-python``, you can call ``filter_`` directly to have ``fmpeg-python`` pass the filter name
     and arguments to ffmpeg verbatim.
 
     Args:
-        parent_node: Source stream to apply filter to.
+        stream_spec: a Stream, list of Streams, or label-to-Stream dictionary mapping
         filter_name: ffmpeg filter name, e.g. `colorchannelmixer`
         *args: list of args to pass to ffmpeg verbatim
         **kwargs: list of keyword-args to pass to ffmpeg verbatim
 
-    This function is used internally by all of the other single-source filters (e.g. ``hflip``, ``crop``, etc.).
-    For custom multi-source filters, see ``filter_multi`` instead.
-
     The function name is suffixed with ``_`` in order avoid confusion with the standard python ``filter`` function.
 
     Example:
 
         ``ffmpeg.input('in.mp4').filter_('hflip').output('out.mp4').run()``
     """
-    return FilterNode([parent_node], filter_name, *args, **kwargs)
+    return filter_multi_output(stream_spec, filter_name, *args, **kwargs).stream()
 
 
-def filter_multi(parent_nodes, filter_name, *args, **kwargs):
-    """Apply custom multi-source filter.
-
-    This is nearly identical to the ``filter`` function except that it allows filters to be applied to multiple
-    streams.  It's normally used by higher-level filter functions such as ``concat``, but if a filter implementation
-    is missing from ``fmpeg-python``, you can call ``filter_multi`` directly.
-
-    Note that because it applies to multiple streams, it can't be used as an operator, unlike the ``filter`` function
-    (e.g. ``ffmpeg.input('in.mp4').filter_('hflip')``)
-
-    Args:
-        parent_nodes: List of source streams to apply filter to.
-        filter_name: ffmpeg filter name, e.g. `concat`
-        *args: list of args to pass to ffmpeg verbatim
-        **kwargs: list of keyword-args to pass to ffmpeg verbatim
-
-    For custom single-source filters, see ``filter_multi`` instead.
-
-    Example:
-
-        ``ffmpeg.filter_multi(ffmpeg.input('in1.mp4'), ffmpeg.input('in2.mp4'), 'concat', n=2).output('out.mp4').run()``
-    """
-    return FilterNode(parent_nodes, filter_name, *args, **kwargs)
-
-
-
-@operator()
-def setpts(parent_node, expr):
+@filter_operator()
+def setpts(stream, expr):
     """Change the PTS (presentation timestamp) of the input frames.
 
     Args:
@@ -66,11 +57,11 @@ def setpts(parent_node, expr):
 
     Official documentation: `setpts, asetpts <https://ffmpeg.org/ffmpeg-filters.html#setpts_002c-asetpts>`__
     """
-    return filter_(parent_node, setpts.__name__, expr)
+    return FilterNode(stream, setpts.__name__, args=[expr]).stream()
 
 
-@operator()
-def trim(parent_node, **kwargs):
+@filter_operator()
+def trim(stream, **kwargs):
     """Trim the input so that the output contains one continuous subpart of the input.
 
     Args:
@@ -88,10 +79,10 @@ def trim(parent_node, **kwargs):
 
     Official documentation: `trim <https://ffmpeg.org/ffmpeg-filters.html#trim>`__
     """
-    return filter_(parent_node, trim.__name__, **kwargs)
+    return FilterNode(stream, trim.__name__, kwargs=kwargs).stream()
 
 
-@operator()
+@filter_operator()
 def overlay(main_parent_node, overlay_parent_node, eof_action='repeat', **kwargs):
     """Overlay one video on top of another.
 
@@ -136,29 +127,29 @@ def overlay(main_parent_node, overlay_parent_node, eof_action='repeat', **kwargs
     Official documentation: `overlay <https://ffmpeg.org/ffmpeg-filters.html#overlay-1>`__
     """
     kwargs['eof_action'] = eof_action
-    return filter_multi([main_parent_node, overlay_parent_node], overlay.__name__, **kwargs)
+    return FilterNode([main_parent_node, overlay_parent_node], overlay.__name__, kwargs=kwargs, max_inputs=2).stream()
 
 
-@operator()
-def hflip(parent_node):
+@filter_operator()
+def hflip(stream):
     """Flip the input video horizontally.
 
     Official documentation: `hflip <https://ffmpeg.org/ffmpeg-filters.html#hflip>`__
     """
-    return filter_(parent_node, hflip.__name__)
+    return FilterNode(stream, hflip.__name__).stream()
 
 
-@operator()
-def vflip(parent_node):
+@filter_operator()
+def vflip(stream):
     """Flip the input video vertically.
 
     Official documentation: `vflip <https://ffmpeg.org/ffmpeg-filters.html#vflip>`__
     """
-    return filter_(parent_node, vflip.__name__)
+    return FilterNode(stream, vflip.__name__).stream()
 
 
-@operator()
-def drawbox(parent_node, x, y, width, height, color, thickness=None, **kwargs):
+@filter_operator()
+def drawbox(stream, x, y, width, height, color, thickness=None, **kwargs):
     """Draw a colored box on the input image.
 
     Args:
@@ -179,11 +170,11 @@ def drawbox(parent_node, x, y, width, height, color, thickness=None, **kwargs):
     """
     if thickness:
         kwargs['t'] = thickness
-    return filter_(parent_node, drawbox.__name__, x, y, width, height, color, **kwargs)
+    return FilterNode(stream, drawbox.__name__, args=[x, y, width, height, color], kwargs=kwargs).stream()
 
 
-@operator()
-def concat(*parent_nodes, **kwargs):
+@filter_operator()
+def concat(*streams, **kwargs):
     """Concatenate audio and video streams, joining them together one after the other.
 
     The filter works on segments of synchronized video and audio streams. All segments must have the same number of
@@ -208,12 +199,12 @@ def concat(*parent_nodes, **kwargs):
 
     Official documentation: `concat <https://ffmpeg.org/ffmpeg-filters.html#concat>`__
     """
-    kwargs['n'] = len(parent_nodes)
-    return filter_multi(parent_nodes, concat.__name__, **kwargs)
+    kwargs['n'] = len(streams)
+    return FilterNode(streams, concat.__name__, kwargs=kwargs, max_inputs=None).stream()
 
 
-@operator()
-def zoompan(parent_node, **kwargs):
+@filter_operator()
+def zoompan(stream, **kwargs):
     """Apply Zoom & Pan effect.
 
     Args:
@@ -228,11 +219,11 @@ def zoompan(parent_node, **kwargs):
 
     Official documentation: `zoompan <https://ffmpeg.org/ffmpeg-filters.html#zoompan>`__
     """
-    return filter_(parent_node, zoompan.__name__, **kwargs)
+    return FilterNode(stream, zoompan.__name__, kwargs=kwargs).stream()
 
 
-@operator()
-def hue(parent_node, **kwargs):
+@filter_operator()
+def hue(stream, **kwargs):
     """Modify the hue and/or the saturation of the input.
 
     Args:
@@ -243,16 +234,16 @@ def hue(parent_node, **kwargs):
 
     Official documentation: `hue <https://ffmpeg.org/ffmpeg-filters.html#hue>`__
     """
-    return filter_(parent_node, hue.__name__, **kwargs)
+    return FilterNode(stream, hue.__name__, kwargs=kwargs).stream()
 
 
-@operator()
-def colorchannelmixer(parent_node, *args, **kwargs):
+@filter_operator()
+def colorchannelmixer(stream, *args, **kwargs):
     """Adjust video input frames by re-mixing color channels.
 
     Official documentation: `colorchannelmixer <https://ffmpeg.org/ffmpeg-filters.html#colorchannelmixer>`__
     """
-    return filter_(parent_node, colorchannelmixer.__name__, **kwargs)
+    return FilterNode(stream, colorchannelmixer.__name__, kwargs=kwargs).stream()
 
 
 __all__ = [
@@ -260,7 +251,6 @@ __all__ = [
     'concat',
     'drawbox',
     'filter_',
-    'filter_multi',
     'hflip',
     'hue',
     'overlay',

ffmpeg/_run.py

@@ -4,22 +4,23 @@ from .dag import topo_sort
 from functools import reduce
 from past.builtins import basestring
 import copy
-import operator as _operator
+import operator
 import subprocess as _subprocess
 
 from ._ffmpeg import (
     input,
-    merge_outputs,
     output,
     overwrite_output,
 )
 from .nodes import (
     GlobalNode,
     InputNode,   
-    operator,
     OutputNode,
+    output_operator,
+    Stream,
 )
 
+
 def _get_stream_name(name):
     return '[{}]'.format(name)
 
@@ -73,32 +74,31 @@ def _get_global_args(node):
 
 
 def _get_output_args(node, stream_name_map):
+    if node.name != output.__name__:
+        raise ValueError('Unsupported output node: {}'.format(node))
     args = []
-    if node.name != merge_outputs.__name__:
-        assert len(node.incoming_edges) == 1
-        stream_name = stream_name_map[node.incoming_edges[0].upstream_node]
-        if stream_name != '[0]':
-            args += ['-map', stream_name]
-        if node.name == output.__name__:
-            kwargs = copy.copy(node.kwargs)
-            filename = kwargs.pop('filename')
-            fmt = kwargs.pop('format', None)
-            if fmt:
-                args += ['-f', fmt]
-            args += _convert_kwargs_to_cmd_line_args(kwargs)
-            args += [filename]
-        else:
-            raise ValueError('Unsupported output node: {}'.format(node))
+    assert len(node.incoming_edges) == 1
+    stream_name = stream_name_map[node.incoming_edges[0].upstream_node]
+    if stream_name != '[0]':
+        args += ['-map', stream_name]
+    kwargs = copy.copy(node.kwargs)
+    filename = kwargs.pop('filename')
+    fmt = kwargs.pop('format', None)
+    if fmt:
+        args += ['-f', fmt]
+    args += _convert_kwargs_to_cmd_line_args(kwargs)
+    args += [filename]
     return args
 
 
-@operator(node_classes={OutputNode, GlobalNode})
-def get_args(node):
+@output_operator()
+def get_args(stream):
     """Get command-line arguments for ffmpeg."""
+    if not isinstance(stream, Stream):
+        raise TypeError('Expected Stream; got {}'.format(type(stream)))
     args = []
     # TODO: group nodes together, e.g. `-i somefile -r somerate`.
-    sorted_nodes, outgoing_edge_maps = topo_sort([node])
-    del(node)
+    sorted_nodes, outgoing_edge_maps = topo_sort([stream.node])
     input_nodes = [node for node in sorted_nodes if isinstance(node, InputNode)]
     output_nodes = [node for node in sorted_nodes if isinstance(node, OutputNode) and not
         isinstance(node, GlobalNode)]
@@ -106,15 +106,15 @@ def get_args(node):
     filter_nodes = [node for node in sorted_nodes if node not in (input_nodes + output_nodes + global_nodes)]
     stream_name_map = {node: _get_stream_name(i) for i, node in enumerate(input_nodes)}
     filter_arg = _get_filter_arg(filter_nodes, stream_name_map)
-    args += reduce(_operator.add, [_get_input_args(node) for node in input_nodes])
+    args += reduce(operator.add, [_get_input_args(node) for node in input_nodes])
     if filter_arg:
         args += ['-filter_complex', filter_arg]
-    args += reduce(_operator.add, [_get_output_args(node, stream_name_map) for node in output_nodes])
-    args += reduce(_operator.add, [_get_global_args(node) for node in global_nodes], [])
+    args += reduce(operator.add, [_get_output_args(node, stream_name_map) for node in output_nodes])
+    args += reduce(operator.add, [_get_global_args(node) for node in global_nodes], [])
     return args
 
 
-@operator(node_classes={OutputNode, GlobalNode})
+@output_operator()
 def run(node, cmd='ffmpeg'):
     """Run ffmpeg on node graph."""
     if isinstance(cmd, basestring):

ffmpeg/_utils.py

@@ -0,0 +1,27 @@
+import hashlib
+
+
+def _recursive_repr(item):
+    """Hack around python `repr` to deterministically represent dictionaries.
+
+    This is able to represent more things than json.dumps, since it does not require things to be JSON serializable
+    (e.g. datetimes).
+    """
+    if isinstance(item, basestring):
+        result = str(item)
+    elif isinstance(item, list):
+        result = '[{}]'.format(', '.join([_recursive_repr(x) for x in item]))
+    elif isinstance(item, dict):
+        kv_pairs = ['{}: {}'.format(_recursive_repr(k), _recursive_repr(item[k])) for k in sorted(item)]
+        result = '{' + ', '.join(kv_pairs) + '}'
+    else:
+        result = repr(item)
+    return result
+
+
+def get_hash(item):
+    repr_ = _recursive_repr(item).encode('utf-8')
+    return hashlib.md5(repr_).hexdigest()
+
+def get_hash_int(item):
+    return int(get_hash(item), base=16)

ffmpeg/dag.py

@@ -1,31 +1,6 @@
+from ._utils import get_hash, get_hash_int
 from builtins import object
 from collections import namedtuple
-import hashlib
-
-
-def _recursive_repr(item):
-    """Hack around python `repr` to deterministically represent dictionaries.
-
-    This is able to represent more things than json.dumps, since it does not require things to be JSON serializable
-    (e.g. datetimes).
-    """
-    if isinstance(item, basestring):
-        result = str(item)
-    elif isinstance(item, list):
-        result = '[{}]'.format(', '.join([_recursive_repr(x) for x in item]))
-    elif isinstance(item, dict):
-        kv_pairs = ['{}: {}'.format(_recursive_repr(k), _recursive_repr(item[k])) for k in sorted(item)]
-        result = '{' + ', '.join(kv_pairs) + '}'
-    else:
-        result = repr(item)
-    return result
-
-
-def _get_hash(item):
-    hasher = hashlib.sha224()
-    repr_ = _recursive_repr(item)
-    hasher.update(repr_.encode('utf-8'))
-    return hasher.hexdigest()
 
 
 class DagNode(object):
@@ -113,20 +88,6 @@ def get_outgoing_edges(upstream_node, outgoing_edge_map):
 class KwargReprNode(DagNode):
     """A DagNode that can be represented as a set of args+kwargs.
     """
-    def __get_hash(self):
-        hashes = self.__upstream_hashes + [self.__inner_hash]
-        hash_strs = [str(x) for x in hashes]
-        hashes_str = ','.join(hash_strs).encode('utf-8')
-        hash_str = hashlib.md5(hashes_str).hexdigest()
-        return int(hash_str, base=16)
-
-    def __init__(self, incoming_edge_map, name, args, kwargs):
-        self.__incoming_edge_map = incoming_edge_map
-        self.name = name
-        self.args = args
-        self.kwargs = kwargs
-        self.__hash = self.__get_hash()
-
     @property
     def __upstream_hashes(self):
         hashes = []
@@ -137,7 +98,18 @@ class KwargReprNode(DagNode):
     @property
     def __inner_hash(self):
         props = {'args': self.args, 'kwargs': self.kwargs}
-        return _get_hash(props)
+        return get_hash(props)
+
+    def __get_hash(self):
+        hashes = self.__upstream_hashes + [self.__inner_hash]
+        return get_hash_int(hashes)
+
+    def __init__(self, incoming_edge_map, name, args, kwargs):
+        self.__incoming_edge_map = incoming_edge_map
+        self.name = name
+        self.args = args
+        self.kwargs = kwargs
+        self.__hash = self.__get_hash()
 
     def __hash__(self):
         return self.__hash
@@ -149,10 +121,16 @@ class KwargReprNode(DagNode):
     def short_hash(self):
         return '{:x}'.format(abs(hash(self)))[:12]
 
-    def __repr__(self):
+    def long_repr(self, include_hash=True):
         formatted_props = ['{!r}'.format(arg) for arg in self.args]
         formatted_props += ['{}={!r}'.format(key, self.kwargs[key]) for key in sorted(self.kwargs)]
-        return '{}({}) <{}>'.format(self.name, ', '.join(formatted_props), self.short_hash)
+        out = '{}({})'.format(self.name, ', '.join(formatted_props))
+        if include_hash:
+            out += ' <{}>'.format(self.short_hash)
+        return out
+
+    def __repr__(self):
+        return self.long_repr()
 
     @property
     def incoming_edges(self):
@@ -190,7 +168,7 @@ def topo_sort(downstream_nodes):
             marked_nodes.remove(upstream_node)
             sorted_nodes.append(upstream_node)
 
-    unmarked_nodes = [(node, 0) for node in downstream_nodes]
+    unmarked_nodes = [(node, None) for node in downstream_nodes]
     while unmarked_nodes:
         upstream_node, upstream_label = unmarked_nodes.pop()
         visit(upstream_node, upstream_label, None, None)

ffmpeg/nodes.py

@@ -1,27 +1,133 @@
 from __future__ import unicode_literals
 
 from .dag import KwargReprNode
+from ._utils import get_hash_int
+
+
+def _is_of_types(obj, types):
+    valid = False
+    for stream_type in types:
+        if isinstance(obj, stream_type):
+            valid = True
+            break
+    return valid
+
+
+def _get_types_str(types):
+    return ', '.join(['{}.{}'.format(x.__module__, x.__name__) for x in types])
+
+
+class Stream(object):
+    """Represents the outgoing edge of an upstream node; may be used to create more downstream nodes."""
+    def __init__(self, upstream_node, upstream_label, node_types):
+        if not _is_of_types(upstream_node, node_types):
+            raise TypeError('Expected upstream node to be of one of the following type(s): {}; got {}'.format(
+                _get_types_str(node_types), type(upstream_node)))
+        self.node = upstream_node
+        self.label = upstream_label
+
+    def __hash__(self):
+        return get_hash_int([hash(self.node), hash(self.label)])
+
+    def __eq__(self, other):
+        return hash(self) == hash(other)
+
+    def __repr__(self):
+        node_repr = self.node.long_repr(include_hash=False)
+        out = '{}[{!r}] <{}>'.format(node_repr, self.label, self.node.short_hash)
+        return out
 
 
 class Node(KwargReprNode):
     """Node base"""
-    def __init__(self, parents, name, *args, **kwargs):
+    @classmethod
+    def __check_input_len(cls, stream_map, min_inputs, max_inputs):
+        if min_inputs is not None and len(stream_map) < min_inputs:
+            raise ValueError('Expected at least {} input stream(s); got {}'.format(min_inputs, len(stream_map)))
+        elif max_inputs is not None and len(stream_map) > max_inputs:
+            raise ValueError('Expected at most {} input stream(s); got {}'.format(max_inputs, len(stream_map)))
+
+    @classmethod
+    def __check_input_types(cls, stream_map, incoming_stream_types):
+        for stream in stream_map.values():
+            if not _is_of_types(stream, incoming_stream_types):
+                raise TypeError('Expected incoming stream(s) to be of one of the following types: {}; got {}'
+                    .format(_get_types_str(incoming_stream_types), type(stream)))
+
+    @classmethod
+    def __get_stream_map(cls, stream_spec):
+        if stream_spec is None:
+            stream_map = {}
+        elif isinstance(stream_spec, Stream):
+            stream_map = {None: stream_spec}
+        elif isinstance(stream_spec, (list, tuple)):
+            stream_map = dict(enumerate(stream_spec))
+        elif isinstance(stream_spec, dict):
+            stream_map = stream_spec
+        return stream_map
+
+    @classmethod
+    def __get_incoming_edge_map(cls, stream_map):
         incoming_edge_map = {}
-        for downstream_label, parent in enumerate(parents):
-            upstream_label = 0  # assume nodes have a single output (FIXME)
-            upstream_node = parent
-            incoming_edge_map[downstream_label] = (upstream_node, upstream_label)
+        for downstream_label, upstream in stream_map.items():
+            incoming_edge_map[downstream_label] = (upstream.node, upstream.label)
+        return incoming_edge_map
+
+    def __init__(self, stream_spec, name, incoming_stream_types, outgoing_stream_type, min_inputs, max_inputs, args,
+            kwargs):
+        stream_map = self.__get_stream_map(stream_spec)
+        self.__check_input_len(stream_map, min_inputs, max_inputs)
+        self.__check_input_types(stream_map, incoming_stream_types)
+        incoming_edge_map = self.__get_incoming_edge_map(stream_map)
         super(Node, self).__init__(incoming_edge_map, name, args, kwargs)
+        self.__outgoing_stream_type = outgoing_stream_type
+
+    def stream(self, label=None):
+        """Create an outgoing stream originating from this node.
+
+        More nodes may be attached onto the outgoing stream.
+        """
+        return self.__outgoing_stream_type(self, label)
+
+    def __getitem__(self, label):
+        """Create an outgoing stream originating from this node; syntactic sugar for ``self.stream(label)``.
+        """
+        return self.stream(label)
+
+
+class FilterableStream(Stream):
+    def __init__(self, upstream_node, upstream_label):
+        super(FilterableStream, self).__init__(upstream_node, upstream_label, {InputNode, FilterNode})
 
 
 class InputNode(Node):
     """InputNode type"""
-    def __init__(self, name, *args, **kwargs):
-        super(InputNode, self).__init__(parents=[], name=name, *args, **kwargs)
+    def __init__(self, name, args=[], kwargs={}):
+        super(InputNode, self).__init__(
+            stream_spec=None,
+            name=name,
+            incoming_stream_types={},
+            outgoing_stream_type=FilterableStream,
+            min_inputs=0,
+            max_inputs=0,
+            args=args,
+            kwargs=kwargs
+        )
 
 
 class FilterNode(Node):
-    """FilterNode"""
+    def __init__(self, stream_spec, name, max_inputs=1, args=[], kwargs={}):
+        super(FilterNode, self).__init__(
+            stream_spec=stream_spec,
+            name=name,
+            incoming_stream_types={FilterableStream},
+            outgoing_stream_type=FilterableStream,
+            min_inputs=1,
+            max_inputs=max_inputs,
+            args=args,
+            kwargs=kwargs
+        )
+
     def _get_filter(self):
         params_text = self.name
         arg_params = ['{}'.format(arg) for arg in self.args]
@@ -33,20 +139,61 @@ class FilterNode(Node):
 
 
 class OutputNode(Node):
-    """OutputNode"""
-    pass
+    def __init__(self, stream, name, args=[], kwargs={}):
+        super(OutputNode, self).__init__(
+            stream_spec=stream,
+            name=name,
+            incoming_stream_types={FilterableStream},
+            outgoing_stream_type=OutputStream,
+            min_inputs=1,
+            max_inputs=1,
+            args=args,
+            kwargs=kwargs
+        )
+
+
+class OutputStream(Stream):
+    def __init__(self, upstream_node, upstream_label):
+        super(OutputStream, self).__init__(upstream_node, upstream_label, {OutputNode, GlobalNode})
+
+
+class MergeOutputsNode(Node):
+    def __init__(self, stream, name):
+        super(MergeOutputsNode, self).__init__(
+            stream_spec=None,
+            name=name,
+            incoming_stream_types={OutputStream},
+            outgoing_stream_type=OutputStream,
+            min_inputs=1,
+            max_inputs=None
+        )
 
 
 class GlobalNode(Node):
-    def __init__(self, parent, name, *args, **kwargs):
-        if not isinstance(parent, OutputNode):
-            raise RuntimeError('Global nodes can only be attached after output nodes')
-        super(GlobalNode, self).__init__([parent], name, *args, **kwargs)
+    def __init__(self, stream, name, args=[], kwargs={}):
+        super(GlobalNode, self).__init__(
+            stream_spec=stream,
+            name=name,
+            incoming_stream_types={OutputStream},
+            outgoing_stream_type=OutputStream,
+            min_inputs=1,
+            max_inputs=1,
+            args=args,
+            kwargs=kwargs
+        )
 
 
-def operator(node_classes={Node}, name=None):
+def stream_operator(stream_classes={Stream}, name=None):
     def decorator(func):
         func_name = name or func.__name__
-        [setattr(node_class, func_name, func) for node_class in node_classes]
+        [setattr(stream_class, func_name, func) for stream_class in stream_classes]
         return func
     return decorator
+
+
+def filter_operator(name=None):
+    return stream_operator(stream_classes={FilterableStream}, name=name)
+
+
+def output_operator(name=None):
+    return stream_operator(stream_classes={OutputStream}, name=name)

ffmpeg/tests/test_ffmpeg.py

@@ -39,11 +39,8 @@ def test_fluent_concat():
     concat1 = ffmpeg.concat(trimmed1, trimmed2, trimmed3)
     concat2 = ffmpeg.concat(trimmed1, trimmed2, trimmed3)
     concat3 = ffmpeg.concat(trimmed1, trimmed3, trimmed2)
-    concat4 = ffmpeg.concat()
-    concat5 = ffmpeg.concat()
     assert concat1 == concat2
     assert concat1 != concat3
-    assert concat4 == concat5
 
 
 def test_fluent_output():
@@ -66,19 +63,28 @@ def test_fluent_complex_filter():
     )
 
 
-def test_repr():
+def test_node_repr():
     in_file = ffmpeg.input('dummy.mp4')
     trim1 = ffmpeg.trim(in_file, start_frame=10, end_frame=20)
     trim2 = ffmpeg.trim(in_file, start_frame=30, end_frame=40)
     trim3 = ffmpeg.trim(in_file, start_frame=50, end_frame=60)
     concatted = ffmpeg.concat(trim1, trim2, trim3)
     output = ffmpeg.output(concatted, 'dummy2.mp4')
-    assert repr(in_file) == "input(filename={!r}) <{}>".format('dummy.mp4', in_file.short_hash)
-    assert repr(trim1) == "trim(end_frame=20, start_frame=10) <{}>".format(trim1.short_hash)
-    assert repr(trim2) == "trim(end_frame=40, start_frame=30) <{}>".format(trim2.short_hash)
-    assert repr(trim3) == "trim(end_frame=60, start_frame=50) <{}>".format(trim3.short_hash)
-    assert repr(concatted) == "concat(n=3) <{}>".format(concatted.short_hash)
-    assert repr(output) == "output(filename={!r}) <{}>".format('dummy2.mp4', output.short_hash)
+    assert repr(in_file.node) == "input(filename={!r}) <{}>".format('dummy.mp4', in_file.node.short_hash)
+    assert repr(trim1.node) == "trim(end_frame=20, start_frame=10) <{}>".format(trim1.node.short_hash)
+    assert repr(trim2.node) == "trim(end_frame=40, start_frame=30) <{}>".format(trim2.node.short_hash)
+    assert repr(trim3.node) == "trim(end_frame=60, start_frame=50) <{}>".format(trim3.node.short_hash)
+    assert repr(concatted.node) == "concat(n=3) <{}>".format(concatted.node.short_hash)
+    assert repr(output.node) == "output(filename={!r}) <{}>".format('dummy2.mp4', output.node.short_hash)
+
+
+def test_stream_repr():
+    in_file = ffmpeg.input('dummy.mp4')
+    assert repr(in_file) == "input(filename={!r})[None] <{}>".format('dummy.mp4', in_file.node.short_hash)
+    split0 = in_file.filter_multi_output('split')[0]
+    assert repr(split0) == "split()[0] <{}>".format(split0.node.short_hash)
+    dummy_out = in_file.filter_multi_output('dummy')['out']
+    assert repr(dummy_out) == "dummy()[{!r}] <{}>".format(dummy_out.label, dummy_out.node.short_hash)
 
 
 def test_get_args_simple():
@@ -201,7 +207,7 @@ def test_pipe():
     p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
 
     in_data = bytes(bytearray([random.randint(0,255) for _ in range(frame_size * frame_count)]))
-    p.stdin.write(in_data)  # note: this could block, in which case need to use threads 
+    p.stdin.write(in_data)  # note: this could block, in which case need to use threads
     p.stdin.close()
 
     out_data = p.stdout.read()