Repository: just-work/fffw

Issue:
Refactor working with filter complex destinations
Now filter nodes are connected to `Dest` object with `fc.get_video_dest()`, and then `Dest` is connected to `Codec` in `fc.add_output`. 

Filter nodes should be connected to codecs directly, and Muxer should accept codec list as input parameters.

Discussion:

Patch:
diff --git a/fffw/encoding/__init__.py b/fffw/encoding/__init__.py
index 7caf1a8..fa90361 100644
--- a/fffw/encoding/__init__.py
+++ b/fffw/encoding/__init__.py
@@ -1,9 +1,5 @@
-from .codec import *
-from .muxer import *
 from .ffmpeg import *
 
 __all__ = (
-        codec.__all__ +  # type: ignore
-        muxer.__all__ +  # type: ignore
         ffmpeg.__all__  # type: ignore
 )
diff --git a/fffw/encoding/codec.py b/fffw/encoding/codec.py
deleted file mode 100644
index 13620b0..0000000
--- a/fffw/encoding/codec.py
+++ /dev/null
@@ -1,130 +0,0 @@
-import abc
-from typing import List
-
-from fffw.graph import base
-from fffw.wrapper import BaseWrapper
-
-__all__ = [
-    'VideoCodec',
-    'AudioCodec',
-]
-
-
-class BaseCodec(BaseWrapper, base.Node, metaclass=abc.ABCMeta):
-    """
-    Abstract base class for codec nodes.
-    """
-    codec_type: base.StreamType
-
-    arguments = [('map', '-map ')]
-
-    @property
-    def enabled(self) -> bool:
-        return True
-
-    @enabled.setter
-    def enabled(self, value: bool) -> None:
-        raise RuntimeError("codecs can't be disabled")
-
-    @property
-    @abc.abstractmethod
-    def codec_name(self) -> str:
-        raise NotImplementedError()
-
-    @property
-    def map(self) -> str:
-        return self._args['map']
-
-    @map.setter
-    def map(self, value: str) -> None:
-        self._args['map'] = value
-
-    def render(self, partial: bool = False) -> List[str]:
-        """ codec output node is already rendered in filter graph."""
-        return []
-
-    def connect(self, dest: base.Dest) -> None:
-        """ Connects destination to codec node."""
-        assert isinstance(dest, base.Dest), "Codec connects to Dest"
-        self.map = f'[{dest.name}]'
-
-    def __repr__(self) -> str:
-        return "<%s>(%s)" % (
-            self.codec_name,
-            ','.join('%s=%s' % (k, self._args[k]) for k in self._key_mapping
-                     if self._args[k])
-        )
-
-    def connect_edge(self, edge: base.Edge) -> base.Edge:
-        """ Connects source to codec through an edge."""
-        src = edge.input
-        assert isinstance(src, base.Source), "Codec connects to Source"
-        assert src.name, "Source file has not stream of desired type"
-        if self.map:
-            # normal Node can connect with source single time only,
-            # BaseCodec can connect multiple times via "-map" arguments
-
-            # FIXME: GH/JW #2 this should return base.Edge. See Source._edge
-            # noinspection PyTypeChecker
-            return None  # type: ignore
-        self.map = src.name
-        return edge
-
-    def __ror__(self, other: base.Dest) -> None:
-        """ Connects destination to codec. """
-        # TODO: unite Dest and codec into single object.
-        if not isinstance(other, base.Dest):
-            return NotImplemented
-        self.connect(other)
-
-
-class VideoCodec(BaseCodec):
-    codec_type = base.VIDEO
-    arguments = [
-        ('map', '-map '),
-        ('vbsf', '-bsf:v '),
-        ('vcodec', '-c:v '),
-        ('pass', '-pass '),
-        ('pix_fmt', '-pix_fmt '),
-        ('preset', '-preset '),
-        ('tune', '-tune '),
-        ('flags', '-flags '),
-        ('force_key_frames', '-force_key_frames '),
-        ('vprofile', '-profile:v '),
-        ('level', '-level '),
-        ('crf', '-crf '),
-        ('minrate', '-minrate '),
-        ('maxrate', '-maxrate '),
-        ('bufsize', '-bufsize '),
-        ('gop', '-g '),
-        ('vrate', '-r '),
-        ('vbitrate', '-b:v '),
-        ('vaspect', '-aspect '),
-        ('reframes', '-refs '),
-        ('mbd', '-mbd '),
-        ('trellis', '-trellis '),
-        ('cmp', '-cmp '),
-        ('subcmp', '-subcmp '),
-        ('x265', '-x265-params '),
-    ]
-
-    @property
-    def codec_name(self) -> str:
-        return self._args['vcodec']
-
-
-class AudioCodec(BaseCodec):
-    codec_type = base.AUDIO
-    arguments = [
-        ('map', '-map '),
-        ('absf', '-bsf:a '),
-        ('acodec', '-c:a '),
-        ('aprofile', '-profile:a '),
-        ('abitrate', '-b:a '),
-        ('arate', '-ar '),
-        ('achannels', '-ac '),
-    ]
-
-    @property
-    def codec_name(self) -> str:
-        return self._args['acodec']
diff --git a/fffw/encoding/ffmpeg.py b/fffw/encoding/ffmpeg.py
index c5ce914..a82c289 100644
--- a/fffw/encoding/ffmpeg.py
+++ b/fffw/encoding/ffmpeg.py
@@ -1,8 +1,7 @@
-from itertools import chain
-from typing import List, Tuple, Any, Union
+from typing import List, Any, Union, Optional
 
-from fffw.encoding import Muxer, codec
-from fffw.graph import FilterComplex, base, inputs
+from fffw.graph import base, inputs, outputs
+from fffw.graph.complex import FilterComplex
 from fffw.wrapper import BaseWrapper, ensure_binary
 
 __all__ = ['FFMPEG']
@@ -52,118 +51,104 @@ class FFMPEG(BaseWrapper):
         ('segment_list_flags', '-segment_list_flags '),
     ]
 
-    def __init__(self, *sources: Union[inputs.Input, str], **kw: Any):
+    def __init__(self, *sources: Union[inputs.Input, str],
+                 output: Union[None, outputs.Output, str] = None, **kw: Any):
         """
         :param sources: list of input files (or another ffmpeg sources)
         :param kw: ffmpeg command line arguments
         """
         super(FFMPEG, self).__init__(**kw)
-        self.__outputs: List[Tuple[Tuple[codec.BaseCodec, ...], Muxer]] = []
-        self.__vdest = self.__adest = 0
         self.__input_list = inputs.InputList(
             *(
                 inputs.Input(input_file=src)
                 if isinstance(src, str) else src
                 for src in sources
             ))
+        self.__output_list = outputs.OutputList()
+        if output:
+            self.__output_list.append(
+                outputs.Output(output_file=output)
+                if isinstance(output, str) else output
+            )
 
-    def init_filter_complex(self) -> FilterComplex:
-        # TODO #9 refactor filter complex initialization
-        assert not self.__outputs, "outputs already defined"
-        fc = FilterComplex(self.__input_list)
-        self._args['filter_complex'] = fc
-        return fc
+    def __lt__(self, other: inputs.Input) -> None:
+        """ Adds new source file.
+        """
+        if not isinstance(other, inputs.Input):
+            return NotImplemented
+        self.add_input(other)
+
+    def __gt__(self, other: outputs.Output) -> None:
+        """ Adds new output file."""
+        if not isinstance(other, outputs.Output):
+            return NotImplemented
+        self.add_output(other)
+
+    def __setattr__(self, key: str, value: Any) -> None:
+        # TODO: #9 refactor working with args
+        if key == 'filter_complex':
+            raise NotImplementedError("use init_filter_complex instead")
+        if key == 'inputfile':
+            raise NotImplementedError("use add_input instead")
+        return super(FFMPEG, self).__setattr__(key, value)
 
     @property
     def filter_complex(self) -> FilterComplex:
+        # TODO #9 refactor filter complex initialization
         return self._args['filter_complex']
 
-    def get_args(self) -> List[bytes]:
-        return (ensure_binary([self.command]) +
-                ensure_binary(self.__input_list.get_args()) +
-                super(FFMPEG, self).get_args() +
-                ensure_binary(self.get_output_args()))
-
-    def get_output_args(self) -> List[bytes]:
-        result: List[bytes] = []
-        for codecs, muxer in self.__outputs:
-            args = list(chain.from_iterable(c.get_args() for c in codecs))
-            output = [ensure_binary(muxer.output)]
-            result.extend(args + muxer.get_args() + output)
-        return result
-
-    def add_input(self, input_file: inputs.Input) -> None:
-        """ Adds new source to ffmpeg.
-        """
-        assert isinstance(input_file, inputs.Input)
-        self.__input_list.append(input_file)
-
-    def get_free_source(self, kind: base.StreamType) -> base.Source:
+    def _get_free_source(self, kind: base.StreamType) -> base.Source:
         """
         :param kind: stream type
         :return: first stream of this kind not connected to destination
         """
         for stream in self.__input_list.streams:
