hemisphere/src/overlay/filters.py

import threading


def filter_escape(s):
    return (str(s)
        .replace(':', '\\:')
        .replace(',', '\\,')
        .replace(';', '\\;')
    )

class FFmpegInput:
    def __init__(self, name, **options):
        self.name = name
        self.options = options

    def format(self):
        args = []
        for (k, v) in self.options.items():
            args.append('-' + k)
            if v is not None:
                args.append(str(v))
        args.append('-i')
        args.append(self.name)
        return args

class FFmpegFilter:
    def __init__(self, name, *, ins=None, outs=None, **args):
        self.name = name
        self.args = args
        self.ins = ins or []
        self.outs = outs or []

    def __repr__(self):
        return '{}({}, {}, ins={!r}, outs={!r})'.format(
            self.__class__.__name__,
            self.name,
            ', '.join('{}={}'.format(k, v) for k, v in self.args.items()),
            self.ins,
            self.outs
        )

    def __str__(self):
        return '{}{}{}{}{}'.format(
            ''.join('[{}] '.format(i) for i in self.ins),
            filter_escape(self.name),
            '=' if self.args else '',
            ':'.join('{}={}'.format(k, filter_escape(v)) for k, v in self.args.items()),
            ''.join(' [{}]'.format(o) for o in self.outs),
        )

class FFmpegChain:
    def __init__(self, *filters, inputs=None, runners=None):
        self.filters = list(filters)
        self.inputs = inputs or []
        self.runners = runners or []

    def append(self, f):
        self.filters.append(f)

    def extend(self, l):
        self.filters.extend(l)

    def __getitem__(self, i):
        return self.filters[i]

    def __add__(self, o):
        if not isinstance(o, FFmpegChain):
            raise TypeError('incompatible')
        return FFmpegChain(*(self.filters + o.filters))

    def __repr__(self):
        return '{}({})'.format(self.__class__.__name__, ', '.join(repr(f) for f in self.filters))

    def format_inputs(self):
        args = []
        for input in self.inputs:
            args.extend(input.format())
        return args

    def format_filters(self):
        return ', '.join(str(f) for f in self.filters)

    def run(self):
        for r in self.runners:
            t = threading.Thread(target=r, daemon=True)
            t.start()

class FFmpegGraph:
    def __init__(self, *chains):
        self.chains = list(chains)

    def append(self, f):
        self.chains.append(f)

    def extend(self, l, link=False):
        if self.chains and link:
            l[0][0].ins = self.chains[-1][-1].outs
        self.chains.extend(l)

    def fixup(self):
        self.chains[-1][-1].outs = []

    def __getitem__(self, i):
        return self.chains[i]

    def __repr__(self):
        return '{}({})'.format(self.__class__.__name__, ', '.join(repr(c) for c in self.chains))

    def format(self):
        args = []

        in_no = 0
        for c in self.chains:
            args.extend(c.format_inputs())
            for i in range(len(c.inputs)):
                c.filters[0].ins.append(str(in_no))
                in_no += 1

        args.append('-filter_complex')
        args.append('; '.join(c.format_filters() for c in self.chains))
        return args

    def run(self):
        for c in self.chains:
            c.run()