프레임 추출 테스트 on Luigi

luigi_test.py

#!/usr/bin/env python
# Usage>
# $ pip install luigi
# $ python luigi_test.py FrameExtractTask --filename small_file.mp4 --local-scheduler

import os
import luigi
import sys
import time
import subprocess
from pathlib import Path


class FFMPEGTask(luigi.Task):
    filename = luigi.Parameter()
    
    def requires(self):
        return []

    def run(self):
        start_time = time.time()
        subprocess.run(['ffmpeg', '-i', self.filename, '-r', '1/1', 'frame%03d.jpg'])
        print("Processing time = {}".format(time.time() - start_time))
        _out = self.output().open('w')
        _out.write(u"done!\n")
        _out.close()

    def output(self):
        return luigi.LocalTarget('result.txt')


class FrameExtractTask(luigi.Task):
    filename = luigi.Parameter()

    def requires(self):
        return FFMPEGTask(filename=self.filename)

    def run(self):
        for f in Path('.').glob('*.jpg'):
            f.unlink()
        os.remove('result.txt')


if __name__ == '__main__':
    luigi.run()

• 다중 파일 처리

#!/usr/bin/env python
# Usage>
# $ pip install luigi
# $ python luigi_test2.py FrameExtractTask --filecount 5 --local-scheduler

import os
import luigi
import sys
import time
import subprocess
from pathlib import Path

class TimeTaskMixin(object):
    '''
    A mixin that when added to a luigi task, will print out
    the tasks execution time to standard out, when the task is
    finished
    '''
    @luigi.Task.event_handler(luigi.Event.PROCESSING_TIME)
    def print_execution_time(self, processing_time):
        print('### PROCESSING TIME ###: ' + str(processing_time))


class FFMPEGTask(luigi.Task):
    index = luigi.IntParameter()
    
    def requires(self):
        return []

    def run(self):
        filename = 'files/small_file{}.mp4'.format(self.index)
        subprocess.run(['ffmpeg', '-i', filename, '-r', '1/1', 'frame%03d.jpg'])

        _out = self.output().open('w')
        _out.write(u"done!\n")
        _out.close()

    def output(self):
        return luigi.LocalTarget('result.txt')


class FrameRemoveTask(luigi.Task, TimeTaskMixin):
    index = luigi.IntParameter()

    def requires(self):
        return FFMPEGTask(index=self.index)

    def run(self):
        for f in Path('.').glob('*.jpg'):
            f.unlink()
        os.remove('result.txt')
        _out = self.output().open('w')
        _out.write(u"done!\n")
        _out.close()

    def output(self):
        return luigi.LocalTarget('done{}.txt'.format(self.index))


class FrameExtractTask(luigi.Task, TimeTaskMixin):
    filecount = luigi.IntParameter()

    def requires(self):
        return [FrameRemoveTask(index=i) for i in range(1, self.filecount+1)]

    def run(self):
        for f in Path('.').glob('*.txt'):
            f.unlink()


if __name__ == '__main__':
    start_time = time.time()
    luigi.run()
    print("Processing time = {}".format(time.time() - start_time))

• 동시 요청 테스트 스크립트

#!/bin/bash
# Usage> $ time ./run_task_parallel.sh 5
for i in $(seq 1 $1)
do
	python luigi_test.py FrameExtractTask --filename files/small_file$i.mp4 --local-scheduler &
done
wait