This is a follow up article of my AV1 video encoding Batch File. This Python Code is the improved version with extra arguments after hours of trial and error. Enjoy~
My external HDD was getting full, so I deleted stuff, moved stuff, and compressed some ...stuff. Most importantly, I converted my old Let's Play videos into compressed version, with my research on AV1 Encoding, I developed (hint: vibecoding with AI) and code that does the compression process for me.
This code converts .mp4 files to .mkv files and vice-versa. The first version was a batch file with limited options, but the version of this article is a full python code that takes multiple arguments to allow better control for the quality of the generated video.
This code uses smarter pre-processing, so while the generated AV1 videos quality drops, I can control how much I want it to drop. I made sure to have multiple arguments and a --help argument that explains how they work:
The sheer volume of videos I had to convert was massive, though. Over 300 hours of footage. The conversion on my PC takes about 9 minutes for an hour long video (at 720p) and twice as long in larger videos (1080p.) Because of that, I decided to drop the frame rate for some of the videos I needed less to 24fps, both for smaller file sizes and shorter conversion time.
I still had to go through hours of trails and error to arrive at the best quality-size trade-off for each of my video series...
Converting to AV1 using AMD's HW encoder leads to bigger sized files compared to H264's software encoder at the "slow" preset. Using lower QVBR quality leads to smaller sizes, but the quality drops noticeably if I aimed for file sizes that small... That surprised me because AV1 should be more than 50% efficient than that. But one possible reason is that I'm converting files that are already heavily encoded, or maybe that's the difference between HW and SW encoding.
...I guess I still have a lot to learn.
What do you think?
@ahmadmanga/re-leothreads-ytupx7sx
#!/usr/bin/env python3
"""
AV1 Batch Converter for MKV/MP4 Files
"""
import os
import sys
import re
import subprocess
import argparse
from collections import Counter
from datetime import datetime
from pathlib import Path
PRESET_CONFIGS = {
'high_quality': {
'usage': 'high_quality',
'preset': 'high_quality',
'preencode': 'true',
'preanalysis': 'true',
'pa_lookahead_buffer_depth': '40',
'pa_taq_mode': '2',
'aq_mode': 'caq',
'pa_caq_strength': 'high',
'pa_scene_change_detection_enable': 'true',
'pa_scene_change_detection_sensitivity': 'high',
'pa_adaptive_mini_gop': 'true',
'pa_ltr_enable': 'true',
'high_motion_quality_boost_enable': 'true',
'async_depth': '42',
},
'balanced': {
'usage': 'transcoding',
'preset': 'balanced',
'preencode': 'false',
'preanalysis': 'true',
'pa_lookahead_buffer_depth': '20',
'pa_taq_mode': '1',
'aq_mode': 'caq',
'pa_caq_strength': 'medium',
'pa_scene_change_detection_enable': 'true',
'pa_scene_change_detection_sensitivity': 'medium',
'pa_adaptive_mini_gop': 'true',
'pa_ltr_enable': 'true',
'high_motion_quality_boost_enable': 'false',
'async_depth': '22',
}
}
# B-frame settings (fallback for hardware that supports it)
BFRAME_CONFIG = {
'enabled': {
'bf': '3',
'pa_adaptive_mini_gop': 'false',
'pa_ltr_enable': 'false',
},
'disabled': {
'bf': '0',
}
}
def parse_args():
parser = argparse.ArgumentParser(
description='AV1 Batch Converter for MKV/MP4 Files',
formatter_class=argparse.RawTextHelpFormatter
)
parser.add_argument('-q', '--qvbr', type=int, default=30,
help='QVBR quality level (default: 30)')
parser.add_argument('-r', '--resolution', type=int, default=None,
help='Target resolution height in pixels (e.g., 720 for 720p)')
parser.add_argument('-f', '--framerate', type=float, default=None,
help='Maximum output framerate (e.g., 30 for 30fps).\n'
'Videos with lower or equal framerate are unchanged.')
parser.add_argument('-p', '--preset', choices=['balanced', 'high_quality'],
default='high_quality',
help='Encoding preset: balanced (faster) or high_quality (default)')
parser.add_argument('-b', '--bframes', action='store_true',
help='Enable B-frames. Requires hardware support.\n'
'Automatically disables LTR and adaptive mini-GOP.')
parser.add_argument('-n', '--newfolder', action='store_true',
help='Output files into a new subfolder named after the current folder')
parser.add_argument('-c', '--crop', nargs='?', const='auto', default=None,
help=(
'Crop the video. Three modes:\n'
' -c Auto-detect black bars by sampling 1 frame every 15s.\n'
' Uses luma threshold 16 (near-pure black only).\n'
' No crop is applied if bars are not consistently detected.\n'
' -c w:h Crop to w x h pixels, automatically centered.\n'
' x and y offsets are calculated from the video dimensions.\n'
' -c w:h:x:y Crop to w x h pixels at exact offset x from left, y from top.\n'
'\n'
' Examples:\n'
' -c crop auto-detect\n'
' -c 1920:800 crop to 1920x800, centered\n'
' -c 1920:800:0:140 crop to 1920x800 at offset 0,140\n'
))
return parser.parse_args()
def get_video_duration(filepath):
"""Get video duration in seconds using ffprobe."""
try:
result = subprocess.run(
['ffprobe', '-v', 'error', '-show_entries', 'format=duration',
'-of', 'default=noprint_wrappers=1:nokey=1', str(filepath)],
capture_output=True, text=True, check=True
)
return float(result.stdout.strip())
except Exception:
return None
def get_video_dimensions(filepath):
"""Get video width and height using ffprobe."""
try:
result = subprocess.run(
['ffprobe', '-v', 'error', '-select_streams', 'v:0',
'-show_entries', 'stream=width,height',
'-of', 'csv=p=0', str(filepath)],
capture_output=True, text=True, check=True
)
w, h = result.stdout.strip().split(',')
return int(w), int(h)
except Exception:
return None, None
def get_video_framerate(filepath):
"""Get video framerate as a float using ffprobe."""
try:
result = subprocess.run(
['ffprobe', '-v', 'error', '-select_streams', 'v:0',
'-show_entries', 'stream=r_frame_rate',
'-of', 'default=noprint_wrappers=1:nokey=1', str(filepath)],
capture_output=True, text=True, check=True
)
fr_str = result.stdout.strip()
if '/' in fr_str:
num, den = fr_str.split('/')
return float(num) / float(den)
return float(fr_str)
except Exception:
return None
def detect_crop(input_path, duration):
"""
Detect black bars by sampling one frame every 15 seconds with luma threshold 16.
Returns a crop filter string like 'crop=1920:800:0:140' or None if no crop needed.
"""
sample_times = list(range(15, int(duration), 15))
if not sample_times:
sample_times = [max(1, int(duration) // 2)]
print(f" Crop detect: sampling {len(sample_times)} point(s)...", end="", flush=True)
crop_values = []
for t in sample_times:
try:
result = subprocess.run(
['ffmpeg', '-ss', str(t), '-i', str(input_path),
'-vf', 'cropdetect=16:16:0', '-frames:v', '2',
'-f', 'null', '-'],
capture_output=True, text=True
)
for line in result.stderr.split('\n'):
if 'crop=' in line:
match = re.search(r'crop=(\d+:\d+:\d+:\d+)', line)
if match:
crop_values.append(match.group(1))
except Exception:
continue
if not crop_values:
print(" no data found.")
return None
counter = Counter(crop_values)
most_common_crop, _ = counter.most_common(1)[0]
orig_w, orig_h = get_video_dimensions(input_path)
if orig_w is not None:
if most_common_crop == f"{orig_w}:{orig_h}:0:0":
print(" no black bars detected.")
return None
parts = most_common_crop.split(':')
print(f" {parts[0]}x{parts[1]} (offset {parts[2]},{parts[3]})")
return f"crop={most_common_crop}"
def resolve_crop(crop_arg, input_path, duration):
"""
Resolve the -c/--crop argument to a crop filter string.
'auto' -> run auto-detect
'w:h' -> center crop, calculate x and y from video dimensions
'w:h:x:y' -> exact crop at given offsets
Returns a filter string like 'crop=1920:800:0:140' or None.
"""
if crop_arg == 'auto':
if duration:
return detect_crop(input_path, duration)
else:
print(" Crop detect: skipped (could not determine duration)")
return None
match_full = re.match(r'^(\d+):(\d+):(\d+):(\d+)$', crop_arg)
if match_full:
w, h, x, y = match_full.groups()
print(f" Crop: {w}x{h} at offset {x},{y} (manual)")
return f"crop={w}:{h}:{x}:{y}"
match_wh = re.match(r'^(\d+):(\d+)$', crop_arg)
if match_wh:
w, h = int(match_wh.group(1)), int(match_wh.group(2))
orig_w, orig_h = get_video_dimensions(input_path)
if orig_w is None:
print(" Crop: skipped (could not determine video dimensions)")
return None
x = (orig_w - w) // 2
y = (orig_h - h) // 2
print(f" Crop: {w}x{h} centered (offset {x},{y})")
return f"crop={w}:{h}:{x}:{y}"
print(f" Crop: invalid format '{crop_arg}', skipping")
return None
def build_video_filter(crop_filter, resolution, target_framerate):
"""Build the complete video filter chain: crop -> scale -> fps -> denoise."""
filters = []
if crop_filter:
filters.append(crop_filter)
if resolution:
filters.append(f"scale=-2:{resolution}")
if target_framerate:
filters.append(f"fps={target_framerate}")
filters.append("hqdn3d=2:2:0:0")
return ",".join(filters)
def format_duration(seconds):
"""Format seconds into minutes and seconds."""
if seconds is None:
return None
mins = int(seconds // 60)
secs = int(seconds % 60)
return f"{mins} min {secs} sec"
def format_size(bytes_size):
"""Format bytes to MB (binary)."""
return round(bytes_size / 1048576, 2)
def copy_file_timestamps(src_path, dst_path):
"""
Copy file timestamps from source to destination.
Handles both modification time (cross-platform) and creation time (Windows only).
"""
try:
src_stat = os.stat(src_path)
os.utime(dst_path, (src_stat.st_atime, src_stat.st_mtime))
except Exception:
pass
if sys.platform == 'win32':
try:
src_stat = os.stat(src_path)
filetime = int(src_stat.st_ctime * 1e7 + 116444736000000000)
subprocess.run(
['powershell', '-command',
f'$dst = Get-Item -LiteralPath \'{dst_path}\'; '
f'$dst.CreationTime = [DateTime]::FromFileTime({filetime})'],
capture_output=True, check=False
)
except Exception:
pass
def build_ffmpeg_cmd(input_path, output_path, qvbr_quality, video_filter, preset_config, use_bframes):
"""Build the ffmpeg command list from preset config."""
p = preset_config
# Merge bframe config based on hardware support
if use_bframes:
bframe_settings = BFRAME_CONFIG['enabled']
else:
bframe_settings = BFRAME_CONFIG['disabled']
cmd = [
'ffmpeg', '-y', '-xerror',
'-i', str(input_path),
'-map_metadata', '0',
'-vf', video_filter,
'-c:v', 'av1_amf',
'-usage', p['usage'],
'-rc', 'qvbr',
'-qvbr_quality_level', str(qvbr_quality),
'-preset', p['preset'],
'-aq_mode', p['aq_mode'],
'-preencode', p['preencode'],
'-preanalysis', p['preanalysis'],
'-pa_lookahead_buffer_depth', p['pa_lookahead_buffer_depth'],
'-pa_taq_mode', p['pa_taq_mode'],
'-pa_caq_strength', p['pa_caq_strength'],
'-pa_scene_change_detection_enable', p['pa_scene_change_detection_enable'],
'-pa_scene_change_detection_sensitivity', p['pa_scene_change_detection_sensitivity'],
'-pa_adaptive_mini_gop', bframe_settings.get('pa_adaptive_mini_gop', p['pa_adaptive_mini_gop']),
'-pa_ltr_enable', bframe_settings.get('pa_ltr_enable', p['pa_ltr_enable']),
'-high_motion_quality_boost_enable', p['high_motion_quality_boost_enable'],
'-async_depth', p['async_depth'],
'-bf', bframe_settings['bf'],
'-g', '180',
'-c:a', 'copy',
str(output_path)
]
return cmd
def convert_file(input_path, output_path, qvbr_quality, resolution, framerate, preset_config, log_file, crop_arg, use_bframes):
"""Convert a single video file to AV1."""
input_size = os.path.getsize(input_path)
input_mb = format_size(input_size)
duration = get_video_duration(input_path)
input_fps = get_video_framerate(input_path)
print(f"\n{'='*44}")
print(f"Converting: {input_path.name}")
print(f"{'='*44}")
print(f"Input size: {input_mb} MB")
if duration:
print(f"Duration: {format_duration(duration)}")
if input_fps:
print(f"Input FPS: {input_fps:.2f}")
crop_filter = None
if crop_arg is not None:
crop_filter = resolve_crop(crop_arg, input_path, duration)
target_fps = None
if framerate is not None and input_fps is not None:
if input_fps > framerate:
target_fps = framerate
print(f" Framerate: {input_fps:.2f} fps -> {framerate} fps (reduced)")
else:
print(f" Framerate: {input_fps:.2f} fps (unchanged, <= {framerate})")
video_filter = build_video_filter(crop_filter, resolution, target_fps)
print("Encoding in progress...\n")
log_file.write(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] START: {input_path.name}\n")
if crop_filter:
log_file.write(f" Crop: {crop_filter}\n")
if target_fps:
log_file.write(f" Framerate: {input_fps:.2f} fps -> {target_fps} fps\n")
log_file.flush()
cmd = build_ffmpeg_cmd(input_path, output_path, qvbr_quality, video_filter, preset_config, use_bframes)
start_time = datetime.now()
result = subprocess.run(cmd, capture_output=False)
end_time = datetime.now()
encode_seconds = (end_time - start_time).total_seconds()
if result.returncode != 0:
print(f"\n[ERROR] FFmpeg failed on {input_path.name}")
if output_path.exists():
output_path.unlink()
log_file.write(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] ERROR: {input_path.name} - Conversion failed\n")
log_file.write("-" * 44 + "\n\n")
log_file.flush()
return False, 0, 0, 0
output_size = os.path.getsize(output_path)
output_mb = format_size(output_size)
ratio = round(input_size / output_size, 2) if output_size > 0 else 0
saved_mb = round((input_size - output_size) / 1048576, 2)
saved_pct = round((1 - output_size / input_size) * 100, 1) if input_size > 0 else 0
encode_min = int(encode_seconds // 60)
encode_sec_rem = int(encode_seconds % 60)
speed = None
bitrate_kbps = None
if duration and encode_seconds > 0:
speed = round(duration / encode_seconds, 2)
bitrate_kbps = round((output_size * 8 / duration) / 1000, 0)
copy_file_timestamps(input_path, output_path)
print(f"\n{'-'*44}")
print(f"[DONE] {input_path.name} -> {output_path.name}")
print(f"{'-'*44}")
print(f"Input: {input_mb} MB")
print(f"Output: {output_mb} MB")
print(f"Ratio: {ratio}x ({saved_pct}% smaller)")
print(f"Saved: {saved_mb} MB")
print(f"Time: {encode_min} min {encode_sec_rem} sec")
if speed:
print(f"Speed: {speed}x realtime")
if bitrate_kbps:
print(f"Bitrate: {bitrate_kbps} kbps")
if crop_filter:
print(f"Crop: {crop_filter}")
if target_fps:
print(f"Framerate: {input_fps:.2f} fps -> {target_fps} fps")
print("-" * 44)
log_file.write(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] DONE: {input_path.name} -> {output_path.name}\n")
log_file.write(f" Input: {input_mb} MB\n")
log_file.write(f" Output: {output_mb} MB\n")
log_file.write(f" Ratio: {ratio}x ({saved_pct}% smaller)\n")
log_file.write(f" Saved: {saved_mb} MB\n")
log_file.write(f" Time: {encode_min} min {encode_sec_rem} sec\n")
if speed:
log_file.write(f" Speed: {speed}x realtime\n")
if bitrate_kbps:
log_file.write(f" Bitrate: {bitrate_kbps} kbps\n")
if crop_filter:
log_file.write(f" Crop: {crop_filter}\n")
if target_fps:
log_file.write(f" Framerate: {input_fps:.2f} fps -> {target_fps} fps\n")
log_file.write("-" * 44 + "\n\n")
log_file.flush()
return True, input_size, output_size, encode_seconds
def collect_files(cwd, output_dir, use_newfolder):
"""
Collect files to convert and files to skip.
Returns (files_to_convert, skipped_list).
"""
files_to_convert = []
skipped_list = []
if use_newfolder:
for f in sorted(cwd.glob("*.mkv")):
out = output_dir / f.with_suffix(".mp4").name
if out.exists():
skipped_list.append(out.name)
else:
files_to_convert.append((f, out))
for f in sorted(cwd.glob("*.mp4")):
out = output_dir / f.with_suffix(".mkv").name
if out.exists():
skipped_list.append(out.name)
else:
files_to_convert.append((f, out))
else:
for f in sorted(cwd.glob("*.mkv")):
out = f.with_suffix(".mp4")
if out.exists():
skipped_list.append(out.name)
else:
files_to_convert.append((f, out))
for f in sorted(cwd.glob("*.mp4")):
out = f.with_suffix(".mkv")
if out.exists():
skipped_list.append(out.name)
else:
files_to_convert.append((f, out))
return files_to_convert, skipped_list
def main():
args = parse_args()
preset_config = PRESET_CONFIGS[args.preset]
cwd = Path.cwd()
folder_name = cwd.name
if args.newfolder:
output_dir = cwd / folder_name
output_dir.mkdir(exist_ok=True)
else:
output_dir = cwd
timestamp = datetime.now().strftime('%Y-%m-%d-%H-%M')
log_path = cwd / f"VidConvert-Log-{folder_name}-{timestamp}.txt"
if args.crop is None:
crop_display = "Off"
elif args.crop == 'auto':
crop_display = "Auto-detect (every 15s, luma threshold 16)"
else:
crop_display = f"Manual ({args.crop})"
bframe_display = "Enabled" if args.bframes else "Disabled (LTR + adaptive mini-GOP)"
print("=" * 44)
print(" AV1 Batch Converter for MKV/MP4 Files")
print("=" * 44)
print()
print("Settings:")
print(f" QVBR Quality: {args.qvbr}")
print(f" Preset: {args.preset}")
print(f" B-frames: {bframe_display}")
if args.resolution:
print(f" Resolution: {args.resolution}p (scaled)")
else:
print(" Resolution: Original")
if args.framerate:
print(f" Max FPS: {args.framerate} (reduce higher framerates)")
else:
print(" Framerate: Original")
if args.newfolder:
print(f" Output dir: ./{folder_name}/")
else:
print(" Output dir: Same folder")
print(f" Crop: {crop_display}")
print(f" Log File: {log_path.name}")
print()
files_to_convert, skipped_list = collect_files(cwd, output_dir, args.newfolder)
for name in skipped_list:
print(f"[SKIP] {name} already exists")
skipped = len(skipped_list)
total = len(files_to_convert)
if total == 0:
print("\nNo files need conversion.")
return
print(f"\nFiles to convert: {total}")
print()
completed = 0
errors = 0
total_input_bytes = 0
total_output_bytes = 0
total_encode_seconds = 0
with open(log_path, 'a', encoding='utf-8') as log_file:
log_file.write("=" * 44 + "\n")
log_file.write(" AV1 Batch Converter Log\n")
log_file.write("=" * 44 + "\n")
log_file.write(f"Date: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
log_file.write(f"Quality Level: {args.qvbr}\n")
log_file.write(f"Preset: {args.preset}\n")
log_file.write(f"B-frames: {bframe_display}\n")
if args.resolution:
log_file.write(f"Resolution: {args.resolution}p\n")
if args.framerate:
log_file.write(f"Max FPS: {args.framerate}\n")
log_file.write(f"Output dir: {'.' + os.sep + folder_name + os.sep if args.newfolder else 'same folder'}\n")
log_file.write(f"Crop: {crop_display}\n")
log_file.write("\n")
log_file.flush()
for idx, (input_path, output_path) in enumerate(files_to_convert, 1):
print(f"\n[{idx}/{total}]", end="")
success, input_size, output_size, encode_seconds = convert_file(
input_path, output_path, args.qvbr, args.resolution,
args.framerate, preset_config, log_file, args.crop, args.bframes
)
if success:
completed += 1
total_input_bytes += input_size
total_output_bytes += output_size
total_encode_seconds += encode_seconds
else:
errors += 1
remaining = total - completed - errors
print(f"\nProgress: Completed={completed} Remaining={remaining} Errors={errors}")
if completed > 0 and completed % 3 == 0:
print()
print("*" * 44)
print(f" STATUS: {completed} done, {remaining} remaining, {errors} errors")
print("*" * 44)
print()
print()
print("=" * 44)
print(" CONVERSION COMPLETE")
print("=" * 44)
print(f" Files converted: {completed}")
print(f" Files skipped: {skipped}")
print(f" Files failed: {errors}")
print()
if completed > 0:
total_input_mb = round(total_input_bytes / 1048576, 0)
total_output_mb = round(total_output_bytes / 1048576, 0)
total_saved_mb = round((total_input_bytes - total_output_bytes) / 1048576, 0)
total_ratio = round(total_input_bytes / total_output_bytes, 2) if total_output_bytes > 0 else 0
total_pct = round((1 - total_output_bytes / total_input_bytes) * 100, 1) if total_input_bytes > 0 else 0
total_hours = int(total_encode_seconds // 3600)
total_mins = int((total_encode_seconds % 3600) // 60)
total_secs = int(total_encode_seconds % 60)
print(" TOTAL COMPRESSION SUMMARY")
print(" " + "-" * 42)
print(f" Input size: {int(total_input_mb)} MB")
print(f" Output size: {int(total_output_mb)} MB")
print(f" Space saved: {int(total_saved_mb)} MB")
print(f" Ratio: {total_ratio}x")
print(f" Reduction: {total_pct}%")
print(f" Total time: {total_hours}h {total_mins}m {total_secs}s")
print("=" * 44)
log_file.write("=" * 44 + "\n")
log_file.write(" FINAL SUMMARY\n")
log_file.write("=" * 44 + "\n")
log_file.write(f"Files converted: {completed}\n")
log_file.write(f"Files skipped: {skipped}\n")
log_file.write(f"Files failed: {errors}\n")
log_file.write(f"Input size: {int(total_input_mb)} MB\n")
log_file.write(f"Output size: {int(total_output_mb)} MB\n")
log_file.write(f"Space saved: {int(total_saved_mb)} MB\n")
log_file.write(f"Ratio: {total_ratio}x\n")
log_file.write(f"Reduction: {total_pct}%\n")
log_file.write(f"Total time: {total_hours}h {total_mins}m {total_secs}s\n")
log_file.write("=" * 44 + "\n")
print(f"\nLog saved to: {log_path.name}")
print("\nPress Enter to close...")
input()
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
print("\n\nConversion interrupted by user.")
sys.exit(1)