-            if stream.kind != kind or stream.edge is not None:
+            if stream.kind != kind or stream.connected:
                 continue
             return stream
         else:
             raise RuntimeError("no free streams")
 
-    def add_codec(self, c: codec.BaseCodec) -> None:
-        """ Connect codec to filter graph output or input stream."""
-        # TODO: #14 refactor connecting codec while merging Dest and BaseCodec
-        node: Union[base.Source, base.Dest]
-        if c.map:
-            try:
-                node = next(filter(lambda s: s.name == c.map,
-                                   self.__input_list.streams))
-            except StopIteration:
-                raise RuntimeError("No stream for map")
-        else:
-            try:
-                node = self.get_free_source(c.codec_type)
-            except RuntimeError:
-                # no free sources, search for free dest in fc
-                fc = self.filter_complex
-                if fc is None:
-                    raise
-                if c.codec_type == base.VIDEO:
-                    node = fc.get_video_dest(self.__vdest, create=False)
-                    self.__vdest += 1
-                else:
-                    node = fc.get_audio_dest(self.__adest, create=False)
-                    self.__adest += 1
-        if c.codec_type != node.kind:
-            raise RuntimeError("stream and codec type mismatch")
-        node | c
-
-    def add_output(self, muxer: Muxer, *codecs: codec.BaseCodec) -> None:
-        # TODO: #14 muxer should contain codecs and codecs should connect
-        #  directly to input stream or graph filter.
-        assert isinstance(muxer, Muxer)
-        for c in codecs:
-            self.add_codec(c)
-        self.__outputs.append((codecs, muxer))
+    def _add_codec(self, c: outputs.Codec) -> Optional[outputs.Codec]:
+        """ Connect codec to filter graph output or input stream.
 
-    @property
-    def outputs(self) -> List[Tuple[Tuple[codec.BaseCodec, ...], Muxer]]:
-        return list(self.__outputs)
+        :param c: codec to connect to free source
+        :returns: None of codec already connected to filter graph or codec
+            itself if it was connected successfully to input stream.
+        """
+        if c.connected:
+            return None
+        node = self._get_free_source(c.kind)
+        node.connect_dest(c)
+        return c
 
-    def __lt__(self, other: inputs.Input) -> None:
-        """ Adds new source file.
+    def init_filter_complex(self) -> FilterComplex:
+        # TODO #9 refactor filter complex initialization
+        fc = FilterComplex(self.__input_list, self.__output_list)
+        self._args['filter_complex'] = fc
+        return fc
+
+    def get_args(self) -> List[bytes]:
+        with base.Namer():
+            # Namer context is used to generate unique output stream names
+            return (ensure_binary([self.command]) +
+                    self.__input_list.get_args() +
+                    super(FFMPEG, self).get_args() +
+                    self.__output_list.get_args())
+
+    def add_input(self, input_file: inputs.Input) -> None:
+        """ Adds new source to ffmpeg.
         """
-        if not isinstance(other, inputs.Input):
-            return NotImplemented
-        self.add_input(other)
+        assert isinstance(input_file, inputs.Input)
+        self.__input_list.append(input_file)
 
-    def __setattr__(self, key: str, value: Any) -> None:
-        # TODO: #9 refactor working with args
-        if key == 'filter_complex':
-            raise NotImplementedError("use init_filter_complex instead")
-        if key == 'inputfile':
-            raise NotImplementedError("use add_input instead")
-        return super(FFMPEG, self).__setattr__(key, value)
+    def add_output(self, output: outputs.Output) -> None:
+        """
+        Adds output file to ffmpeg and connect it's codecs to free sources.
+        """
+        self.__output_list.append(output)
+        for codec in output.codecs:
+            self._add_codec(codec)
 
     def handle_stderr(self, line: str) -> None:
         """
diff --git a/fffw/encoding/muxer.py b/fffw/encoding/muxer.py
deleted file mode 100644
index 3d812f5..0000000
--- a/fffw/encoding/muxer.py
+++ /dev/null
@@ -1,101 +0,0 @@
-from typing import List, Any, Tuple
-
-from fffw.wrapper import BaseWrapper, ensure_binary
-
-__all__ = [
-    'FLVMuxer',
-    'HLSMuxer',
-    'MP3Muxer',
-    'MP4Muxer',
-    'NullMuxer',
-    'Muxer',
-    'TeeMuxer',
-]
-
-
-class Muxer(BaseWrapper):
-    format: str
-
-    def __init__(self, output: str, **kw: Any):
-        self.__output = output
-        super(Muxer, self).__init__(**kw)
-
-    def __repr__(self) -> str:
-        return "<%s>(%s)" % (
-            self.output,
-            self.get_opts()
-        )
-
-    @property
-    def output(self) -> str:
-        return self.__output
-
-    @output.setter
-    def output(self, value: str) -> None:
-        self.__output = value
-
-    def get_args(self) -> List[bytes]:
-        args = super(Muxer, self).get_args()
-        return ensure_binary(['-f', self.format]) + args
-
-    def get_opts(self) -> str:
-        """ Returns muxer options formatted with ':' delimiter."""
-        return ':'.join('%s=%s' % (self.key_to_opt(k), self._args[k])
-                        for k in self._args_order if self._args[k])
-
-    def key_to_opt(self, k: str) -> str:
-        return self._key_mapping[k].strip('- ')
-
-
-class FLVMuxer(Muxer):
-    format = 'flv'
-
-
-class MP4Muxer(Muxer):
-    format = 'mp4'
-
-
-class MP3Muxer(Muxer):
-    format = 'mp3'
-
-
-class NullMuxer(Muxer):
-    format = 'null'
-
-
-class HLSMuxer(Muxer):
-    format = 'hls'
-
-    arguments = [
-        ('method', '-method '),
-        ('segment_size', '-hls_time '),
-        ('manifest_size', '-hls_list_size '),
-    ]
-
-
-class TeeMuxer(Muxer):
-    format = 'tee'
-
-    def __init__(self, *muxers: Muxer) -> None:
-        # tee muxer does not require output file argument, instead it aggregates
-        # multiple other muxer definitions.
-        # FIXME: fix typing for this.
-        # noinspection PyTypeChecker
-        super(TeeMuxer, self).__init__(None)  # type: ignore
-        self.muxers = muxers
-
-    @property
-    def output(self) -> str:
-        return '|'.join(['[f={format}{opts}]{output}'.format(
-            format=m.format, opts=self._format_opts(m.get_opts()),
-            output=m.output) for m in self.muxers])
-
-    @output.setter
-    def output(self, muxers: Tuple[Muxer]) -> None:
-        for m in muxers:
-            assert isinstance(m, Muxer)
-        self.muxers = muxers
-
-    @staticmethod
-    def _format_opts(opts: str) -> str:
-        return (':' + opts) if opts else ''
diff --git a/fffw/graph/__init__.py b/fffw/graph/__init__.py
index 34b6153..8838a4b 100644
--- a/fffw/graph/__init__.py
+++ b/fffw/graph/__init__.py
@@ -1,11 +1,15 @@
+from .base import *
 from .complex import *
 from .filters import *
-from .base import *
 from .inputs import *
+from .meta import *
+from .outputs import *
 
 __all__ = (
+        base.__all__ +  # type: ignore
         complex.__all__ +  # type: ignore
         filters.__all__ +  # type: ignore
-        base.__all__ +  # type: ignore
-        inputs.__all__  # type: ignore
+        inputs.__all__ +  # type: ignore
+        meta.__all__ +  # type: ignore
+        outputs.__all__  # type: ignore
 )
