hpmor_split/splitSrc/ffmpeg.py

84 lines
1.9 KiB
Python

#!/usr/bin/env python3
# -*- coding: UTF-8 -*-
#
# ffmpeg.py
#
# Copyright 2023 Beat Jäckle <beat@git.jdmweb2.ch>
from ffmpeg import FFmpeg
import asyncio
async def main(part, cid, time: dict, end, title, filenamemuster, fileformat):
utitle = title['t_']
title = title['t']
filename = f'out/{cid:03}_{utitle}.{fileformat}'
ffmpeg = (
FFmpeg()
.option('y')
.option('hide_banner')
)
if end is not None:
ffmpeg.input(
filenamemuster[0] +
str(part) +
filenamemuster[1],
ss=time,
to=end,
)
else:
ffmpeg.input(
filenamemuster[0] +
str(part) +
filenamemuster[1],
ss=time,
to=end,
)
ffmpeg.output(
filename,
# metadata=[f'title={title}', f'track={cid}'],
options={
'metadata': [f'title={title}', f'track={cid}'],
"codec:a": "copy"
}
)
@ffmpeg.on("start")
def on_start(arguments):
print("Arguments:", arguments)
@ffmpeg.on("stderr")
def on_stderr(line):
pass
# print("stderr:", line)
@ffmpeg.on("progress")
def on_progress(progress):
print(progress)
@ffmpeg.on("progress")
def time_to_terminate(progress):
# Gracefully terminate when more than 200 frames are processed
if progress.frame > 200:
ffmpeg.terminate()
@ffmpeg.on("completed")
def on_completed():
print("Completed")
@ffmpeg.on("terminated")
def on_terminated():
print("Terminated")
@ffmpeg.on("error")
def on_error(code):
print("Error:", code)
await ffmpeg.execute()
def ffmpeg(part, cid, time: dict, end, title, filenamemuster, fileformat):
asyncio.run(main(
part, cid, time, end, title, filenamemuster, fileformat
))