commit de958344b94af4ff6de35b07466355a49ed20b92
Author: wanglei.w <wanglei.w@bytedance.com>
Date:   Tue Nov 10 15:49:38 2020 +0800

    init with test

diff --git a/README b/README
new file mode 100644
index 0000000..904a2e8
--- /dev/null
+++ b/README
@@ -0,0 +1,5 @@
+# ffmpeg-go
+
+ffmpeg-go is golang port of https://github.com/kkroening/ffmpeg-python
+
+check ffmpeg_test.go for examples.
\ No newline at end of file
diff --git a/dag.go b/dag.go
new file mode 100644
index 0000000..24e0877
--- /dev/null
+++ b/dag.go
@@ -0,0 +1,152 @@
+package ffmpeg_go
+
+import (
+	"errors"
+)
+
+// Node in a directed-acyclic graph (DAG).
+//
+// Edges:
+// DagNodes are connected by edges.  An edge connects two nodes with a label for each side:
+// - ``upstream_node``: upstream/parent node
+// - ``upstream_label``: label on the outgoing side of the upstream node
+// - ``downstream_node``: downstream/child node
+// - ``downstream_label``: label on the incoming side of the downstream node
+//
+// For example, DagNode A may be connected to DagNode B with an edge labelled "foo" on A's side, and "bar" on B's
+// side:
+//
+// _____               _____
+// |     |             |     |
+// |  A  >[foo]---[bar]>  B  |
+// |_____|             |_____|
+//
+// Edge labels may be integers or strings, and nodes cannot have more than one incoming edge with the same label.
+//
+// DagNodes may have any number of incoming edges and any number of outgoing edges.  DagNodes keep track only of
+// their incoming edges, but the entire graph structure can be inferred by looking at the furthest downstream
+// nodes and working backwards.
+//
+// Hashing:
+// DagNodes must be hashable, and two nodes are considered to be equivalent if they have the same hash value.
+//
+// Nodes are immutable, and the hash should remain constant as a result.  If a node with new contents is required,
+// create a new node and throw the old one away.
+//
+// String representation:
+// In order for graph visualization tools to show useful information, nodes must be representable as strings.  The
+// ``repr`` operator should provide a more or less "full" representation of the node, and the ``short_repr``
+// property should be a shortened, concise representation.
+//
+// Again, because nodes are immutable, the string representations should remain constant.
+type DagNode interface {
+	Hash() int
+	// Compare two nodes
+	Equal(other DagNode) bool
+	// Return a full string representation of the node.
+	String() string
+	// Return a partial/concise representation of the node
+	ShortRepr() string
+	// Provides information about all incoming edges that connect to this node.
+	//
+	//        The edge map is a dictionary that maps an ``incoming_label`` to ``(outgoing_node, outgoing_label)``.  Note that
+	//        implicity, ``incoming_node`` is ``self``.  See "Edges" section above.
+	IncomingEdgeMap() map[Label]NodeInfo
+}
+
+type Label string
+type NodeInfo struct {
+	Node     DagNode
+	Label    Label
+	Selector Selector
+}
+type Selector string
+
+type DagEdge struct {
+	DownStreamNode   DagNode
+	DownStreamLabel  Label
+	UpStreamNode     DagNode
+	UpStreamLabel    Label
+	UpStreamSelector Selector
+}
+
+func GetInComingEdges(downStreamNode DagNode, inComingEdgeMap map[Label]NodeInfo) []DagEdge {
+	var edges []DagEdge
+	for _, downStreamLabel := range _getAllLabelsInSorted(inComingEdgeMap) {
+		upStreamInfo := inComingEdgeMap[downStreamLabel]
+		edges = append(edges, DagEdge{
+			DownStreamNode:   downStreamNode,
+			DownStreamLabel:  downStreamLabel,
+			UpStreamNode:     upStreamInfo.Node,
+			UpStreamLabel:    upStreamInfo.Label,
+			UpStreamSelector: upStreamInfo.Selector,
+		})
+	}
+	return edges
+}
+
+func GetOutGoingEdges(upStreamNode DagNode, outOutingEdgeMap map[Label][]NodeInfo) []DagEdge {
+	var edges []DagEdge
+	for _, upStreamLabel := range _getAllLabelsSorted(outOutingEdgeMap) {
+		downStreamInfos := outOutingEdgeMap[upStreamLabel]
+		for _, downStreamInfo := range downStreamInfos {
+			edges = append(edges, DagEdge{
+				DownStreamNode:   downStreamInfo.Node,
+				DownStreamLabel:  downStreamInfo.Label,
+				UpStreamNode:     upStreamNode,
+				UpStreamLabel:    upStreamLabel,
+				UpStreamSelector: downStreamInfo.Selector,
+			})
+		}
+
+	}
+	return edges
+}
+
+func TopSort(downStreamNodes []DagNode) (sortedNodes []DagNode, outOutingEdgeMaps map[int]map[Label][]NodeInfo, err error) {
+	markedNodes := map[int]struct{}{}
+	markedSortedNodes := map[int]struct{}{}
+	outOutingEdgeMaps = map[int]map[Label][]NodeInfo{}
+
+	var visit func(upStreamNode, downstreamNode DagNode, upStreamLabel, downStreamLabel Label, downStreamSelector Selector) error
+	visit = func(upStreamNode, downstreamNode DagNode, upStreamLabel, downStreamLabel Label, downStreamSelector Selector) error {
+		if _, ok := markedNodes[upStreamNode.Hash()]; ok {
+			return errors.New("graph if not DAG")
+		}
+		if downstreamNode != nil {
+			if a, ok := outOutingEdgeMaps[upStreamNode.Hash()]; !ok || a == nil {
+				outOutingEdgeMaps[upStreamNode.Hash()] = map[Label][]NodeInfo{}
+			}
+			outgoingEdgeMap := outOutingEdgeMaps[upStreamNode.Hash()]
+			outgoingEdgeMap[upStreamLabel] = append(outgoingEdgeMap[upStreamLabel], NodeInfo{
+				Node:     downstreamNode,
+				Label:    downStreamLabel,
+				Selector: downStreamSelector,
+			})
+		}
+
+		if _, ok := markedSortedNodes[upStreamNode.Hash()]; !ok {
+			markedNodes[upStreamNode.Hash()] = struct{}{}
+			for _, edge := range GetInComingEdges(upStreamNode, upStreamNode.IncomingEdgeMap()) {
+				err := visit(edge.UpStreamNode, edge.DownStreamNode, edge.UpStreamLabel, edge.DownStreamLabel, edge.UpStreamSelector)
+				if err != nil {
+					return err
+				}
+			}
+			delete(markedNodes, upStreamNode.Hash())
+			sortedNodes = append(sortedNodes, upStreamNode)
+			markedSortedNodes[upStreamNode.Hash()] = struct{}{}
+		}
+		return nil
+	}
+
+	for len(downStreamNodes) > 0 {
+		node := downStreamNodes[len(downStreamNodes)-1]
+		downStreamNodes = downStreamNodes[:len(downStreamNodes)-1]
+		err = visit(node, nil, "", "", "")
+		if err != nil {
+			return
+		}
+	}
+	return
+}
diff --git a/debug.go b/debug.go
new file mode 100644
index 0000000..74c448e
--- /dev/null
+++ b/debug.go
@@ -0,0 +1,43 @@
+// +build debug
+
+package ffmpeg_go
+
+import (
+	"fmt"
+	"log"
+	"strings"
+)
+
+func DebugNodes(node []DagNode) {
+	b := strings.Builder{}
+	for _, n := range node {
+		b.WriteString(fmt.Sprintf("%s\n", n.String()))
+	}
+	log.Println(b.String())
+}
+
+func DebugOutGoingMap(node []DagNode, m map[int]map[Label][]NodeInfo) {
+	b := strings.Builder{}
+	h := map[int]DagNode{}
+	for _, n := range node {
+		h[n.Hash()] = n
+	}
+	for k, v := range m {
+		b.WriteString(fmt.Sprintf("[Key]: %s", h[k].String()))
+		b.WriteString(" [Value]: {")
+		for l, mm := range v {
+			if l == "" {
+				l = "None"
+			}
+			b.WriteString(fmt.Sprintf("%s: [", l))
+			for _, x := range mm {
+				b.WriteString(x.Node.String())
+				b.WriteString(", ")
+			}
+			b.WriteString("]")
+		}
+		b.WriteString("}")
+		b.WriteString("\n")
+	}
+	log.Println(b.String())
+}
diff --git a/debugx.go b/debugx.go
new file mode 100644
index 0000000..933dd72
--- /dev/null
+++ b/debugx.go
@@ -0,0 +1,5 @@
+package ffmpeg_go
+
+func DebugNodes(node []DagNode) {}
+
+func DebugOutGoingMap(node []DagNode, m map[int]map[Label][]NodeInfo) {}
diff --git a/ffmpeg.go b/ffmpeg.go
new file mode 100644
index 0000000..f42e606
--- /dev/null
+++ b/ffmpeg.go
@@ -0,0 +1,92 @@
+package ffmpeg_go
+
+import (
+	"errors"
+)
+
+// Input file URL (ffmpeg ``-i`` option)
+//
+// Any supplied kwargs are passed to ffmpeg verbatim (e.g. ``t=20``,
+// ``f='mp4'``, ``acodec='pcm'``, etc.).
+//
+// To tell ffmpeg to read from stdin, use ``pipe:`` as the filename.
+//
+// Official documentation: `Main options <https://ffmpeg.org/ffmpeg.html#Main-options>`__
+func Input(filename string, kwargs ...KwArgs) *Stream {
+	args := MergeKwArgs(kwargs)
+	args["filename"] = filename
+	if fmt := args.PopString("f"); fmt != "" {
+		if args.HasKey("format") {
+			panic(errors.New("can't specify both `format` and `f` options"))
+		}
+		args["format"] = fmt
+	}
+	return NewInputNode("input", nil, args).Stream("", "")
+}
+
+// Add extra global command-line argument(s), e.g. ``-progress``.
+func (s *Stream) GlobalArgs(args ...string) *Stream {
+	if s.Type != "OutputStream" {
+		panic("cannot overwrite outputs on non-OutputStream")
+	}
+	return NewGlobalNode("global_args", []*Stream{s}, args, nil).Stream("", "")
+}
+
+// Overwrite output files without asking (ffmpeg ``-y`` option)
+//
+// Official documentation: `Main options <https://ffmpeg.org/ffmpeg.html#Main-options>`_
+func (s *Stream) OverwriteOutput(stream *Stream) *Stream {
+	if s.Type != "OutputStream" {
+		panic("cannot overwrite outputs on non-OutputStream")
+	}
+	return NewGlobalNode("overwrite_output", []*Stream{stream}, []string{"-y"}, nil).Stream("", "")
+}
+
+// Include all given outputs in one ffmpeg command line
+func (s *Stream) MergeOutputs(streams ...*Stream) *Stream {
+	if s.Type != "OutputStream" {
+		panic("cannot merge outputs on non-OutputStream")
+	}
+	return NewMergeOutputsNode("merge_output", streams).Stream("", "")
+}
+
+func Output(streams []*Stream, fileName string, kwargs ...KwArgs) *Stream {
+	args := MergeKwArgs(kwargs)
+	if !args.HasKey("filename") {
+		if fileName == "" {
+			panic("filename must be provided")
+		}
+		args["filename"] = fileName
+	}
+
+	return NewOutputNode("output", streams, nil, args).Stream("", "")
+}
+
+//Output file URL
+//
+//    Syntax:
+//        `ffmpeg.output(stream1[, stream2, stream3...], filename, **ffmpeg_args)`
+//
+//    Any supplied keyword arguments are passed to ffmpeg verbatim (e.g.
+//    ``t=20``, ``f='mp4'``, ``acodec='pcm'``, ``vcodec='rawvideo'``,
+//    etc.).  Some keyword-arguments are handled specially, as shown below.
+//
+//    Args:
+//        video_bitrate: parameter for ``-b:v``, e.g. ``video_bitrate=1000``.
+//        audio_bitrate: parameter for ``-b:a``, e.g. ``audio_bitrate=200``.
+//        format: alias for ``-f`` parameter, e.g. ``format='mp4'``
+//            (equivalent to ``f='mp4'``).
+//
+//    If multiple streams are provided, they are mapped to the same
+//    output.
+//
+//    To tell ffmpeg to write to stdout, use ``pipe:`` as the filename.
+//
+//    Official documentation: `Synopsis <https://ffmpeg.org/ffmpeg.html#Synopsis>`__
+//    """
+func (s *Stream) Output(fileName string, kwargs ...KwArgs) *Stream {
+	if s.Type != "FilterableStream" {
+		panic("cannot output on non-FilterableStream")
+	}
+	return Output([]*Stream{s}, fileName, kwargs...)
+}
diff --git a/ffmpeg_test.go b/ffmpeg_test.go
new file mode 100644
index 0000000..2634350
--- /dev/null
+++ b/ffmpeg_test.go
@@ -0,0 +1,280 @@
+package ffmpeg_go
+
+import (
+	"bytes"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/u2takey/go-utils/rand"
+)
+
+func TestFluentEquality(t *testing.T) {
+	base1 := Input("dummy1.mp4")
+	base2 := Input("dummy1.mp4")
+	base3 := Input("dummy2.mp4")
+	t1 := base1.Trim(KwArgs{"start_frame": 10, "end_frame": 20})
+	t2 := base1.Trim(KwArgs{"start_frame": 10, "end_frame": 20})
+	t3 := base1.Trim(KwArgs{"start_frame": 10, "end_frame": 30})
+	t4 := base2.Trim(KwArgs{"start_frame": 10, "end_frame": 20})
+	t5 := base3.Trim(KwArgs{"start_frame": 10, "end_frame": 20})
+
+	assert.Equal(t, t1.Hash(), t2.Hash())
+	assert.Equal(t, t1.Hash(), t4.Hash())
+	assert.NotEqual(t, t1.Hash(), t3.Hash())
+	assert.NotEqual(t, t1.Hash(), t5.Hash())
+}
+
+func TestFluentConcat(t *testing.T) {
+	base1 := Input("dummy1.mp4", nil)
+	trim1 := base1.Trim(KwArgs{"start_frame": 10, "end_frame": 20})
+	trim2 := base1.Trim(KwArgs{"start_frame": 30, "end_frame": 40})
+	trim3 := base1.Trim(KwArgs{"start_frame": 50, "end_frame": 60})
+	concat1 := Concat([]*Stream{trim1, trim2, trim3})
+	concat2 := Concat([]*Stream{trim1, trim2, trim3})
+	concat3 := Concat([]*Stream{trim1, trim3, trim2})
+	assert.Equal(t, concat1.Hash(), concat2.Hash())
+	assert.NotEqual(t, concat1.Hash(), concat3.Hash())
+}
+
+func TestRepeatArgs(t *testing.T) {
+	o := Input("dummy.mp4", nil).Output("dummy2.mp4",
+		KwArgs{"streamid": []string{"0:0x101", "1:0x102"}})
+	assert.Equal(t, o.GetArgs(), []string{"-i", "dummy.mp4", "-streamid", "0:0x101", "-streamid", "1:0x102", "dummy2.mp4"})
+}
+
+func TestGlobalArgs(t *testing.T) {
+	o := Input("dummy.mp4", nil).Output("dummy2.mp4", nil).GlobalArgs("-progress", "someurl")
+
+	assert.Equal(t, o.GetArgs(), []string{
+		"-i",
+		"dummy.mp4",
+		"dummy2.mp4",
+		"-progress",
+		"someurl",
+	})
+}
+
+func TestSimpleExample(t *testing.T) {
+	err := Input(TestInputFile1, nil).
+		Output(TestOutputFile1, nil).
+		OverWriteOutput().
+		Run()
+	assert.Nil(t, err)
+}
+
+func ComplexFilterExample() *Stream {
+	split := Input(TestInputFile1).VFlip().Split()
+	split0, split1 := split.Get("0"), split.Get("1")
+	overlayFile := Input(TestOverlayFile).Crop(10, 10, 158, 112)
+	return Concat([]*Stream{
+		split0.Trim(KwArgs{"start_frame": 10, "end_frame": 20}),
+		split1.Trim(KwArgs{"start_frame": 30, "end_frame": 40})}).
+		Overlay(overlayFile.HFlip(), "").
+		DrawBox(50, 50, 120, 120, "red", 5).
+		Output(TestOutputFile1).
+		OverWriteOutput()
+}
+
+func TestComplexFilterExample(t *testing.T) {
+	assert.Equal(t, []string{
+		"-i",
+		TestInputFile1,
+		"-i",
+		TestOverlayFile,
+		"-filter_complex",
+		"[0]vflip[s0];" +
+			"[s0]split=2[s1][s2];" +
+			"[s1]trim=end_frame=20:start_frame=10[s3];" +
+			"[s2]trim=end_frame=40:start_frame=30[s4];" +
+			"[s3][s4]concat=n=2[s5];" +
+			"[1]crop=158:112:10:10[s6];" +
+			"[s6]hflip[s7];" +
+			"[s5][s7]overlay=eof_action=repeat[s8];" +
+			"[s8]drawbox=50:50:120:120:red:t=5[s9]",
+		"-map",
+		"[s9]",
+		TestOutputFile1,
+		"-y",
+	}, ComplexFilterExample().GetArgs())
+}
+
+func TestCombinedOutput(t *testing.T) {
+	i1 := Input(TestInputFile1)
+	i2 := Input(TestOverlayFile)
+	out := Output([]*Stream{i1, i2}, TestOutputFile1)
+	assert.Equal(t, []string{
+		"-i",
+		TestInputFile1,
+		"-i",
+		TestOverlayFile,
+		"-map",
+		"0",
+		"-map",
+		"1",
+		TestOutputFile1,
+	}, out.GetArgs())
+}
+
+func TestFilterWithSelector(t *testing.T) {
+	i := Input(TestInputFile1)
+
+	v1 := i.Video().HFlip()
+	a1 := i.Audio().Filter("aecho", []string{"0.8", "0.9", "1000", "0.3"})
+
+	out := Output([]*Stream{a1, v1}, TestOutputFile1)
+	assert.Equal(t, []string{
+		"-i",
+		TestInputFile1,
+		"-filter_complex",
+		"[0:a]aecho=0.8:0.9:1000:0.3[s0];[0:v]hflip[s1]",
+		"-map",
+		"[s0]",
+		"-map",
+		"[s1]",
+		TestOutputFile1}, out.GetArgs())
+
+}
+
+func ComplexFilterAsplitExample() *Stream {
+	split := Input(TestInputFile1).VFlip().ASplit()
+	split0 := split.Get("0")
+	split1 := split.Get("1")
+
+	return Concat([]*Stream{
+		split0.Filter("atrim", nil, KwArgs{"start": 10, "end": 20}),
+		split1.Filter("atrim", nil, KwArgs{"start": 30, "end": 40}),
+	}).Output(TestOutputFile1).OverWriteOutput()
+}
+
+func TestFilterConcatVideoOnly(t *testing.T) {
+	in1 := Input("in1.mp4")
+	in2 := Input("in2.mp4")
+	args := Concat([]*Stream{in1, in2}).Output("out.mp4").GetArgs()
+	assert.Equal(t, []string{
+		"-i",
+		"in1.mp4",
+		"-i",
+		"in2.mp4",
+		"-filter_complex",
+		"[0][1]concat=n=2[s0]",
+		"-map",
+		"[s0]",
+		"out.mp4",
+	}, args)
+}
+
+func TestFilterConcatAudioOnly(t *testing.T) {
+	in1 := Input("in1.mp4")
+	in2 := Input("in2.mp4")
+	args := Concat([]*Stream{in1, in2}, KwArgs{"v": 0, "a": 1}).Output("out.mp4").GetArgs()
+	assert.Equal(t, []string{
+		"-i",
+		"in1.mp4",
+		"-i",
+		"in2.mp4",
+		"-filter_complex",
+		"[0][1]concat=a=1:n=2:v=0[s0]",
+		"-map",
+		"[s0]",
+		"out.mp4",
+	}, args)
+}
+
+func TestFilterConcatAudioVideo(t *testing.T) {
+	in1 := Input("in1.mp4")
+	in2 := Input("in2.mp4")
+	joined := Concat([]*Stream{in1.Video(), in1.Audio(), in2.HFlip(), in2.Get("a")}, KwArgs{"v": 1, "a": 1}).Node
+	args := Output([]*Stream{joined.Get("0"), joined.Get("1")}, "out.mp4").GetArgs()
+	assert.Equal(t, []string{
+		"-i",
+		"in1.mp4",
+		"-i",
+		"in2.mp4",
+		"-filter_complex",
+		"[1]hflip[s0];[0:v][0:a][s0][1:a]concat=a=1:n=2:v=1[s1][s2]",
+		"-map",
+		"[s1]",
+		"-map",
+		"[s2]",
+		"out.mp4",
+	}, args)
+}
+
+func TestFilterASplit(t *testing.T) {
+	out := ComplexFilterAsplitExample()
+	args := out.GetArgs()
+	assert.Equal(t, []string{
+		"-i",
+		TestInputFile1,
+		"-filter_complex",
+		"[0]vflip[s0];[s0]asplit=2[s1][s2];[s1]atrim=end=20:start=10[s3];[s2]atrim=end=40:start=30[s4];[s3][s4]concat=n=2[s5]",
+		"-map",
+		"[s5]",
+		TestOutputFile1,
+		"-y",
+	}, args)
+}
+
+func TestOutputBitrate(t *testing.T) {
+	args := Input("in").Output("out", KwArgs{"video_bitrate": 1000, "audio_bitrate": 200}).GetArgs()
+	assert.Equal(t, []string{"-i", "in", "-b:v", "1000", "-b:a", "200", "out"}, args)
+}
+
+func TestOutputVideoSize(t *testing.T) {
+	args := Input("in").Output("out", KwArgs{"video_size": "320x240"}).GetArgs()
+	assert.Equal(t, []string{"-i", "in", "-video_size", "320x240", "out"}, args)
+}
+
+func TestCompile(t *testing.T) {
+	out := Input("dummy.mp4").Output("dummy2.mp4")
+	assert.Equal(t, out.Compile().Args, []string{"ffmpeg", "-i", "dummy.mp4", "dummy2.mp4"})
+}
+
+func TestPipe(t *testing.T) {
+
+	width, height := 32, 32
+	frameSize := width * height * 3
+	frameCount, startFrame := 10, 2
+	_, _ = frameCount, frameSize
+
+	out := Input(
+		"pipe:0",
+		KwArgs{
+			"format":       "rawvideo",
+			"pixel_format": "rgb24",
+			"video_size":   fmt.Sprintf("%dx%d", width, height),
+			"framerate":    10}).
+		Trim(KwArgs{"start_frame": startFrame}).
+		Output("pipe:1", KwArgs{"format": "rawvideo"})
+
+	args := out.GetArgs()
+	assert.Equal(t, args, []string{
+		"-f",
+		"rawvideo",
+		"-video_size",
+		fmt.Sprintf("%dx%d", width, height),
+		"-framerate",
+		"10",
+		"-pixel_format",
+		"rgb24",
+		"-i",
+		"pipe:0",
+		"-filter_complex",
+		"[0]trim=start_frame=2[s0]",
+		"-map",
+		"[s0]",
+		"-f",
+		"rawvideo",
+		"pipe:1",
+	})
+
+	inBuf := bytes.NewBuffer(nil)
+	for i := 0; i < frameSize*frameCount; i++ {
+		inBuf.WriteByte(byte(rand.IntnRange(0, 255)))
+	}
+	outBuf := bytes.NewBuffer(nil)
+	err := out.WithInput(inBuf).WithOutput(outBuf).Run()
+	assert.Nil(t, err)
+	assert.Equal(t, outBuf.Len(), frameSize*(frameCount-startFrame))
+}
diff --git a/filters.go b/filters.go
new file mode 100644
index 0000000..9a8f479
--- /dev/null
+++ b/filters.go
@@ -0,0 +1,134 @@
+package ffmpeg_go
+
+import (
+	"fmt"
+	"strconv"
+)
+
+func AssetType(hasType, expectType string, action string) {
+	if hasType != expectType {
+		panic(fmt.Sprintf("cannot %s on non-%s", action, expectType))
+	}
+}
+
+func FilterMultiOutput(streamSpec []*Stream, filterName string, args []string, kwArgs ...KwArgs) *Node {
+	return NewFilterNode(filterName, streamSpec, -1, args, MergeKwArgs(kwArgs))
+}
+
+func Filter(streamSpec []*Stream, filterName string, args []string, kwArgs ...KwArgs) *Stream {
+	return FilterMultiOutput(streamSpec, filterName, args, MergeKwArgs(kwArgs)).Stream("", "")
+}
+
+func (s *Stream) Filter(filterName string, args []string, kwArgs ...KwArgs) *Stream {
+	AssetType(s.Type, "FilterableStream", "filter")
+	return Filter([]*Stream{s}, filterName, args, MergeKwArgs(kwArgs))
+}
+
+func (s *Stream) Split() *Node {
+	AssetType(s.Type, "FilterableStream", "split")
+	return NewFilterNode("split", []*Stream{s}, 1, nil, nil)
+}
+
+func (s *Stream) ASplit() *Node {
+	AssetType(s.Type, "FilterableStream", "asplit")
+	return NewFilterNode("asplit", []*Stream{s}, 1, nil, nil)
+}
+
+func (s *Stream) SetPts(expr string) *Node {
+	AssetType(s.Type, "FilterableStream", "setpts")
+	return NewFilterNode("setpts", []*Stream{s}, 1, []string{expr}, nil)
+}
+
+func (s *Stream) Trim(kwargs ...KwArgs) *Stream {
+	AssetType(s.Type, "FilterableStream", "trim")
+	return NewFilterNode("trim", []*Stream{s}, 1, nil, MergeKwArgs(kwargs)).Stream("", "")
+}
+
+func (s *Stream) Overlay(overlayParentNode *Stream, eofAction string, kwargs ...KwArgs) *Stream {
+	AssetType(s.Type, "FilterableStream", "overlay")
+	if eofAction == "" {
+		eofAction = "repeat"
+	}
+	args := MergeKwArgs(kwargs)
+	args["eof_action"] = eofAction
+	return NewFilterNode("overlay", []*Stream{s, overlayParentNode}, 2, nil, args).Stream("", "")
+}
+
+func (s *Stream) HFlip(kwargs ...KwArgs) *Stream {
+	AssetType(s.Type, "FilterableStream", "hflip")
+	return NewFilterNode("hflip", []*Stream{s}, 1, nil, MergeKwArgs(kwargs)).Stream("", "")
+}
+
+func (s *Stream) VFlip(kwargs ...KwArgs) *Stream {
+	AssetType(s.Type, "FilterableStream", "vflip")
+	return NewFilterNode("vflip", []*Stream{s}, 1, nil, MergeKwArgs(kwargs)).Stream("", "")
+}
+
+func (s *Stream) Crop(x, y, w, h int, kwargs ...KwArgs) *Stream {
+	AssetType(s.Type, "FilterableStream", "crop")
+	return NewFilterNode("crop", []*Stream{s}, 1, []string{
+		strconv.Itoa(w), strconv.Itoa(h), strconv.Itoa(x), strconv.Itoa(y),
+	}, MergeKwArgs(kwargs)).Stream("", "")
+}
+
+func (s *Stream) DrawBox(x, y, w, h int, color string, thickness int, kwargs ...KwArgs) *Stream {
+	AssetType(s.Type, "FilterableStream", "drawbox")
+	args := MergeKwArgs(kwargs)
+	if thickness != 0 {
+		args["t"] = thickness
+	}
+	return NewFilterNode("drawbox", []*Stream{s}, 1, []string{
+		strconv.Itoa(x), strconv.Itoa(y), strconv.Itoa(w), strconv.Itoa(h), color,
+	}, args).Stream("", "")
+}
+
+func (s *Stream) Drawtext(text string, x, y int, escape bool, kwargs ...KwArgs) *Stream {
+	AssetType(s.Type, "FilterableStream", "drawtext")
+	args := MergeKwArgs(kwargs)
+	if escape {
+		text = fmt.Sprintf("%q", text)
+	}
+	if text != "" {
+		args["text"] = text
+	}
+	if x != 0 {
+		args["x"] = x
+	}
+
+	if y != 0 {
+		args["y"] = y
+	}
+
+	return NewFilterNode("drawtext", []*Stream{s}, 1, nil, args).Stream("", "")
+}
+
+func Concat(streams []*Stream, kwargs ...KwArgs) *Stream {
+	args := MergeKwArgs(kwargs)
+	vsc := args.GetDefault("v", 1).(int)
+	asc := args.GetDefault("a", 0).(int)
+	sc := vsc + asc
+	if len(streams)%sc != 0 {
+		panic("streams count not valid")
+	}
+	args["n"] = len(streams) / sc
+	return NewFilterNode("concat", streams, -1, nil, args).Stream("", "")
+}
+
+func (s *Stream) Concat(streams []*Stream, kwargs ...KwArgs) *Stream {
+	return Concat(append(streams, s), MergeKwArgs(kwargs))
+}
+
+func (s *Stream) ZoomPan(kwargs ...KwArgs) *Stream {
+	AssetType(s.Type, "FilterableStream", "zoompan")
+	return NewFilterNode("zoompan", []*Stream{s}, 1, nil, MergeKwArgs(kwargs)).Stream("", "")
+}
+
+func (s *Stream) Hue(kwargs ...KwArgs) *Stream {
+	AssetType(s.Type, "FilterableStream", "hue")
+	return NewFilterNode("hue", []*Stream{s}, 1, nil, MergeKwArgs(kwargs)).Stream("", "")
+}
+
+func (s *Stream) ColorChannelMixer(kwargs ...KwArgs) *Stream {
+	AssetType(s.Type, "FilterableStream", "colorchannelmixer")
+	return NewFilterNode("colorchannelmixer", []*Stream{s}, 1, nil, MergeKwArgs(kwargs)).Stream("", "")
+}
diff --git a/go.mod b/go.mod
new file mode 100644
index 0000000..2648d1c
--- /dev/null
+++ b/go.mod
@@ -0,0 +1,9 @@
+module github.com/u2takey/ffmpeg-go
+
+go 1.14
+
+require (
+	github.com/stretchr/testify v1.4.0
+	github.com/tidwall/gjson v1.6.3
+	github.com/u2takey/go-utils v0.0.0-20200713025200-4704d09fc2c7
+)
diff --git a/go.sum b/go.sum
new file mode 100644
index 0000000..fdd0fa1
--- /dev/null
+++ b/go.sum
@@ -0,0 +1,44 @@
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/docker/distribution v2.7.1+incompatible/go.mod h1:J2gT2udsDAN96Uj4KfcMRqY0/ypR+oyYUYmja8H+y+w=
+github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
+github.com/go-logr/logr v0.1.0/go.mod h1:ixOQHD9gLJUVQQ2ZOR7zLEifBX6tGkNJF4QyIY7sIas=
+github.com/golang/groupcache v0.0.0-20190129154638-5b532d6fd5ef/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
+github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
+github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
+github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
+github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
+github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
+github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
+github.com/opencontainers/go-digest v1.0.0-rc1/go.mod h1:cMLVZDEM3+U2I4VmLI6N8jQYUd2OVphdqWwCJHrFt2s=
+github.com/opencontainers/selinux v1.5.2/go.mod h1:yTcKuYAh6R95iDpefGLQaPaRwJFwyzAJufJyiTt7s0g=
+github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/spf13/afero v1.2.2/go.mod h1:9ZxEEn6pIJ8Rxe320qSDBk6AsU0r9pR7Q4OcevTdifk=
+github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJyk=
+github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
+github.com/tidwall/gjson v1.6.3 h1:aHoiiem0dr7GHkW001T1SMTJ7X5PvyekH5WX0whWGnI=
+github.com/tidwall/gjson v1.6.3/go.mod h1:BaHyNc5bjzYkPqgLq7mdVzeiRtULKULXLgZFKsxEHI0=
+github.com/tidwall/match v1.0.1 h1:PnKP62LPNxHKTwvHHZZzdOAOCtsJTjo6dZLCwpKm5xc=
+github.com/tidwall/match v1.0.1/go.mod h1:LujAq0jyVjBy028G1WhWfIzbpQfMO8bBZ6Tyb0+pL9E=
+github.com/tidwall/pretty v1.0.2 h1:Z7S3cePv9Jwm1KwS0513MRaoUe3S01WPbLNV40pwWZU=
+github.com/tidwall/pretty v1.0.2/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
+github.com/u2takey/go-utils v0.0.0-20200713025200-4704d09fc2c7 h1:PT7mE8HJE1mwaSazrOdSeByJ1FoV33/fHUZrBB+zwVU=
+github.com/u2takey/go-utils v0.0.0-20200713025200-4704d09fc2c7/go.mod h1:ATqKFpgjUIlhGRs8j59gXmu8Cmpo1QQEHV6vwu1hs28=
+golang.org/x/sys v0.0.0-20191115151921-52ab43148777/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
+golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
+gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.2.8 h1:obN1ZagJSUGI0Ek/LBmuj4SNLPfIny3KsKFopxRdj10=
+gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+sigs.k8s.io/yaml v1.2.0/go.mod h1:yfXDCHCao9+ENCvLSE62v9VSji2MKu5jeNfTrofGhJc=
diff --git a/node.go b/node.go
new file mode 100644
index 0000000..76df745
--- /dev/null
+++ b/node.go
@@ -0,0 +1,269 @@
+package ffmpeg_go
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/u2takey/go-utils/sets"
+)
+
+type Stream struct {
+	Node     *Node
+	Label    Label
+	Selector Selector
+	Type     string
+	Context  context.Context
+}
+
+func NewStream(node *Node, streamType string, label Label, selector Selector) *Stream {
+	return &Stream{
+		Node:     node,
+		Label:    label,
+		Selector: selector,
+		Type:     streamType,
+		Context:  context.Background(),
+	}
+}
+
+func (s *Stream) Hash() int {
+	return s.Node.Hash() + getHash(s.Label)
+}
+
+func (s *Stream) Equal(other Stream) bool {
+	return s.Hash() == other.Hash()
+}
+
+func (s *Stream) String() string {
+	return fmt.Sprintf("node: %s, label: %s, selector: %s", s.Node.String(), s.Label, s.Selector)
+}
+
+func (s *Stream) Get(index string) *Stream {
+	if s.Selector != "" {
+		panic(errors.New("stream already has a selector"))
+	}
+	return s.Node.Stream(s.Label, Selector(index))
+}
+
+func (s *Stream) Audio() *Stream {
+	return s.Get("a")
+}
+
+func (s *Stream) Video() *Stream {
+	return s.Get("v")
+}
+
+func getStreamMap(streamSpec []*Stream) map[int]*Stream {
+	m := map[int]*Stream{}
+	for i := range streamSpec {
+		m[i] = streamSpec[i]
+	}
+	return m
+}
+
+func getStreamMapNodes(streamMap map[int]*Stream) (ret []*Node) {
+	for k := range streamMap {
+		ret = append(ret, streamMap[k].Node)
+	}
+	return ret
+}
+
+func getStreamSpecNodes(streamSpec []*Stream) []*Node {
+	return getStreamMapNodes(getStreamMap(streamSpec))
+}
+
+type Node struct {
+	streamSpec          []*Stream
+	name                string
+	incomingStreamTypes sets.String
+	outgoingStreamType  string
+	minInputs           int
+	maxInputs           int
+	args                []string
+	kwargs              KwArgs
+	nodeType            string
+}
+
+func NewNode(streamSpec []*Stream,
+	name string,
+	incomingStreamTypes sets.String,
+	outgoingStreamType string,
+	minInputs int,
+	maxInputs int,
+	args []string,
+	kwargs KwArgs,
+	nodeType string) *Node {
+	n := &Node{
+		streamSpec:          streamSpec,
+		name:                name,
+		incomingStreamTypes: incomingStreamTypes,
+		outgoingStreamType:  outgoingStreamType,
+		minInputs:           minInputs,
+		maxInputs:           maxInputs,
+		args:                args,
+		kwargs:              kwargs,
+		nodeType:            nodeType,
+	}
+	n.__checkInputLen()
+	n.__checkInputTypes()
+	return n
+}
+
+func NewInputNode(name string, args []string, kwargs KwArgs) *Node {
+	return NewNode(nil,
+		name,
+		nil,
+		"FilterableStream",
+		0,
+		0,
+		args,
+		kwargs,
+		"InputNode")
+}
+
+func NewFilterNode(name string, streamSpec []*Stream, maxInput int, args []string, kwargs KwArgs) *Node {
+	return NewNode(streamSpec,
+		name,
+		sets.NewString("FilterableStream"),
+		"FilterableStream",
+		1,
+		maxInput,
+		args,
+		kwargs,
+		"FilterNode")
+}
+
+func NewOutputNode(name string, streamSpec []*Stream, args []string, kwargs KwArgs) *Node {
+	return NewNode(streamSpec,
+		name,
+		sets.NewString("FilterableStream"),
+		"OutputStream",
+		1,
+		-1,
+		args,
+		kwargs,
+		"OutputNode")
+}
+
+func NewMergeOutputsNode(name string, streamSpec []*Stream) *Node {
+	return NewNode(streamSpec,
+		name,
+		sets.NewString("OutputStream"),
+		"OutputStream",
+		1,
+		-1,
+		nil,
+		nil,
+		"MergeOutputsNode")
+}
+
+func NewGlobalNode(name string, streamSpec []*Stream, args []string, kwargs KwArgs) *Node {
+	return NewNode(streamSpec,
+		name,
+		sets.NewString("OutputStream"),
+		"OutputStream",
+		1,
+		1,
+		args,
+		kwargs,
+		"GlobalNode")
+}
+
+func (n *Node) __checkInputLen() {
+	streamMap := getStreamMap(n.streamSpec)
+	if n.minInputs >= 0 && len(streamMap) < n.minInputs {
+		panic(fmt.Sprintf("Expected at least %d input stream(s); got %d", n.minInputs, len(streamMap)))
+	}
+	if n.maxInputs >= 0 && len(streamMap) > n.maxInputs {
+		panic(fmt.Sprintf("Expected at most %d input stream(s); got %d", n.maxInputs, len(streamMap)))
+	}
+}
+
+func (n *Node) __checkInputTypes() {
+	streamMap := getStreamMap(n.streamSpec)
+	for _, s := range streamMap {
+		if !n.incomingStreamTypes.Has(s.Type) {
+			panic(fmt.Sprintf("Expected incoming stream(s) to be of one of the following types: %s; got %s", n.incomingStreamTypes, s.Type))
+		}
+	}
+}
+
+func (n *Node) __getIncomingEdgeMap() map[Label]NodeInfo {
+	incomingEdgeMap := map[Label]NodeInfo{}
+	streamMap := getStreamMap(n.streamSpec)
+	for i, s := range streamMap {
+		incomingEdgeMap[Label(fmt.Sprintf("%v", i))] = NodeInfo{
+			Node:     s.Node,
+			Label:    s.Label,
+			Selector: s.Selector,
+		}
+	}
+	return incomingEdgeMap
+}
+
+func (n *Node) Hash() int {
+	b := 0
+	for downStreamLabel, upStreamInfo := range n.IncomingEdgeMap() {
+		b += getHash(fmt.Sprintf("%s%d%s%s", downStreamLabel, upStreamInfo.Node.Hash(), upStreamInfo.Label, upStreamInfo.Selector))
+	}
+	b += getHash(n.args)
+	b += getHash(n.kwargs)
+	return b
+}
+
+func (n *Node) String() string {
+	return fmt.Sprintf("%s (%s) <%s>", n.name, getString(n.args), getString(n.kwargs))
+}
+
+func (n *Node) Equal(other DagNode) bool {
+	return n.Hash() == other.Hash()
+}
+
+func (n *Node) ShortRepr() string {
+	return n.name
+}
+
+func (n *Node) IncomingEdgeMap() map[Label]NodeInfo {
+	return n.__getIncomingEdgeMap()
+}
+
+func (n *Node) GetInComingEdges() []DagEdge {
+	return GetInComingEdges(n, n.IncomingEdgeMap())
+}
+
+func (n *Node) Stream(label Label, selector Selector) *Stream {
+	return NewStream(n, n.outgoingStreamType, label, selector)
+}
+
+func (n *Node) Get(a string) *Stream {
+	l := strings.Split(a, ":")
+	if len(l) == 2 {
+		return n.Stream(Label(l[0]), Selector(l[1]))
+	}
+	return n.Stream(Label(a), "")
+}
+
+func (n *Node) GetFilter(outgoingEdges []DagEdge) string {
+	if n.nodeType != "FilterNode" {
+		panic("call GetFilter on non-FilterNode")
+	}
+	args := n.args
+	kwargs := n.kwargs
+	if n.name == "split" || n.name == "asplit" {
+		args = []string{fmt.Sprintf("%d", len(outgoingEdges))}
+	}
+	// todo escape char
+	for _, k := range kwargs.SortedKeys() {
+		v := getString(kwargs[k])
+		if v != "" {
+			args = append(args, fmt.Sprintf("%s=%s", k, v))
+		} else {
+			args = append(args, fmt.Sprintf("%s", k))
+		}
+	}
+	if len(args) > 0 {
+		return fmt.Sprintf("%s=%s", n.name, strings.Join(args, ":"))
+	}
+	return fmt.Sprintf("%s", n.name)
+}
diff --git a/probe.go b/probe.go
new file mode 100644
index 0000000..578bfa0
--- /dev/null
+++ b/probe.go
@@ -0,0 +1,40 @@
+package ffmpeg_go
+
+import (
+	"bytes"
+	"context"
+	"os/exec"
+	"time"
+)
+
+// Run ffprobe on the specified file and return a JSON representation of the output.
+//
+//  Raises:
+//        :class:`ffmpeg.Error`: if ffprobe returns a non-zero exit code,
+//            an :class:`Error` is returned with a generic error message.
+//            The stderr output can be retrieved by accessing the
+//            ``stderr`` property of the exception.
+
+func Probe(fileName string, kwargs KwArgs) (string, error) {
+	return ProbeWithTimeout(fileName, 0, kwargs)
+}
+
+func ProbeWithTimeout(fileName string, timeOut time.Duration, kwargs KwArgs) (string, error) {
+	args := []string{"-show_format", "-show_streams", "-of", "json"}
+	args = append(args, ConvertKwargsToCmdLineArgs(kwargs)...)
+	args = append(args, fileName)
+	ctx := context.Background()
+	if timeOut > 0 {
+		var cancel func()
+		ctx, cancel = context.WithTimeout(context.Background(), timeOut)
+		defer cancel()
+	}
+	cmd := exec.CommandContext(ctx, "ffprobe", args...)
+	buf := bytes.NewBuffer(nil)
+	cmd.Stdout = buf
+	err := cmd.Run()
+	if err != nil {
+		return "", err
+	}
+	return string(buf.Bytes()), nil
+}
diff --git a/probe_test.go b/probe_test.go
new file mode 100644
index 0000000..243b055
--- /dev/null
+++ b/probe_test.go
@@ -0,0 +1,20 @@
+package ffmpeg_go
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/tidwall/gjson"
+)
+
+const (
+	TestInputFile1  = "./sample_data/in1.mp4"
+	TestOutputFile1 = "./sample_data/out1.mp4"
+	TestOverlayFile = "./sample_data/overlay.png"
+)
+
+func TestProbe(t *testing.T) {
+	data, err := Probe(TestInputFile1, nil)
+	assert.Nil(t, err)
+	assert.Equal(t, gjson.Get(data, "format.duration").String(), "7.036000")
+}
diff --git a/run.go b/run.go
new file mode 100644
index 0000000..390ee74
--- /dev/null
+++ b/run.go
@@ -0,0 +1,249 @@
+package ffmpeg_go
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"os/exec"
+	"sort"
+	"strings"
+	"time"
+)
+
+func getInputArgs(node *Node) []string {
+	var args []string
+	if node.name == "input" {
+		kwargs := node.kwargs.Copy()
+		filename := kwargs.PopString("filename")
+		format := kwargs.PopString("format")
+		videoSize := kwargs.PopString("video_size")
+		if format != "" {
+			args = append(args, "-f", format)
+		}
+		if videoSize != "" {
+			args = append(args, "-video_size", videoSize)
+		}
+		args = append(args, ConvertKwargsToCmdLineArgs(kwargs)...)
+		args = append(args, "-i", filename)
+	} else {
+		panic("unsupported node input name")
+	}
+	return args
+}
+
+func formatInputStreamName(streamNameMap map[string]string, edge DagEdge, finalArg bool) string {
+	prefix := streamNameMap[fmt.Sprintf("%d%s", edge.UpStreamNode.Hash(), edge.UpStreamLabel)]
+	suffix := ""
+	format := "[%s%s]"
+	if edge.UpStreamSelector != "" {
+		suffix = fmt.Sprintf(":%s", edge.UpStreamSelector)
+	}
+	if finalArg && edge.UpStreamNode.(*Node).nodeType == "InputNode" {
+		format = "%s%s"
+	}
+	return fmt.Sprintf(format, prefix, suffix)
+}
+
+func formatOutStreamName(streamNameMap map[string]string, edge DagEdge) string {
+	return fmt.Sprintf("[%s]", streamNameMap[fmt.Sprintf("%d%s", edge.UpStreamNode.Hash(), edge.UpStreamLabel)])
+}
+
+func _getFilterSpec(node *Node, outOutingEdgeMap map[Label][]NodeInfo, streamNameMap map[string]string) string {
+	var input, output []string
+	for _, e := range node.GetInComingEdges() {
+		input = append(input, formatInputStreamName(streamNameMap, e, false))
+	}
+	outEdges := GetOutGoingEdges(node, outOutingEdgeMap)
+	for _, e := range outEdges {
+		output = append(output, formatOutStreamName(streamNameMap, e))
+	}
+	//sort.Strings(input)
+	//sort.Strings(output)
+	return fmt.Sprintf("%s%s%s", strings.Join(input, ""), node.GetFilter(outEdges), strings.Join(output, ""))
+}
+
+func _getAllLabelsInSorted(m map[Label]NodeInfo) []Label {
+	var r []Label
+	for a := range m {
+		r = append(r, a)
+	}
+	sort.Slice(r, func(i, j int) bool {
+		return r[i] < r[j]
+	})
+	return r
+}
+
+func _getAllLabelsSorted(m map[Label][]NodeInfo) []Label {
+	var r []Label
+	for a := range m {
+		r = append(r, a)
+	}
+	sort.Slice(r, func(i, j int) bool {
+		return r[i] < r[j]
+	})
+	return r
+}
+
+func _allocateFilterStreamNames(nodes []*Node, outOutingEdgeMaps map[int]map[Label][]NodeInfo, streamNameMap map[string]string) {
+	sc := 0
+	for _, n := range nodes {
+		om := outOutingEdgeMaps[n.Hash()]
+		// todo sort
+		for _, l := range _getAllLabelsSorted(om) {
+			if len(om[l]) > 1 {
+				panic(fmt.Sprintf(`encountered %s with multiple outgoing edges 
+with same upstream label %s; a 'split'' filter is probably required`, n.name, l))
+			}
+			streamNameMap[fmt.Sprintf("%d%s", n.Hash(), l)] = fmt.Sprintf("s%d", sc)
+			sc += 1
+		}
+	}
+}
+
+func _getFilterArg(nodes []*Node, outOutingEdgeMaps map[int]map[Label][]NodeInfo, streamNameMap map[string]string) string {
+	_allocateFilterStreamNames(nodes, outOutingEdgeMaps, streamNameMap)
+	var filterSpec []string
+	for _, n := range nodes {
+		filterSpec = append(filterSpec, _getFilterSpec(n, outOutingEdgeMaps[n.Hash()], streamNameMap))
+	}
+	return strings.Join(filterSpec, ";")
+}
+
+func _getGlobalArgs(node *Node) []string {
+	return node.args
+}
+
+func _getOutputArgs(node *Node, streamNameMap map[string]string) []string {
+	if node.name != "output" {
+		panic("Unsupported output node")
+	}
+	var args []string
+	if len(node.GetInComingEdges()) == 0 {
+		panic("Output node has no mapped streams")
+	}
+	for _, e := range node.GetInComingEdges() {
+		streamName := formatInputStreamName(streamNameMap, e, true)
+		if streamName != "0" || len(node.GetInComingEdges()) > 1 {
+			args = append(args, "-map", streamName)
+		}
+	}
+	kwargs := node.kwargs.Copy()
+
+	filename := kwargs.PopString("filename")
+	if kwargs.HasKey("format") {
+		args = append(args, "-f", kwargs.PopString("format"))
+	}
+	if kwargs.HasKey("video_bitrate") {
+		args = append(args, "-b:v", kwargs.PopString("video_bitrate"))
+	}
+	if kwargs.HasKey("audio_bitrate") {
+		args = append(args, "-b:a", kwargs.PopString("audio_bitrate"))
+	}
+	if kwargs.HasKey("video_size") {
+		args = append(args, "-video_size", kwargs.PopString("video_size"))
+	}
+
+	args = append(args, ConvertKwargsToCmdLineArgs(kwargs)...)
+	args = append(args, filename)
+	return args
+}
+
+func (s *Stream) GetArgs() []string {
+	var args []string
+	nodes := getStreamSpecNodes([]*Stream{s})
+	var dagNodes []DagNode
+	streamNameMap := map[string]string{}
+	for i := range nodes {
+		dagNodes = append(dagNodes, nodes[i])
+	}
+	sorted, outGoingMap, err := TopSort(dagNodes)
+	if err != nil {
+		panic(err)
+	}
+	DebugNodes(sorted)
+	DebugOutGoingMap(sorted, outGoingMap)
+	var inputNodes, outputNodes, globalNodes, filterNodes []*Node
+	for i := range sorted {
+		n := sorted[i].(*Node)
+		switch n.nodeType {
+		case "InputNode":
+			streamNameMap[fmt.Sprintf("%d", n.Hash())] = fmt.Sprintf("%d", len(inputNodes))
+			inputNodes = append(inputNodes, n)
+		case "OutputNode":
+			outputNodes = append(outputNodes, n)
+		case "GlobalNode":
+			globalNodes = append(globalNodes, n)
+		case "FilterNode":
+			filterNodes = append(filterNodes, n)
+		}
+	}
+
+	for _, n := range inputNodes {
+		args = append(args, getInputArgs(n)...)
+	}
+	filterArgs := _getFilterArg(filterNodes, outGoingMap, streamNameMap)
+	if filterArgs != "" {
+		args = append(args, "-filter_complex", filterArgs)
+	}
+	for _, n := range outputNodes {
+		args = append(args, _getOutputArgs(n, streamNameMap)...)
+	}
+	for _, n := range globalNodes {
+		args = append(args, _getGlobalArgs(n)...)
+	}
+	if s.Context.Value("OverWriteOutput") != nil {
+		args = append(args, "-y")
+	}
+	return args
+}
+
+func (s *Stream) WithTimeout(timeOut time.Duration) *Stream {
+	if timeOut > 0 {
+		s.Context, _ = context.WithTimeout(s.Context, timeOut)
+	}
+	return s
+}
+
+func (s *Stream) OverWriteOutput() *Stream {
+	s.Context = context.WithValue(s.Context, "OverWriteOutput", struct{}{})
+	return s
+}
+
+func (s *Stream) WithInput(reader io.Reader) *Stream {
+	s.Context = context.WithValue(s.Context, "Stdin", reader)
+	return s
+}
+
+func (s *Stream) WithOutput(out ...io.Writer) *Stream {
+	if len(out) > 0 {
+		s.Context = context.WithValue(s.Context, "Stdout", out[0])
+	}
+	if len(out) > 1 {
+		s.Context = context.WithValue(s.Context, "Stderr", out[1])
+	}
+	return s
+}
+
+// for test
+func (s *Stream) Compile() *exec.Cmd {
+	args := s.GetArgs()
+	cmd := exec.CommandContext(s.Context, "ffmpeg", args...)
+	if a, ok := s.Context.Value("Stdin").(io.Reader); ok {
+		cmd.Stdin = a
+	}
+	if a, ok := s.Context.Value("Stdout").(io.Writer); ok {
+		cmd.Stdout = a
+	}
+	if a, ok := s.Context.Value("Stderr").(io.Writer); ok {
+		cmd.Stderr = a
+	}
+	return cmd
+}
+
+func (s *Stream) Run() error {
+	err := s.Compile().Run()
+	if err != nil {
+		return err
+	}
+	return nil
+}
diff --git a/sample_data/in1.mp4 b/sample_data/in1.mp4
new file mode 100644
index 0000000..2c7d59e
Binary files /dev/null and b/sample_data/in1.mp4 differ
diff --git a/sample_data/out1.mp4 b/sample_data/out1.mp4
new file mode 100644
index 0000000..b10b518
Binary files /dev/null and b/sample_data/out1.mp4 differ
diff --git a/sample_data/overlay.png b/sample_data/overlay.png
new file mode 100644
index 0000000..5da0087
Binary files /dev/null and b/sample_data/overlay.png differ
diff --git a/utils.go b/utils.go
new file mode 100644
index 0000000..7987da8
--- /dev/null
+++ b/utils.go
@@ -0,0 +1,183 @@
+package ffmpeg_go
+
+import (
+	"fmt"
+	"hash/fnv"
+	"sort"
+	"strconv"
+	"strings"
+)
+
+func getString(item interface{}) string {
+	if a, ok := item.(interface{ String() string }); ok {
+		return a.String()
+	}
+	switch a := item.(type) {
+	case string:
+		return a
+	case []string:
+		return strings.Join(a, ", ")
+	case []interface{}:
+		var r []string
+		for _, b := range a {
+			r = append(r, getString(b))
+		}
+		return strings.Join(r, ", ")
+	case KwArgs:
+		var keys, r []string
+		for k := range a {
+			keys = append(keys, k)
+		}
+		sort.Strings(keys)
+		for _, k := range keys {
+			r = append(r, fmt.Sprintf("%s: %s", k, getString(a[k])))
+		}
+		return fmt.Sprintf("{%s}", strings.Join(r, ", "))
+	case map[string]interface{}:
+		var keys, r []string
+		for k := range a {
+			keys = append(keys, k)
+		}
+		sort.Strings(keys)
+		for _, k := range keys {
+			r = append(r, fmt.Sprintf("%s: %s", k, getString(a[k])))
+		}
+		return fmt.Sprintf("{%s}", strings.Join(r, ", "))
+	}
+	return fmt.Sprintf("%v", item)
+}
+
+func getHash(item interface{}) int {
+	h := fnv.New64()
+	switch a := item.(type) {
+	case interface{ Hash() int }:
+		return a.Hash()
+	case string:
+		_, _ = h.Write([]byte(a))
+		return int(h.Sum64())
+	case []byte:
+		_, _ = h.Write(a)
+		return int(h.Sum64())
+	case map[string]interface{}:
+		b := 0
+		for k, v := range a {
+			b += getHash(k) + getHash(v)
+		}
+		return b
+	case KwArgs:
+		b := 0
+		for k, v := range a {
+			b += getHash(k) + getHash(v)
+		}
+		return b
+	default:
+		_, _ = h.Write([]byte(getString(item)))
+		return int(h.Sum64())
+	}
+}
+
+//def escape_chars(text, chars):
+//    """Helper function to escape uncomfortable characters."""
+//    text = str(text)
+//    chars = list(set(chars))
+//    if '\\' in chars:
+//        chars.remove('\\')
+//        chars.insert(0, '\\')
+//    for ch in chars:
+//        text = text.replace(ch, '\\' + ch)
+//    return text
+//
+
+type KwArgs map[string]interface{}
+
+func MergeKwArgs(args []KwArgs) KwArgs {
+	a := KwArgs{}
+	for _, b := range args {
+		for c := range b {
+			a[c] = b[c]
+		}
+	}
+	return a
+}
+
+func (a KwArgs) Copy() KwArgs {
+	r := KwArgs{}
+	for k := range a {
+		r[k] = a[k]
+	}
+	return r
+}
+
+func (a KwArgs) SortedKeys() []string {
+	var r []string
+	for k := range a {
+		r = append(r, k)
+	}
+	sort.Strings(r)
+	return r
+}
+
+func (a KwArgs) GetString(k string) string {
+	if v, ok := a[k]; ok {
+		return fmt.Sprintf("%v", v)
+	}
+	return ""
+}
+
+func (a KwArgs) PopString(k string) string {
+	if c, ok := a[k]; ok {
+		defer delete(a, k)
+		return fmt.Sprintf("%v", c)
+	}
+	return ""
+}
+
+func (a KwArgs) HasKey(k string) bool {
+	_, ok := a[k]
+	return ok
+}
+
+func (a KwArgs) GetDefault(k string, defaultV interface{}) interface{} {
+	if v, ok := a[k]; ok {
+		return v
+	}
+	return defaultV
+}
+
+func ConvertKwargsToCmdLineArgs(kwargs KwArgs) []string {
+	var keys, args []string
+	for k := range kwargs {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+
+	for _, k := range keys {
+		v := kwargs[k]
+		switch a := v.(type) {
+		case string:
+			args = append(args, fmt.Sprintf("-%s", k))
+			if a != "" {
+				args = append(args, a)
+			}
+		case []string:
+			for _, r := range a {
+				args = append(args, fmt.Sprintf("-%s", k))
+				if r != "" {
+					args = append(args, r)
+				}
+			}
+		case []int:
+			for _, r := range a {
+				args = append(args, fmt.Sprintf("-%s", k))
+				args = append(args, strconv.Itoa(r))
+			}
+		case int:
+			args = append(args, fmt.Sprintf("-%s", k))
+			args = append(args, strconv.Itoa(a))
+		default:
+			args = append(args, fmt.Sprintf("-%s", k))
+			args = append(args, fmt.Sprintf("%v", a))
+		}
+	}
+	return args
+}