프레임 추출 테스트 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()

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

#!/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