Files
media-batch-tools/tools/video_reporting.py
T

375 lines
13 KiB
Python

from __future__ import annotations
from datetime import timedelta, timezone
from tools.avisynth_render import AvisynthClipInfo
from tools.console import light_blue, light_green, light_red
from tools.video_formatting import (
audio_parts,
format_bitrate,
format_channels,
format_chroma,
format_frame_count,
format_framerate,
format_seconds,
format_size,
join_parts,
matrix_name,
rate_value,
video_parts,
)
from tools.video_inputs import VideoInput
from tools.video_outputs import output_begin_timestamp
from tools.video_probe import VideoProbe
from tools.video_timeline import OutputTimeline
def print_validation_summary(
*,
accepted: bool,
output_frame_count: int,
dropped_frame_count: int,
timeline: OutputTimeline,
clip_info: AvisynthClipInfo | None,
probe: VideoProbe,
) -> None:
if not accepted:
print(" not valid")
return
for line in format_validation_summary_lines(
output_frame_count=output_frame_count,
dropped_frame_count=dropped_frame_count,
timeline=timeline,
clip_info=clip_info,
probe=probe,
):
print(f" {line}")
if clip_info is not None:
duration = clip_info.duration_seconds
if duration is not None and abs(duration - timeline.duration_seconds) > 0.05:
print(
" warning: Avisynth playback duration differs from source-frame "
"timeline; encoding will use the validated timeline where supported"
)
def format_validation_summary_lines(
*,
output_frame_count: int,
dropped_frame_count: int,
timeline: OutputTimeline,
clip_info: AvisynthClipInfo | None,
probe: VideoProbe,
) -> list[str]:
return [
_validation_duration_line(
output_frame_count,
dropped_frame_count,
timeline,
probe,
),
f"source span: {_format_source_span(timeline)}",
f"video: {_format_validation_video(clip_info, timeline, probe)}",
f"audio: {_format_validation_audio(clip_info, probe)}",
f"timestamp: {_format_validation_timestamp(timeline, probe)}",
]
def _validation_duration_line(
output_frame_count: int,
dropped_frame_count: int,
timeline: OutputTimeline,
probe: VideoProbe,
) -> str:
if dropped_frame_count:
frame_text = (
f"{output_frame_count} script frames, {dropped_frame_count} dropped, "
f"{len(timeline.frames)} final frames"
)
else:
frame_text = f"{len(timeline.frames)} frames"
output = f"{timeline.duration_seconds:.3f}s, {frame_text}"
source_frames = probe.stream_frame_count or probe.timing.frame_count
source_duration = probe.duration_seconds
duration_changed = (
source_duration is not None
and abs(source_duration - timeline.duration_seconds) >= 0.0005
)
frames_changed = source_frames != len(timeline.frames)
if source_duration is not None and (duration_changed or frames_changed):
old_duration = f"{source_duration:.3f}s"
old_frames = f"{source_frames} frames"
output += (
f" ({light_red(old_duration) if duration_changed else old_duration}, "
f"{light_red(old_frames) if frames_changed else old_frames})"
)
return f"duration: {output}"
def _format_source_span(timeline: OutputTimeline) -> str:
frames = sorted(set(timeline.consumed_source_frames))
if not frames:
return "none"
spans: list[tuple[int, int]] = []
start = previous = frames[0]
for frame in frames[1:]:
if frame != previous + 1:
spans.append((start, previous))
start = frame
previous = frame
spans.append((start, previous))
if len(spans) == 1:
selection = _format_frame_range(*spans[0])
elif len(spans) <= 4:
selection = ", ".join(_format_frame_range(*span) for span in spans)
else:
selection = (
f"{frames[0]} -> {frames[-1]}, {len(frames)} source frames "
f"in {len(spans)} spans"
)
return f"{selection} of {timeline.source_frame_count}"
def _format_frame_range(start: int, end: int) -> str:
return str(start) if start == end else f"{start} -> {end}"
def _format_validation_video(
clip_info: AvisynthClipInfo | None,
timeline: OutputTimeline,
probe: VideoProbe,
) -> str:
output_width = clip_info.width if clip_info else probe.width
output_height = clip_info.height if clip_info else probe.height
output_chroma = format_chroma(clip_info.chroma if clip_info else probe.pixel_format)
output_depth = clip_info.bit_depth if clip_info else probe.bit_depth
output_matrix = matrix_name(timeline.matrix)
output_fps = len(timeline.frames) / timeline.duration_seconds
output = join_parts([
f"{timeline.timing_kind.upper()} {output_fps:.3f} fps",
*video_parts(
width=output_width,
height=output_height,
pixel_format=clip_info.chroma if clip_info else probe.pixel_format,
bit_depth=output_depth,
matrix=output_matrix,
),
])
source = [probe.codec or "unknown codec"]
source_fps = rate_value(probe.avg_frame_rate)
if source_fps is not None:
source_rate = f"{source_fps:.3f} fps"
if probe.timing.timing_kind != timeline.timing_kind:
source_rate = f"{probe.timing.timing_kind.upper()} {source_rate}"
source_rate = light_red(source_rate)
elif f"{source_fps:.3f}" != f"{output_fps:.3f}":
source_rate = light_red(source_rate)
source.append(source_rate)
if (probe.width, probe.height) != (output_width, output_height):
source.append(light_red(format_size(probe.width, probe.height) or "unknown size"))
source_chroma = format_chroma(probe.pixel_format)
if source_chroma and source_chroma != output_chroma:
source.append(light_red(source_chroma))
if probe.bit_depth and probe.bit_depth != output_depth:
source.append(light_red(f"{probe.bit_depth}-bit"))
source_matrix = probe.color_matrix or "unknown"
if source_matrix != output_matrix:
source.append(light_red(source_matrix))
source.append(format_bitrate(probe.video_bit_rate))
return f"{output} ({join_parts(source)})"
def _format_validation_audio(
clip_info: AvisynthClipInfo | None,
probe: VideoProbe,
) -> str:
output_has_audio = bool(clip_info and clip_info.has_audio)
if output_has_audio:
output = join_parts(
audio_parts(
channels=clip_info.audio_channels,
sample_rate=clip_info.audio_rate,
)
)
else:
output = "none"
if not probe.audio_streams:
return f"{output} (none)" if output_has_audio else output
source_stream = probe.audio_streams[0]
source = [source_stream.codec or "unknown codec"]
if not output_has_audio or source_stream.channels != clip_info.audio_channels:
source.append(light_red(format_channels(source_stream.channels) or "unknown channels"))
if not output_has_audio or source_stream.sample_rate != clip_info.audio_rate:
source.append(light_red(
f"{source_stream.sample_rate} Hz"
if source_stream.sample_rate is not None
else "unknown sample rate"
))
source.append(format_bitrate(source_stream.bit_rate))
return f"{output} ({join_parts(source)})"
def _format_validation_timestamp(
timeline: OutputTimeline,
probe: VideoProbe,
) -> str:
if probe.filename_begin is None:
return "unknown"
display_timezone = probe.source_timezone or timezone.utc
try:
output = output_begin_timestamp(probe, timeline, display_timezone)
except RuntimeError:
return "unknown"
output = _round_datetime_to_second(output)
source = _round_datetime_to_second(probe.filename_begin.astimezone(display_timezone))
result = _format_datetime_without_zone(output)
if output != source:
old: list[str] = []
if output.date() != source.date():
old.append(light_red(source.date().isoformat()))
if output.time() != source.time():
old.append(light_red(source.time().isoformat()))
result += f" ({' '.join(old)})"
return result
def _format_datetime_without_zone(value) -> str:
return f"{light_green(value.date().isoformat())} {light_blue(value.time().isoformat())}"
def print_probe_summary(inputs: list[VideoInput], probes: list[VideoProbe]) -> None:
by_path = {probe.path.resolve(): probe for probe in probes}
print("\nProbe summary:")
for item in inputs:
probe = by_path[item.path.resolve()]
print(f"\n{item.collapsed_relative}")
print(
f" duration: {format_seconds(probe.duration_seconds)}, "
f"{format_frame_count(probe.stream_frame_count, probe.timing.frame_count)}"
)
print(f" video: {_format_probe_video_line(probe)}")
if probe.audio_streams:
for index, stream in enumerate(probe.audio_streams, start=1):
print(f" audio {index}: {_format_audio_stream(stream)}")
else:
print(" audio: none")
print(f" framerate: {_format_probe_framerate(probe)}")
if probe.timing.timing_kind == "vfr":
print(_format_vfr_timing(probe))
print(f" timestamp: {_format_probe_timestamp(probe)}")
if probe.metadata_begin_tag is not None:
display_timezone = probe.source_timezone or timezone.utc
print(
" possible beginning tag: "
f"{_format_datetime(probe.metadata_begin_tag.astimezone(display_timezone))}"
)
for warning in probe.warnings:
print(f" warning: {warning}")
def _format_probe_video_line(probe: VideoProbe) -> str:
return join_parts(
video_parts(
codec=probe.codec or "unknown codec",
width=probe.width,
height=probe.height,
pixel_format=probe.pixel_format,
bit_depth=probe.bit_depth,
matrix=probe.color_matrix or "unknown",
bit_rate=probe.video_bit_rate,
)
)
def _format_audio_stream(stream) -> str:
return join_parts(
audio_parts(
codec=stream.codec or "unknown codec",
channels=stream.channels,
sample_rate=stream.sample_rate,
bit_rate=stream.bit_rate,
)
)
def _format_probe_framerate(probe: VideoProbe) -> str:
return format_framerate(probe.timing.timing_kind, probe.avg_frame_rate)
def _format_probe_timestamp(probe: VideoProbe) -> str:
if probe.metadata_end is None:
return "unknown"
display_timezone = probe.source_timezone or timezone.utc
end = _round_datetime_to_second(probe.metadata_end.astimezone(display_timezone))
if probe.filename_begin is None:
return _format_datetime(end)
begin = _round_datetime_to_second(probe.filename_begin.astimezone(display_timezone))
begin_date = light_green(begin.date().isoformat())
begin_time = light_blue(begin.time().isoformat())
end_time = light_blue(end.time().isoformat())
if begin.date() == end.date():
span = f"{begin_date} {begin_time} - {end_time}"
else:
end_date = light_green(end.date().isoformat())
span = f"{begin_date} {begin_time} - {end_date} {end_time}"
return f"{span} {light_red(_format_timezone(end))}"
def _format_vfr_timing(probe: VideoProbe) -> str:
timing = probe.timing
delta_line = (
" "
f"{timing.unique_delta_count} rounded deltas (ms): "
f"{light_green(_format_ms(timing.min_delta_ms))} min, "
f"{light_green(_format_ms(timing.median_delta_ms))} median, "
f"{light_green(_format_ms(timing.max_delta_ms))} max"
)
common = [
f"{light_green(_format_ms(delta))} {count}x"
for delta, count in timing.common_deltas_ms
]
if not common:
return delta_line
common_line = " common deltas (ms): " + ", ".join(common)
return "\n".join([delta_line, common_line])
def _format_ms(value: float | None) -> str:
if value is None:
return "unknown"
return f"{value:g}"
def _format_datetime(value) -> str:
if value is None:
return "unknown"
rounded = value
date = rounded.date().isoformat()
time = rounded.timetz().isoformat()
if rounded.tzinfo is not None:
time, tz = _split_time_zone(time)
return f"{light_green(date)} {light_blue(time)} {light_red(tz)}"
return f"{light_green(date)} {light_blue(time)}"
def _format_timezone(value) -> str:
offset = value.strftime("%z")
return f"{offset[:3]}:{offset[3:]}" if offset else ""
def _round_datetime_to_second(value):
return (value + timedelta(microseconds=500_000)).replace(microsecond=0)
def _split_time_zone(value: str) -> tuple[str, str]:
if value.endswith("Z"):
return value[:-1], "Z"
for index in range(len(value) - 6, 0, -1):
if value[index] in "+-":
return value[:index], value[index:]
return value, ""