diff --git a/fffw/graph/base.py b/fffw/graph/base.py
index 0244603..9300564 100644
--- a/fffw/graph/base.py
+++ b/fffw/graph/base.py
@@ -7,7 +7,7 @@ __all__ = [
 import abc
 from collections import Counter
 from enum import Enum
-from typing import Dict, Any
+from typing import Dict, Any, TypeVar, Type, overload
 from typing import Optional, List, Union
 
 from fffw.graph.meta import Meta
@@ -21,42 +21,6 @@ class StreamType(Enum):
 VIDEO = StreamType.VIDEO
 AUDIO = StreamType.AUDIO
 
-
-class Namer:
-    """ Unique stream identifiers generator."""
-    _stack: List["Namer"] = []
-
-    @classmethod
-    def name(cls, edge: "Edge") -> str:
-        current = cls._stack[0]
-        return current._name(edge)
-
-    def __init__(self) -> None:
-        self._counters: Dict[str, int] = Counter()
-        self._cache: Dict[int, str] = dict()
-
-    def __enter__(self) -> "Namer":
-        self._stack.append(self)
-        return self._stack[0]
-
-    def __exit__(self, *_: Any) -> None:
-        self._stack.pop(-1)
-
-    def _name(self, edge: "Edge") -> str:
-        if id(edge) not in self._cache:
-            node = edge.input
-            if isinstance(node, Node):
-                prefix = f'{node.kind.value}:{node.name}'
-                # generating unique edge id by src node kind and name
-                name = f'{prefix}{self._counters[prefix]}'
-                self._counters[prefix] += 1
-            else:
-                name = node.name
-            # caching edge name for idempotency
-            self._cache[id(edge)] = name
-        return self._cache[id(edge)]
-
-
 InputType = Union["Source", "Node"]
 OutputType = Union["Dest", "Node"]
 
@@ -94,21 +58,24 @@ class Dest(Traversable):
     Must connect to single filter output only.
     """
 
-    def __init__(self, name: str, kind: StreamType) -> None:
+    def __init__(self, kind: StreamType) -> None:
         """
-        :param name: internal ffmpeg stream name ("v:0", "a:1")
         :param kind: stream kind (VIDEO/AUDIO)
         """
         self._edge: Optional[Edge] = None
         self._kind = kind
-        self._name = name
 
     def __repr__(self) -> str:
         return f"Dest('{self.name}')"
 
     @property
     def name(self) -> str:
-        return self._name
+        """
+        :returns: edge name (i.e. [vout0]) for codec `-map` argument only.
+        """
+        if self._edge is None:
+            raise RuntimeError("Dest not connected")
+        return f'[{self._edge.name}]'
 
     @property
     def kind(self) -> StreamType:
@@ -118,6 +85,10 @@ class Dest(Traversable):
     def meta(self) -> Optional[Meta]:
         return self.get_meta_data(self)
 
+    @property
+    def edge(self) -> Optional["Edge"]:
+        return self._edge
+
     def get_meta_data(self, dst: OutputType) -> Optional[Meta]:
         if self._edge is None:
             raise RuntimeError("Dest not connected")
@@ -184,7 +155,10 @@ class Edge(Traversable):
         current node.
         """
         if isinstance(self.output, Dest):
-            return self.output.name
+            # For final edges name is generated from destination node, like
+            # [vout0] or [aout1]
+            return Namer.name(self)
+        # For edges connected to other filters disabled source nodes are skipped
         edge = self
         node = self.input
         while not getattr(node, 'enabled', True) and isinstance(node, Node):
@@ -194,7 +168,7 @@ class Edge(Traversable):
             node = edge.input
         return Namer.name(edge)
 
-    def _connect_source(self, src: Union["Source", "Node"]) -> None:
+    def _connect_source(self, src: InputType) -> None:
         """ Connects input node to the edge.
 
         :param src: source stream or filter output node.
@@ -204,8 +178,15 @@ class Edge(Traversable):
                                % self.__input)
         self.__input = src
 
-    def _connect_dest(self,
-                      dest: Union["Dest", "Node"]) -> Union["Dest", "Node"]:
+    @overload
+    def _connect_dest(self, dest: "Node") -> "Node":
+        ...
+
+    @overload
+    def _connect_dest(self, dest: Dest) -> Dest:
+        ...
+
+    def _connect_dest(self, dest: OutputType) -> OutputType:
         """ Connects output node to the edge.
 
         :param dest: output stream or filter input node
@@ -232,7 +213,7 @@ class Node(Traversable):
     """ Graph node describing ffmpeg filter."""
 
     kind: StreamType  # filter type (VIDEO/AUDIO)
-    name: str  # filter name
+    filter: str  # filter name
     input_count: int = 1  # number of inputs
     output_count: int = 1  # number of outputs
 
@@ -244,14 +225,24 @@ class Node(Traversable):
     def __repr__(self) -> str:
         inputs = [f"[{str(i.name if i else '---')}]" for i in self.inputs]
         outputs = [f"[{str(i.name if i else '---')}]" for i in self.outputs]
-        return f"{''.join(inputs)}{self.name}{''.join(outputs)}"
+        return f"{''.join(inputs)}{self.filter}{''.join(outputs)}"
 
-    def __or__(self, other: Union["Node", "Dest"]) -> Union["Node", "Dest"]:
+    def __or__(self, other: "Node") -> "Node":
         """
         connect output edge to node
-        :return: connected edge
+        :return: connected node
         """
-        if not isinstance(other, (Node, Dest)):
+        if not isinstance(other, Node):
+            return NotImplemented
+        return self.connect_dest(other)
+
+    def __gt__(self, other: Dest) -> Dest:
+        """
+        connects output edge to destination
+        :param other: destination codec
+        :return: connected codec
+        """
+        if not isinstance(other, Dest):
             return NotImplemented
         return self.connect_dest(other)
 
@@ -336,14 +327,19 @@ class Node(Traversable):
         for edge in self.inputs:
             if edge is None:
                 raise RuntimeError("input is none")
-            inputs.append(f"[{edge.name}]")
+            # Add Source name (i.e. 0:v) or unique node name (v:scale1) as
+            # filter input.
+            inputs.append(f'[{edge.name}]')
 
         for edge in self.outputs:
             if edge is None:
                 if partial:
+                    # outputs not connected, using a stub.
                     outputs.append('[---]')
                     continue
                 raise RuntimeError("output is none")
+            # Skip outgoing disabled nodes till Dest node, to get proper
+            # output name if current node is last enabled before destination.
             node = edge.output
             while isinstance(node, Node) and not node.enabled:
                 # if next node is disabled, use next edge id
@@ -353,9 +349,10 @@ class Node(Traversable):
                 if edge is None:
                     raise RuntimeError("output is none")
                 node = edge.output
+            # Add unique output edge name (vout0 or a:volume1) to filter output
             outputs.append(f"[{edge.name}]")
         args = '=' + self.args if self.args else ''
-        return ''.join(inputs) + self.name + args + ''.join(outputs)
+        return ''.join(inputs) + self.filter + args + ''.join(outputs)
 
     def connect_edge(self, edge: "Edge") -> "Edge":
         """ Connects and edge to one of filter inputs
@@ -368,8 +365,15 @@ class Node(Traversable):
         self.inputs[self.inputs.index(None)] = edge
         return edge
 
-    def connect_dest(self,
-                     other: Union["Node", "Dest"]) -> Union["Node", "Dest"]:
+    @overload
+    def connect_dest(self, other: "Node") -> "Node":
+        ...
+
+    @overload
+    def connect_dest(self, other: Dest) -> Dest:
+        ...
+
+    def connect_dest(self, other: OutputType) -> OutputType:
         """ Connects next filter or output to one of filter outputs.
 
         :param other: next filter or output stream
@@ -383,22 +387,20 @@ class Node(Traversable):
         return other
 
 
-class Source(Traversable):
+class Source(Traversable, metaclass=abc.ABCMeta):
     """ Graph node containing audio or video input.
 
     Must connect to single graph edge only as a source
     """
 
-    def __init__(self, name: Optional[str], kind: StreamType,
+    def __init__(self, kind: StreamType,
                  meta: Optional[Meta] = None) -> None:
         """
-        :param name: ffmpeg internal input stream name ("v:0", "a:1")
         :param kind: stream type (VIDEO/AUDIO)
         :param meta: stream metadata
         """
-        self._edge: Optional[Edge] = None
+        self._outputs: List[Edge] = []
         self._kind = kind
-        self._name = name
         self._meta = meta
 
     def __repr__(self) -> str:
@@ -411,20 +413,26 @@ class Source(Traversable):
         """
         if not isinstance(other, Node):
             return NotImplemented
-        return self.connect(other)
+        return self.connect_dest(other)
+
+    def __gt__(self, other: Dest) -> Dest:
+        """
+        Connects a codec to a source
+        :param other: codec that will process current source stream
+        :return: destination object
+        """
+        if not isinstance(other, Dest):
+            return NotImplemented
+        return self.connect_dest(other)
 
     @property
-    def name(self) -> str:
-        if self._name is None:
-            raise RuntimeError("Source name not set")
-        return self._name
+    def connected(self) -> bool:
+        return bool(self._outputs)
 
     @property
-    def edge(self) -> Optional["Edge"]:
-        """
-        :returns: an edge connected to current source.
-        """
-        return self._edge
+    @abc.abstractmethod
+    def name(self) -> str:
+        raise NotImplementedError()
 
     @property
     def kind(self) -> StreamType:
@@ -440,36 +448,109 @@ class Source(Traversable):
         """
         return self._meta
 
-    def connect(self, other: Node) -> Node:
-        """ Connects a source to a filter or output
+    @overload
+    def connect_dest(self, other: Node) -> Node:
+        ...
 
-        :param other: filter consuming current input stream
-        :return filter connected to current stream
-        """
-        if not isinstance(other, Node):
-            raise ValueError("only node allowed")
-        if self._edge is not None and not getattr(other, 'map', None):
-            raise RuntimeError("Source %s is already connected to %s"
-                               % (self.name, self._edge))
+    @overload
+    def connect_dest(self, other: Dest) -> Dest:
+        ...
+
+    def connect_dest(self, other: OutputType) -> OutputType:
+        if not isinstance(other, (Node, Dest)):
+            raise ValueError("only node or dest allowed")
         edge = Edge(input=self, output=other)
-        self._edge = self._edge or other.connect_edge(edge)
+        other.connect_edge(edge)
+        self._outputs.append(edge)
         return other
 
     def get_meta_data(self, dst: OutputType) -> Optional[Meta]:
         return self._meta
 
     def render(self, partial: bool = False) -> List[str]:
-        if self._edge is None:
-            return []
+        result = []
+        edge: Optional[Edge]
+        for edge in self._outputs:
+            node = edge.output
+            # if output node is disabled, use next edge identifier.
+            if isinstance(node, Node) and not node.enabled:
+                edge = node.outputs[0]
+                if edge is None:
+                    if partial:
+                        return []
+                    raise RuntimeError("Skipped node is not ready for render")
+            result.extend(edge.render(partial=partial))
+        return result
 
-        node = self._edge.output
-        # if output node is disabled, use next edge identifier.
-        if isinstance(node, Node) and not node.enabled:
-            edge: Optional[Edge] = node.outputs[0]
-            if edge is None:
-                if partial:
-                    return []
-                raise RuntimeError("Skipped node is not ready for render")
-        else:
-            edge = self._edge
-        return edge.render(partial=partial)
+
+Obj = TypeVar('Obj')
+
+
+class Once:
+    """ Property that must be set exactly once."""
+
+    def __init__(self, attr_name: str) -> None:
+        """
+        :param attr_name: instance attribute name
+        """
+        self.attr_name = attr_name
+
+    def __get__(self, instance: Obj, owner: Type[Obj]) -> Any:
+        try:
+            return instance.__dict__[self.attr_name]
+        except KeyError:
+            raise RuntimeError(f"{self.attr_name} is not initialized")
+
+    def __set__(self, instance: Obj, value: Any) -> None:
+        if self.attr_name in instance.__dict__:
+            raise RuntimeError(f"{self.attr_name} already initialized")
+        instance.__dict__[self.attr_name] = value
+
+
+class Namer:
+    """ Unique stream identifiers generator."""
+    _stack: List["Namer"] = []
+
+    @classmethod
+    def name(cls, obj: Edge) -> str:
+        current = cls._stack[0]
+        return current._name(obj)
+
+    def __init__(self) -> None:
+        self._counters: Dict[str, int] = Counter()
+        self._cache: Dict[int, str] = dict()
+
+    def __enter__(self) -> "Namer":
+        self._stack.append(self)
+        return self._stack[0]
+
+    def __exit__(self, *_: Any) -> None:
+        self._stack.pop(-1)
+
+    def _name(self, edge: Edge) -> str:
+        """
+        Generates name for an edge in filter graph.
+
+        :param edge: edge that needs to be named
+        :returns: unique Dest name if edge leads to destination (i.e. vout0),
+        Source name if edge starts from input stream (i.e. 0:v) or unique
+        input Node name generated from node filter (i.e. v:overlay1).
+        """
+        if id(edge) not in self._cache:
+            src = edge.input
+            dst = edge.output
+            if isinstance(dst, Dest):
+                prefix = f'{dst.kind.value}out'
+                # generating unique edge id by dst kind
+                name = f'{prefix}{self._counters[prefix]}'
+                self._counters[prefix] += 1
+            elif isinstance(src, Node):
+                prefix = f'{src.kind.value}:{src.filter}'
+                # generating unique edge id by src node kind and name
+                name = f'{prefix}{self._counters[prefix]}'
+                self._counters[prefix] += 1
+            else:
+                name = f'{src.name}'
+            # caching edge name for idempotency
+            self._cache[id(edge)] = name
+        return self._cache[id(edge)]
diff --git a/fffw/graph/complex.py b/fffw/graph/complex.py
index 67f75e6..9ffe1c0 100644
--- a/fffw/graph/complex.py
+++ b/fffw/graph/complex.py
@@ -1,7 +1,6 @@
 import collections
-from typing import Dict
 
-from fffw.graph import base, inputs
+from fffw.graph import base, inputs, outputs
 
 __all__ = [
     'FilterComplex'
@@ -11,14 +10,15 @@ __all__ = [
 class FilterComplex:
     """ ffmpeg filter graph wrapper."""
 
-    def __init__(self, input_list: inputs.InputList):
+    def __init__(self, input_list: inputs.InputList,
+                 output_list: outputs.OutputList):
         """
         :param input_list: list of input files, containing video and audio
         streams.
+        :param output_list: list of output files, with codecs defined.
         """
         self.__input_list = input_list
-        self.__video_outputs: Dict[int, base.Dest] = {}
-        self.__audio_outputs: Dict[int, base.Dest] = {}
+        self.__output_list = output_list
 
     @property
     def video(self) -> base.Source:
@@ -38,42 +38,12 @@ class FilterComplex:
         :return: first stream of this kind not connected to filter graph
         """
         for stream in self.__input_list.streams:
-            if stream.kind != kind or stream.edge is not None:
+            if stream.kind != kind or stream.connected:
                 continue
             return stream
         else:
             raise RuntimeError("No free streams")
 
-    def get_video_dest(self, index: int = 0, create: bool = True) -> base.Dest:
-        """ Returns video output by index.
-        :param index: video output index.
-        :param create: create new video output flag
-        :return: output video stream
-        """
-        try:
-            return self.__video_outputs[index]
-        except KeyError:
-            if not create:
-                raise IndexError(index)
-            self.__video_outputs[index] = base.Dest(
-                'vout%s' % index, base.VIDEO)
-        return self.__video_outputs[index]
-
-    def get_audio_dest(self, index: int = 0, create: bool = True) -> base.Dest:
-        """ Returns audio output by index.
-        :param index: audio output index.
-        :param create: create new audio output flag
-        :return: output audio stream
-        """
-        try:
-            return self.__audio_outputs[index]
-        except KeyError:
-            if not create:
-                raise IndexError(index)
-            self.__audio_outputs[index] = base.Dest(
-                'aout%s' % index, base.AUDIO)
-        return self.__audio_outputs[index]
-
     def render(self, partial: bool = False) -> str:
         """
         Returns filter_graph description in corresponding ffmpeg param syntax.
diff --git a/fffw/graph/filters.py b/fffw/graph/filters.py
index 46960c9..c09fa11 100644
--- a/fffw/graph/filters.py
+++ b/fffw/graph/filters.py
@@ -1,58 +1,17 @@
-from typing import Optional, Tuple, Any
-
 from fffw.graph import base
-from fffw.graph.base import VIDEO, AUDIO
+from fffw.graph.base import VIDEO
 
 __all__ = [
-    'Pass',
-    'AudioPass',
-    'Deint',
     'Scale',
-    'SetSAR',
-    'Crop',
     'Split',
     'Concat',
-    'AudioConcat',
-    'Trim',
-    'AudioTrim',
-    'SetPTS',
-    'AudioSetPTS',
-    'AudioSplit',
     'Overlay',
-    'Volume',
-    'Rotate',
-    'Drawtext',
-    'HWUpload',
-    'ScaleNPP',
 ]
 
 
-class Pass(base.Node):
-    kind = VIDEO
-    enabled = False
-
-
-class AudioPass(base.Node):
-    kind = AUDIO
-    enabled = False
-
-
-class Deint(base.Node):
-    kind = VIDEO
-    name = 'yadif'
-
-    def __init__(self, mode: str = '0', enabled: bool = True):
-        super(Deint, self).__init__(enabled=enabled)
-        self.mode = mode
-
-    @property
-    def args(self) -> str:
-        return "%s" % self.mode
-
-
 class Scale(base.Node):
     kind = VIDEO
-    name = "scale"
+    filter = "scale"
 
     def __init__(self, width: int, height: int, enabled: bool = True):
         super(Scale, self).__init__(enabled=enabled)
@@ -64,51 +23,13 @@ class Scale(base.Node):
         return "%sx%s" % (self.width, self.height)
 
 
-class ScaleNPP(Scale):
-    name = 'scale_npp'
-
-    @property
-    def args(self) -> str:
-        return "w=%s:h=%s" % (self.width, self.height)
-
-
-class SetSAR(base.Node):
-    kind = VIDEO
-    name = "setsar"
-
-    def __init__(self, sar: float, enabled: bool = True):
-        super(SetSAR, self).__init__(enabled=enabled)
-        self.sar = sar
-
-    @property
-    def args(self) -> str:
-        return "%s" % self.sar
-
-
-class Crop(base.Node):
-    kind = VIDEO
-    name = "crop"
-
-    def __init__(self, width: int, height: int, left: int, top: int,
-                 enabled: bool = True):
-        super(Crop, self).__init__(enabled=enabled)
-        self.width = width
-        self.height = height
-        self.left = left
-        self.top = top
-
-    @property
-    def args(self) -> str:
-        return "%s:%s:%s:%s" % (self.width, self.height, self.left, self.top)
-
-
 class Split(base.Node):
-    kind = VIDEO
-    name = "split"
 
-    def __init__(self, output_count: int = 2):
+    def __init__(self, kind: base.StreamType = VIDEO, output_count: int = 2):
         enabled = output_count > 1
         self.output_count = output_count
+        self.kind = kind
+        self.filter = 'split' if kind == VIDEO else 'asplit'
         super(Split, self).__init__(enabled=enabled)
 
     @property
@@ -118,90 +39,28 @@ class Split(base.Node):
         return '%s' % self.output_count
 
 
-class AudioSplit(Split):
-    kind = AUDIO
-    name = "asplit"
-
-
-class HWUpload(base.Node):
-    kind = VIDEO
-    name = 'hwupload_cuda'
-
-
 class Concat(base.Node):
-    kind = VIDEO
-    name = 'concat'
+    filter = 'concat'
 
-    def __init__(self, input_count: int = 2):
+    def __init__(self, kind: base.StreamType = VIDEO, input_count: int = 2):
         enabled = input_count > 1
         self.input_count = input_count
+        self.kind = kind
         super(Concat, self).__init__(enabled=enabled)
 
     @property
     def args(self) -> str:
-        if self.input_count == 2:
-            return ''
-        return 'n=%s' % self.input_count
-
-
-class AudioConcat(base.Node):
-    kind = AUDIO
-    name = 'concat'
-
-    def __init__(self, input_count: int = 2):
-        enabled = input_count > 1
-        self.input_count = input_count
-        super(AudioConcat, self).__init__(enabled=enabled)
-
-    @property
-    def args(self) -> str:
+        if self.kind == VIDEO:
+            if self.input_count == 2:
+                return ''
+            return 'n=%s' % self.input_count
         return 'v=0:a=1:n=%s' % self.input_count
 
 
-class Trim(base.Node):
-    kind = VIDEO
-    name = 'trim'
-
-    def __init__(self,
-                 start: Optional[str] = None,
-                 end: Optional[str] = None,
-                 enabled: bool = True):
-        self.start = start
-        self.end = end
-        super(Trim, self).__init__(enabled=enabled)
-
-    @property
-    def args(self) -> str:
-        return 'start=%s:end=%s' % (self.start, self.end)
-
-
-class AudioTrim(Trim):
-    kind = AUDIO
-    name = 'atrim'
-
-
-class SetPTS(base.Node):
-    kind = VIDEO
-    name = 'setpts'
-
-    def __init__(self, mode: str = 'PTS-STARTPTS', enabled: bool = True):
-        self.mode = mode
-        super(SetPTS, self).__init__(enabled=enabled)
-
-    @property
-    def args(self) -> str:
-        return self.mode
-
-
-class AudioSetPTS(SetPTS):
-    kind = AUDIO
-    name = 'asetpts'
-
-
 class Overlay(base.Node):
     kind = VIDEO
     input_count = 2
-    name = "overlay"
+    filter = "overlay"
 
     def __init__(self, left: int, top: int, enabled: bool = True):
         super(Overlay, self).__init__(enabled=enabled)
@@ -213,55 +72,3 @@ class Overlay(base.Node):
         return "x=%s:y=%s" % (self.left, self.top)
 
 
-class Volume(base.Node):
-    kind = AUDIO
-    name = 'volume'
-
-    def __init__(self, volume: float, enabled: bool = True):
-        super(Volume, self).__init__(enabled=enabled)
-        self.volume = volume
-
-    @property
-    def args(self) -> str:
-        return "%.2f" % self.volume
-
-
-class Rotate(base.Node):
-    kind = VIDEO
-    name = "rotate"
-
-    def __init__(self, degrees: int, output_size: Tuple[int, int],
-                 enabled: bool = True):
-        super(Rotate, self).__init__(enabled=enabled)
-        self.degrees = degrees
-        self.output_size = output_size
-
-    @property
-    def args(self) -> str:
-        if self.degrees is not None:
-            result = "%s*PI/180" % self.degrees
-            if self.output_size:
-                w, h = self.output_size
-                result += ':ow=%s:oh=%s' % (w, h)
-            return result
-        else:
-            raise ValueError(self.degrees)
-
-
-class Drawtext(base.Node):
-    kind = VIDEO
-    name = 'drawtext'
-
-    def __init__(self,
-                 text: str,
-                 fontfile: str = 'font.ttf',
-                 x: str = '(w-text_w)/2',
-                 y: str = '(h-text_h)/2',
-                 enabled: bool = True,
-                 **opts: Any):
-        super(Drawtext, self).__init__(enabled=enabled)
-        self.opts = dict(opts, fontfile=fontfile, text=text, x=x, y=y)
-
-    @property
-    def args(self) -> str:
-        return ':'.join('%s=%s' % t for t in sorted(self.opts.items()))
diff --git a/fffw/graph/inputs.py b/fffw/graph/inputs.py
index 29451a2..22628d9 100644
--- a/fffw/graph/inputs.py
+++ b/fffw/graph/inputs.py
@@ -1,4 +1,4 @@
-from typing import Optional, List, Tuple, TypeVar, Type, cast, Any
+from typing import Optional, List, Tuple, cast, Any
 
 from fffw.graph import base
 from fffw.graph.meta import Meta
@@ -11,35 +11,12 @@ __all__ = [
     'Input',
 ]
 
-Obj = TypeVar('Obj')
-
-
-class Once:
-    """ Property that must be set exactly once."""
-
-    def __init__(self, attr_name: str) -> None:
-        """
-        :param attr_name: instance attribute name
-        """
-        self.attr_name = attr_name
-
-    def __get__(self, instance: Obj, owner: Type[Obj]) -> Any:
-        try:
-            return instance.__dict__[self.attr_name]
-        except KeyError:
-            raise RuntimeError(f"{self.attr_name} is not initialized")
-
-    def __set__(self, instance: Obj, value: Any) -> None:
-        if self.attr_name in instance.__dict__:
-            raise RuntimeError(f"{self.attr_name} already initialized")
-        instance.__dict__[self.attr_name] = value
-
 
 class Stream(base.Source):
     """ Video or audio stream in input file."""
-    source = cast("Input", Once('source'))
+    source = cast("Input", base.Once('source'))
     """ Source file that contains current stream."""
-    index = cast(int, Once('index'))
+    index = cast(int, base.Once('index'))
     """ Index of current stream in source file."""
 
     def __init__(self, kind: base.StreamType, meta: Optional[Meta] = None):
@@ -47,7 +24,7 @@ class Stream(base.Source):
         :param kind: stream kind, video or audio
         :param meta: stream metadata
         """
-        super().__init__(name=None, kind=kind, meta=meta)
+        super().__init__(kind=kind, meta=meta)
 
     @property
     def name(self) -> str:
@@ -61,9 +38,9 @@ class Input(BaseWrapper):
     Input command line params generator for FFMPEG.
     """
     """ Filename or url, value for `-i` argument."""
-    streams = cast(Tuple[Stream, ...], Once("streams"))
+    streams = cast(Tuple[Stream, ...], base.Once("streams"))
     """ List of audio and video streams for input file."""
-    index = cast(int, Once("index"))
+    index = cast(int, base.Once("index"))
     """ Internal ffmpeg source file index."""
 
     def __init__(self, *streams: Stream, input_file: str = ''):
@@ -92,8 +69,9 @@ class Input(BaseWrapper):
                 raise ValueError(stream.kind)
             stream.source = self
 
-    def get_args(self) -> List[bytes]:
-        return ensure_binary(["-i", self.input_file])
+    @ensure_binary
+    def get_args(self) -> List[Any]:
+        return ["-i", self.input_file]
 
 
 class InputList:
@@ -139,7 +117,7 @@ class InputList:
         self.__inputs.extend(sources)
 
     def get_args(self) -> List[bytes]:
-        result = []
+        result: List[bytes] = []
         for source in self.__inputs:
             result.extend(source.get_args())
         return result
diff --git a/fffw/graph/meta.py b/fffw/graph/meta.py
index adba8f9..d62b8a3 100644
--- a/fffw/graph/meta.py
+++ b/fffw/graph/meta.py
@@ -4,6 +4,15 @@ from typing import List, Union
 
 from pymediainfo import MediaInfo  # type: ignore
 
+__all__ = [
+    'Meta',
+    'VideoMeta',
+    'AudioMeta',
+    'video_meta_data',
+    'audio_meta_data',
+    'TS'
+]
+
 
 class TS(timedelta):
     """
diff --git a/fffw/graph/outputs.py b/fffw/graph/outputs.py
new file mode 100644
index 0000000..2977aa4
--- /dev/null
+++ b/fffw/graph/outputs.py
@@ -0,0 +1,164 @@
+import os
+from itertools import chain
+from typing import List, Tuple, cast, Optional, Any
+
+from fffw.graph import base
+from fffw.wrapper import BaseWrapper, ensure_binary
+
+__all__ = [
+    'Codec',
+    'Output',
+    'OutputList'
+]
+
+
+class Codec(base.Dest, BaseWrapper):
+    index = cast(int, base.Once('index'))
+    """ Index of current codec in ffmpeg output streams."""
+    # TODO #9: implement single argument with stream type modifier
+    arguments = [
+        ('vbitrate', '-b:v '),
+        ('abitrate', '-b:a '),
+    ]
+
+    def __init__(self, kind: base.StreamType, codec: str = None,
+                 **kwargs: Any) -> None:
+        super().__init__(kind)
+        BaseWrapper.__init__(self, **kwargs)
+        self._codec = codec
+
+    @property
+    def map(self) -> Optional[str]:
+        """
+        :returns: `-map` argument value depending of a node or a source
+        connected to codec.
+        """
+        if self.edge is None:
+            raise RuntimeError("Codec not connected to source")
+        source = self.edge.input
+        # Source input has name generated from input file index, stream
+        # specifier and stream index. Node has no attribute index, so we use
+        # Dest name ("[vout0]") as map argument. See also `Node.get_filter_args`
+        return getattr(source, 'name', self.name)
+
+    @property
+    def connected(self) -> bool:
+        """
+        :return: True if codec is already connected to a node or a source.
+        """
+        return bool(self.edge)
+
+    def get_args(self) -> List[bytes]:
+        args = ['-map', self.map]
+        if self._codec:
+            args.extend([f'-c:{self.kind.value}', self._codec])
+        return ensure_binary(args) + super().get_args()
+
+
+class Output(BaseWrapper):
+    arguments = [
+        ('format', '-f '),
+    ]
+
+    def __init__(self, output_file: str, *codecs: Codec, **kwargs: Any) -> None:
+        ext = os.path.splitext(output_file)[-1].lstrip('.')
+        kwargs.setdefault('format', ext)
+        super().__init__(**kwargs)
+        self._output_file = output_file
+        self._codecs = list(codecs)
+
+    def __lt__(self, other: base.InputType) -> Codec:
+        """
+        Connects a source or a filter to a first free codec.
+
+        If there is no free codecs, new codec stub is created.
+        """
+        codec = self.get_free_codec(other.kind)
+        other.connect_dest(codec)
+        return codec
+
+    @property
+    def codecs(self) -> Tuple[Codec, ...]:
+        return tuple(self._codecs)
+
+    @property
+    def video(self) -> Codec:
+        return self.get_free_codec(base.VIDEO)
+
+    @property
+    def audio(self) -> Codec:
+        return self.get_free_codec(base.AUDIO)
+
+    def get_free_codec(self, kind: base.StreamType) -> Codec:
+        try:
+            codec = next(filter(lambda c: not c.connected, self._codecs))
+        except StopIteration:
+            codec = Codec(kind)
+            self._codecs.append(codec)
+        return codec
+
+    def get_args(self) -> List[bytes]:
+        args = (
+                list(chain(*(codec.get_args() for codec in self._codecs))) +
+                super().get_args() +
+                ensure_binary([self._output_file])
+        )
+        return args
+
+
+class OutputList:
+    """ Supports unique output streams names generation."""
+
+    def __init__(self, *outputs: Output) -> None:
+        """
+        :param outputs: list of output files
+        """
+        self.__outputs: List[Output] = []
+        self.__video_index = 0
+        self.__audio_index = 0
+        self.extend(*outputs)
+
+    @property
+    def outputs(self) -> Tuple[Output, ...]:
+        return tuple(self.__outputs)
+
+    @property
+    def codecs(self) -> List[Codec]:
+        result: List[Codec] = []
+        for output in self.__outputs:
+            result.extend(output.codecs)
+        return result
+
+    def append(self, output: Output) -> None:
+        """
+        Adds new output file to output list.
+
+        :param output: output file
+        """
+        for codec in output.codecs:
+            self.__set_index(codec)
+        self.__outputs.append(output)
+
+    def extend(self, *outputs: Output) -> None:
+        """
+        Adds multiple output files to output list.
+
+        :param outputs: list of output files
+        """
+        for codec in chain(*map(lambda output: output.codecs, outputs)):
+            self.__set_index(codec)
+        self.__outputs.extend(outputs)
+
+    def get_args(self) -> List[bytes]:
+        result = []
+        for source in self.__outputs:
+            result.extend(source.get_args())
+        return result
+
+    def __set_index(self, codec: Codec) -> None:
+        if codec.kind == base.VIDEO:
+            codec.index = self.__video_index
+            self.__video_index += 1
+        else:
+            codec.index = self.__audio_index
+            self.__audio_index += 1
diff --git a/fffw/wrapper.py b/fffw/wrapper.py
index fb8f192..328495a 100644
--- a/fffw/wrapper.py
+++ b/fffw/wrapper.py
@@ -1,7 +1,8 @@
 import re
 import subprocess
+from functools import wraps
 from logging import getLogger
-from typing import Tuple, List, Any, Dict, Union, overload, IO, cast
+from typing import Tuple, List, Any, Dict, Union, overload, IO, cast, Callable
 
 
 def quote(token: Any) -> str:
@@ -27,14 +28,45 @@ def ensure_binary(x: Union[int, float, str, bytes, None]) -> bytes:
     ...
 
 
+@overload
+def ensure_binary(x: Callable[..., Tuple[Any]]
+                  ) -> Callable[..., Tuple[bytes]]:
+    ...
+
+
+@overload
+def ensure_binary(x: Callable[..., List[Any]]
+                  ) -> Callable[..., List[bytes]]:
+    ...
+
+
+@overload
+def ensure_binary(x: Callable[..., Union[int, float, str, bytes, None]]
+                  ) -> Callable[..., bytes]:
+    ...
+
+
 def ensure_binary(x: Any) -> Any:
-    """ Recursively ensures that all values except collections are bytes."""
+    """ Recursively ensures that all values except collections are bytes.
+
+    * tuples and lists are encoded recursively
+    * strings are encoded with utf-8
+    * bytes are left intact
+    * functions are decorated with ensuring binary results
+    * all other types are converted to string and encoded
+    """
     if isinstance(x, tuple):
         return tuple(ensure_binary(y) for y in x)
     if isinstance(x, list):
         return list(ensure_binary(y) for y in x)
     if isinstance(x, str):
         return x.encode("utf-8")
+    if callable(x):
+        @wraps(x)
+        def wrapper(*args: Any, **kwargs: Any) -> Any:
+            return ensure_binary(x(*args, **kwargs))
+
+        return wrapper
     if not isinstance(x, bytes):
         return str(x).encode("utf-8")
     return x
@@ -102,7 +134,8 @@ class BaseWrapper:
         else:
             object.__setattr__(self, key, value)
 
-    def get_args(self) -> List[bytes]:
+    @ensure_binary
+    def get_args(self) -> List[Any]:
         result = []
         for k in self._args_order:
             v = self._args[k]
@@ -125,7 +158,7 @@ class BaseWrapper:
                     else:
                         result.append("%s%s" % (param, v))
 
-        return list(map(ensure_binary, result))
+        return result
 
     def set_args(self, **kwargs: Any) -> None:
         for k, v in kwargs.items():


Test patch:
diff --git a/tests/test_ffmpeg.py b/tests/test_ffmpeg.py
index b4ea39b..1ca86b3 100644
--- a/tests/test_ffmpeg.py
+++ b/tests/test_ffmpeg.py
@@ -1,10 +1,37 @@
-from unittest import TestCase
+from unittest import TestCase, expectedFailure
 
 from fffw.encoding import *
 from fffw.graph import *
+from fffw.graph import base, VIDEO, AUDIO
 from fffw.wrapper import ensure_binary
 
 
+class SetSAR(base.Node):
+    kind = VIDEO
+    filter = "setsar"
+
+    def __init__(self, sar: float, enabled: bool = True):
+        super(SetSAR, self).__init__(enabled=enabled)
+        self.sar = sar
+
+    @property
+    def args(self) -> str:
+        return "%s" % self.sar
+
+
+class Volume(base.Node):
+    kind = AUDIO
+    filter = 'volume'
+
+    def __init__(self, volume: float, enabled: bool = True):
+        super(Volume, self).__init__(enabled=enabled)
+        self.volume = volume
+
+    @property
+    def args(self) -> str:
+        return "%.2f" % self.volume
+
+
 class FFMPEGTestCase(TestCase):
 
     def test_ffmpeg(self):
@@ -12,24 +39,24 @@ class FFMPEGTestCase(TestCase):
         ff = FFMPEG()
         ff < Input(input_file='/tmp/input.mp4')
 
-        fc = ff.init_filter_complex()
+        cv0 = Codec(VIDEO, 'libx264', vbitrate='700000')
+        ca0 = Codec(AUDIO, 'aac', abitrate='128000')
+        ca1 = Codec(AUDIO, 'libmp3lame', abitrate='394000')
 
-        asplit = fc.audio | AudioSplit()
+        fc = ff.init_filter_complex()
 
-        fc.video | Scale(640, 360) | fc.get_video_dest(0)
+        asplit = fc.audio | Split(AUDIO)
 
-        asplit.connect_dest(fc.get_audio_dest(0))
-        asplit.connect_dest(fc.get_audio_dest(1))
+        fc.video | Scale(640, 360) > cv0
 
-        out0 = FLVMuxer('/tmp/out.flv')
-        out1 = MP3Muxer('/tmp/out.mp3')
+        asplit.connect_dest(ca0)
+        asplit.connect_dest(ca1)
 
-        cv0 = VideoCodec(vcodec='libx264', vbitrate='700000', size='640x360')
-        ca0 = AudioCodec(acodec='aac', abitrate='128000')
-        ca1 = AudioCodec(acodec='libmp3lame', abitrate='394000')
+        out0 = Output('/tmp/out.flv', cv0, ca0)
+        out1 = Output('/tmp/out.mp3', ca1)
 
-        ff.add_output(out0, cv0, ca0)
-        ff.add_output(out1, ca1)
+        ff.add_output(out0)
+        ff.add_output(out1)
 
         expected = [
             'ffmpeg',
@@ -52,14 +79,13 @@ class FFMPEGTestCase(TestCase):
     def test_bypass_with_filter_complex(self):
         """ Audio stream bypass mode."""
         ff = FFMPEG('/tmp/input.mp4')
+        cv0 = Codec(VIDEO, 'libx264', vbitrate='700000', size='640x360')
+        ca0 = Codec(AUDIO, 'aac', abitrate='128000')
 
         fc = ff.init_filter_complex()
-        fc.video | Scale(640, 360) | fc.get_video_dest(0)
+        fc.video | Scale(640, 360) > cv0
 
-        cv0 = VideoCodec(vcodec='libx264', vbitrate='700000', size='640x360')
-        ca0 = AudioCodec(acodec='aac', abitrate='128000')
-        out0 = FLVMuxer('/tmp/out.flv')
-        ff.add_output(out0, cv0, ca0)
+        ff > Output('/tmp/out.flv', cv0, ca0)
 
         expected = [
             'ffmpeg',
@@ -77,10 +103,9 @@ class FFMPEGTestCase(TestCase):
         """ Stream bypass with filter_complex missing."""
         ff = FFMPEG('/tmp/input.mp4')
 
-        cv0 = VideoCodec(vcodec='libx264', vbitrate='700000', size='640x360')
-        ca0 = AudioCodec(acodec='aac', abitrate='128000')
-        out0 = FLVMuxer('/tmp/out.flv')
-        ff.add_output(out0, cv0, ca0)
+        cv0 = Codec(VIDEO, 'libx264', vbitrate='700000')
+        ca0 = Codec(AUDIO, 'aac', abitrate='128000')
+        ff > Output('/tmp/out.flv', cv0, ca0)
 
         expected = [
             'ffmpeg',
@@ -98,17 +123,19 @@ class FFMPEGTestCase(TestCase):
         ff = FFMPEG()
         ff < Input(Stream(VIDEO), input_file='/tmp/input.jpg')
         ff < Input(input_file='/tmp/input.mp4')
+
+        cv0 = Codec(VIDEO, 'libx264', vbitrate='700000')
+        ca0 = Codec(AUDIO, 'aac', abitrate='128000')
+
         fc = ff.init_filter_complex()
         overlay = Overlay(0, 0)
         fc.video | Scale(640, 360) | overlay
         fc.video | Scale(1280, 720) | overlay
-        fc.audio | Volume(-20) | fc.get_audio_dest(0)
-        overlay | fc.get_video_dest(0)
+        fc.audio | Volume(-20) > ca0
+        overlay > cv0
 
-        cv0 = VideoCodec(vcodec='libx264', vbitrate='700000', size='640x360')
-        ca0 = AudioCodec(acodec='aac', abitrate='128000')
-        out0 = FLVMuxer('/tmp/out.flv')
-        ff.add_output(out0, cv0, ca0)
+        out0 = Output('/tmp/out.flv', cv0, ca0)
+        ff.add_output(out0)
 
         expected = [
             'ffmpeg',
@@ -133,12 +160,13 @@ class FFMPEGTestCase(TestCase):
 
         fc = ff.init_filter_complex()
 
-        fc.audio | Volume(20) | fc.get_audio_dest(0)
+        cv0 = Codec(VIDEO, 'copy')
+        ca0 = Codec(AUDIO, 'aac', abitrate='128000')
 
-        cv0 = VideoCodec(vcodec='copy')
-        ca0 = AudioCodec(acodec='aac', abitrate='128000')
-        out0 = FLVMuxer('/tmp/out.flv')
-        ff.add_output(out0, cv0, ca0)
+        fc.audio | Volume(20) > ca0
+
+        out0 = Output('/tmp/out.flv', cv0, ca0)
+        ff.add_output(out0)
         expected = [
             'ffmpeg',
             '-i', '/tmp/input.mp4',
@@ -155,16 +183,20 @@ class FFMPEGTestCase(TestCase):
 
     def test_reuse_input_files(self):
         """ Reuse input files multiple times."""
-        ff = FFMPEG('/tmp/input.mp4')
-        cv0 = VideoCodec(map='0:v', vcodec='copy')
-        ca0 = AudioCodec(map='0:a', acodec='copy')
-        out0 = FLVMuxer('/tmp/out0.flv')
-        ff.add_output(out0, cv0, ca0)
-
-        cv1 = VideoCodec(map='0:v', vcodec='copy')
-        ca1 = AudioCodec(map='0:a', acodec='copy')
-        out1 = FLVMuxer('/tmp/out1.flv')
-        ff.add_output(out1, cv1, ca1)
+        v = Stream(VIDEO)
+        a = Stream(AUDIO)
+        ff = FFMPEG(Input(v, a, input_file='/tmp/input.mp4'))
+        cv0 = Codec(VIDEO, 'copy')
+        ca0 = Codec(AUDIO, 'copy')
+        out0 = Output('/tmp/out0.flv', cv0, ca0)
+        ff > out0
+
+        cv1 = Codec(VIDEO, 'copy')
+        ca1 = Codec(AUDIO, 'copy')
+        out1 = Output('/tmp/out1.flv', cv1, ca1)
+        v > cv1
+        a > ca1
+        ff > out1
         expected = [
             'ffmpeg',
             '-i', '/tmp/input.mp4',
@@ -185,21 +217,25 @@ class FFMPEGTestCase(TestCase):
 
     def test_handle_codec_copy_with_other_filters(self):
         """ vcodec=copy with separate transcoded output."""
-        ff = FFMPEG('/tmp/input.mp4')
+        v = Stream(VIDEO)
+        a = Stream(AUDIO)
+        ff = FFMPEG(Input(v, a, input_file='/tmp/input.mp4'))
+        ff.init_filter_complex()
 
-        fc = ff.init_filter_complex()
+        cv0 = Codec(VIDEO, 'copy')
+        ca0 = Codec(AUDIO, 'copy')
+        out0 = Output('/tmp/copy.flv', cv0, ca0)
 
-        fc.video | Scale(640, 360) | fc.get_video_dest(0)
+        ff > out0
 
-        cv0 = VideoCodec(map='0:v', vcodec='copy')
-        ca0 = AudioCodec(map='0:a', acodec='copy')
-        out0 = FLVMuxer('/tmp/copy.flv')
-        ff.add_output(out0, cv0, ca0)
+        cv1 = Codec(VIDEO, 'libx264')
+        ca1 = Codec(AUDIO, 'aac')
+        out1 = Output('/tmp/out.flv', cv1, ca1)
 
-        cv1 = VideoCodec(vcodec='libx264')
-        ca1 = AudioCodec(acodec='aac')
-        out1 = FLVMuxer('/tmp/out.flv')
-        ff.add_output(out1, cv1, ca1)
+        v | Scale(640, 360) > cv1
+        a > ca1
+
+        ff.add_output(out1)
 
         expected = [
             'ffmpeg',
@@ -220,13 +256,13 @@ class FFMPEGTestCase(TestCase):
             '/tmp/out.flv'
 
         ]
-        self.assertEqual(ff.get_args(), ensure_binary(expected))
+        self.assertEqual(ensure_binary(expected), ff.get_args())
 
     def test_transcoding_without_graph(self):
         """ Transcoding works without filter graph."""
         ff = FFMPEG()
         ff < Input(input_file='input.mp4')
-        ff.add_output(NullMuxer('/dev/null'))
+        ff.add_output(Output('/dev/null', format='null'))
         expected = [
             'ffmpeg',
             '-i', 'input.mp4',
@@ -235,36 +271,14 @@ class FFMPEGTestCase(TestCase):
         ]
         self.assertEqual(ff.get_args(), ensure_binary(expected))
 
-    def test_hls_muxer(self):
-        """ hls muxer args."""
-        ff = FFMPEG('/tmp/input.mp4')
-
-        cv0 = VideoCodec(vcodec='libx264')
-        ca0 = AudioCodec(acodec='aac')
-        out0 = HLSMuxer('http://ya.ru/out.m3u8', segment_size=4,
-                        manifest_size=5, method='PUT')
-        ff.add_output(out0, cv0, ca0)
-        expected = [
-            'ffmpeg',
-            '-i', '/tmp/input.mp4',
-            '-map', '0:v',
-            '-c:v', 'libx264',
-            '-map', '0:a',
-            '-c:a', 'aac',
-            '-f', 'hls',
-            '-method', 'PUT',
-            '-hls_time', '4',
-            '-hls_list_size', '5',
-            'http://ya.ru/out.m3u8'
-        ]
-        self.assertEqual(ff.get_args(), ensure_binary(expected))
-
+    # TODO #19 reimplement TeeMuxer
+    @expectedFailure
     def test_tee_muxer(self):
         """ tee muxer args."""
         ff = FFMPEG('/tmp/input.mp4')
 
-        cv0 = VideoCodec(vcodec='libx264')
-        ca0 = AudioCodec(acodec='aac')
+        cv0 = Codec(VIDEO, 'libx264')
+        ca0 = Codec(AUDIO, 'aac')
         out0 = HLSMuxer('http://ya.ru/1.m3u8', segment_size=2)
 
         out1 = HLSMuxer('http://ya.ru/2.m3u8', manifest_size=5)
@@ -289,6 +303,9 @@ class FFMPEGTestCase(TestCase):
         ff < Input(input_file='preroll.mp4')
         ff < Input(input_file='input.mp4')
 
+        cv0 = Codec(VIDEO, 'libx264')
+        ca0 = Codec(AUDIO, 'aac')
+
         fc = ff.init_filter_complex()
 
         preroll_ready = fc.video | Scale(640, 480) | SetSAR(1)
@@ -296,14 +313,14 @@ class FFMPEGTestCase(TestCase):
         preroll_ready | concat
         fc.video | concat
 
-        concat | fc.get_video_dest(0)
+        concat > cv0
 
-        aconcat = AudioConcat()
+        aconcat = Concat(AUDIO)
         fc.audio | aconcat
         fc.audio | aconcat
 
-        aconcat | fc.get_audio_dest(0)
-        ff.add_output(MP4Muxer('output.mp4'))
+        aconcat > ca0
+        ff.add_output(Output('output.mp4', cv0, ca0))
 
         expected = [
             'ffmpeg',
@@ -314,6 +331,10 @@ class FFMPEGTestCase(TestCase):
             "[v:scale0]setsar=1[v:setsar0];"
             "[v:setsar0][1:v]concat[vout0];"
             "[0:a][1:a]concat=v=0:a=1:n=2[aout0]",
+            '-map', '[vout0]',
+            '-c:v', 'libx264',
+            '-map', '[aout0]',
+            '-c:a', 'aac',
             '-f', 'mp4',
             'output.mp4'
         ]
diff --git a/tests/test_graph.py b/tests/test_graph.py
index fa845e5..189b042 100644
--- a/tests/test_graph.py
+++ b/tests/test_graph.py
@@ -1,7 +1,20 @@
 from unittest import TestCase
 
 from fffw.graph import *
-from fffw.graph import meta, inputs
+from fffw.graph import base, VIDEO
+
+
+class Deint(base.Node):
+    kind = VIDEO
+    filter = 'yadif'
+
+    def __init__(self, mode: str = '0', enabled: bool = True):
+        super(Deint, self).__init__(enabled=enabled)
+        self.mode = mode
+
+    @property
+    def args(self) -> str:
+        return "%s" % self.mode
 
 
 class FilterGraphTestCase(TestCase):
@@ -15,11 +28,14 @@ class FilterGraphTestCase(TestCase):
                                             |
                                             ------<Scale>--[O/720p]
         """
-        logo = inputs.Input(inputs.Stream(VIDEO), input_file='logo.png')
-        main = inputs.Input(input_file='main.mp4')
-        il = inputs.InputList(main, logo)
+        logo = Input(Stream(VIDEO), input_file='logo.png')
+        main = Input(input_file='main.mp4')
+        il = InputList(main, logo)
+        out0 = Output('out0.mp4')
+        out1 = Output('out1.mp4')
+        ol = OutputList(out0, out1)
 
-        fc = FilterComplex(il)
+        fc = FilterComplex(il, ol)
 
         deint = Deint(enabled=False)  # deinterlace is disabled
 
@@ -39,25 +55,25 @@ class FilterGraphTestCase(TestCase):
         next_node = fc.video | Scale(logo_width, logo_height) | over
 
         # audio is split to two streams
-        asplit = fc.audio | AudioSplit(2)
+        asplit = fc.audio | Split(AUDIO)
 
-        for i in range(2):
-            asplit.connect_dest(fc.get_audio_dest(i))
+        for out in ol.outputs:
+            asplit > out
 
         # video split to two steams
 
         # connect split filter to overlayed video stream
-        split = next_node | Split(2)
+        split = next_node | Split()
 
         # intermediate video stream scaling
         sizes = [(640, 480), (1280, 720)]
 
-        for i, size in enumerate(sizes):
+        for out, size in zip(ol.outputs, sizes):
             # add scale filters to video streams
             w, h = size
             scale = Scale(w, h, enabled=True)
             # connect scaled streams to video destinations
-            split | scale | fc.get_video_dest(i)
+            split | scale > out
 
         result = fc.render()
 
@@ -82,61 +98,64 @@ class FilterGraphTestCase(TestCase):
 
     def test_disabled_filters(self):
         """ Filter skipping."""
-        source = inputs.Input(inputs.Stream(VIDEO), input_file="input.mp4")
-        fc = FilterComplex(inputs.InputList(source))
 
-        dest = fc.get_video_dest(0)
-        fc.video | Scale(640, 360) | Deint(enabled=False) | dest
-        self.assertEqual(fc.render(), '[0:v]scale=640x360[vout0]')
+        # noinspection PyShadowingNames
+        def fc_factory():
+            src = Input(Stream(VIDEO), input_file="input.mp4")
+            dst = Output('output.mp4')
+            fc = FilterComplex(InputList(src), OutputList(dst))
+            return fc, dst
+
+        fc, dst = fc_factory()
+
+        fc.video | Scale(640, 360) | Deint(enabled=False) > dst.video
+        self.assertEqual('[0:v]scale=640x360[vout0]', fc.render())
 
-        source = inputs.Input(inputs.Stream(VIDEO), input_file="input.mp4")
-        fc = FilterComplex(inputs.InputList(source))
+        fc, dst = fc_factory()
 
-        dest = fc.get_video_dest(0)
-        fc.video | Deint(enabled=False) | Scale(640, 360) | dest
-        self.assertEqual(fc.render(), '[0:v]scale=640x360[vout0]')
+        fc.video | Deint(enabled=False) | Scale(640, 360) > dst.video
+        self.assertEqual('[0:v]scale=640x360[vout0]', fc.render())
 
-        source = inputs.Input(inputs.Stream(VIDEO), input_file="input.mp4")
-        fc = FilterComplex(inputs.InputList(source))
+        fc, dst = fc_factory()
 
-        dest = fc.get_video_dest(0)
         tmp = fc.video | Deint(enabled=False)
         tmp = tmp | Deint(enabled=False)
-        tmp | Scale(640, 360) | dest
-        self.assertEqual(fc.render(), '[0:v]scale=640x360[vout0]')
+        tmp | Scale(640, 360) > dst.video
+        self.assertEqual('[0:v]scale=640x360[vout0]', fc.render())
 
-        source = inputs.Input(inputs.Stream(VIDEO), input_file="input.mp4")
-        fc = FilterComplex(inputs.InputList(source))
+        fc, dst = fc_factory()
 
-        dest = fc.get_video_dest(0)
         tmp = fc.video | Scale(640, 360)
         tmp = tmp | Deint(enabled=False)
-        tmp | Deint(enabled=False) | dest
-        self.assertEqual(fc.render(), '[0:v]scale=640x360[vout0]')
+        tmp | Deint(enabled=False) > dst.video
+        self.assertEqual('[0:v]scale=640x360[vout0]', fc.render())
 
     def test_skip_not_connected_sources(self):
         """ Skip unused sources in filter complex.
         """
-        source = inputs.Input(input_file='input.mp4')
-        il = inputs.InputList(source)
+        source = Input(input_file='input.mp4')
+        output = Output('output.mp4')
+        il = InputList(source)
+        ol = OutputList(output)
         # passing only video to FilterComplex
-        fc = FilterComplex(il)
-        dest = fc.get_video_dest(0)
-        fc.video | Scale(640, 360) | dest
+        fc = FilterComplex(il, ol)
+        fc.video | Scale(640, 360) > output
 
-        self.assertEqual(fc.render(), '[0:v]scale=640x360[vout0]')
+        self.assertEqual('[0:v]scale=640x360[vout0]', fc.render())
 
     def test_pass_metadata(self):
         """
         stream metadata is passed from source to destination
         """
-        metadata = meta.video_meta_data()
-
-        source = inputs.Input(inputs.Stream(VIDEO, meta=metadata),
-                              input_file='input.mp4')
-        il = inputs.InputList(source)
-        fc = FilterComplex(il)
-        dest = fc.get_video_dest(0)
-        fc.video | Scale(640, 360) | dest
+        metadata = video_meta_data()
+
+        source = Input(Stream(VIDEO, meta=metadata),
+                       input_file='input.mp4')
+        output = Output('output.mp4')
+        il = InputList(source)
+        ol = OutputList(output)
+        fc = FilterComplex(il, ol)
+        dest = output.video
+        fc.video | Scale(640, 360) > dest
 
         self.assertIs(dest.get_meta_data(dest), metadata)