Kiwy-Signage/linux/video_normalizer.py

"""video_normalizer.py — downscale oversized videos for Raspberry Pi playback.

The problem
-----------
The Pi 4 has no usable 4K decode path. Measured on the target device:

    sample-30s.mp4 (1920x1080)          decode 3.03x realtime   ✅
    16118765_3840_2160_30fps.mp4 (4K)   decode 0.90x realtime   ❌

ffpyplayer decodes in software (there is no hardware H.264 *decode* wired into
its pipeline), so a 4K clip cannot be decoded fast enough to feed the screen in
real time. The visible symptom is a video that shows one frame and then sits
still, or stutters badly, while the playlist timer ticks on.

The fix
-------
Normalise oversized media to at most 1920x1080 **once, at sync time**, and hand
the player the smaller file. Playback then always runs against a resolution the
Pi can decode comfortably.

Timing (measured, 18.3s 4K clip -> 1080p):

    hardware encode (h264_v4l2m2m)   31s   one-off, at download time
    software encode (libx264)        much slower

31 s is a real cost, but it is paid **once per file** during the playlist sync —
which already runs on a worker thread and already downloads tens of megabytes.
It is never paid during playback, which is the only place it would matter.

Hardware encoding is used when available because the Pi 4's H.264 *encoder* is a
separate block from its decoder and works well; software encoding of 4K on this
SoC is slow enough to be impractical.

Scope
-----
Triggered by **resolution only** — ``width > max_width or height > max_height``.
A video already within bounds is left untouched (byte-identical), so this never
degrades content that already plays.

Audio is preserved: if the source has an audio track it is copied through
(``-c:a copy``, falling back to AAC). That matters because a *silent* video
triggers a separate bug in the SDL2_mixer path — see ``_video_has_audio`` in
``src/main.py`` — so we must not accidentally create one.
"""

from __future__ import annotations

import json
import os
import shutil
import subprocess
import sys
import time

# Importing kivy.logger installs Kivy's own argument parser, which then rejects
# this module's CLI flags ("option --dry-run not recognized") and exits. Setting
# this BEFORE the import keeps Kivy out of argv handling. It must precede the
# kivy import below, hence the import-order exception.
os.environ.setdefault('KIVY_NO_ARGS', '1')

try:
    from kivy.logger import Logger
except Exception:  # pragma: no cover - importable without Kivy (tests/CLI)
    class Logger:  # type: ignore
        @staticmethod
        def _noop(*args, **kwargs):
            pass
        info = debug = warning = error = staticmethod(_noop)

# The player and the normaliser must agree on the on-disk contract for
# "is this converting / has it been converted". That contract lives in one
# place (src/media_state.py) and is shared rather than reimplemented.
_SRC_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'src')
if _SRC_DIR not in sys.path:
    sys.path.insert(0, _SRC_DIR)

import media_state  # noqa: E402


def _log(message, level='info'):
    try:
        getattr(Logger, level, Logger.info)(f'[VideoNormalizer] {message}')
    except Exception:
        pass


#: Default output ceiling. The Pi's practical decode limit; matches the
#: ``max_resolution`` default used elsewhere in the project.
DEFAULT_MAX_WIDTH = 1920
DEFAULT_MAX_HEIGHT = 1080

#: Marker suffix written next to a normalised file, recording what was done.
#: Re-exported from :mod:`media_state` so the player and normaliser cannot drift.
MARKER_SUFFIX = media_state.MARKER_SUFFIX

#: Encoder preference: the Pi's hardware block first, then a CPU fallback.
#: ``h264_v4l2m2m`` is the V4L2 mem2mem H.264 encoder (verified working on the
#: target Pi 4); libx264 is the portable fallback for other Linux hosts.
HW_ENCODER = 'h264_v4l2m2m'
SW_ENCODER = 'libx264'

#: Bitrate for the normalised output. 1080p signage at ~5 Mbps is visually
#: lossless for this use and keeps files small.
TARGET_BITRATE = '5M'
MAX_BITRATE = '8M'
BUFSIZE = '10M'


def _run(args, timeout=30):
    """Run a command; return (returncode, stdout+stderr). Never raises."""
    try:
        result = subprocess.run(
            args, capture_output=True, text=True, timeout=timeout, check=False,
        )
        return result.returncode, (result.stdout or '') + (result.stderr or '')
    except FileNotFoundError:
        return 127, 'not found'
    except subprocess.TimeoutExpired:
        return 124, 'timeout'
    except Exception as exc:
        return 1, str(exc)


def find_ffprobe():
    return shutil.which('ffprobe')


def find_ffmpeg():
    return shutil.which('ffmpeg')


def probe_video(path):
    """Return a dict describing ``path``, or None when it cannot be probed.

    Keys: width, height, codec, duration, has_audio, pix_fmt, level.
    """
    ffprobe = find_ffprobe()
    if not ffprobe or not os.path.isfile(path):
        return None

    code, out = _run([
        ffprobe, '-v', 'error', '-print_format', 'json',
        '-show_streams', '-show_format', path,
    ], timeout=20)
    if code != 0:
        _log(f'ffprobe failed for {os.path.basename(path)}: {out.strip()[:200]}',
             'warning')
        return None

    try:
        data = json.loads(out)
    except Exception:
        return None

    info = {
        'width': None, 'height': None, 'codec': None, 'pix_fmt': None,
        'level': None, 'duration': None, 'has_audio': False,
    }
    for stream in data.get('streams', []):
        if stream.get('codec_type') == 'video' and info['width'] is None:
            info['width'] = stream.get('width')
            info['height'] = stream.get('height')
            info['codec'] = stream.get('codec_name')
            info['pix_fmt'] = stream.get('pix_fmt')
            info['level'] = stream.get('level')
        elif stream.get('codec_type') == 'audio':
            info['has_audio'] = True

    fmt = data.get('format', {})
    try:
        info['duration'] = float(fmt.get('duration'))
    except (TypeError, ValueError):
        info['duration'] = None
    return info


def hardware_encoder_available():
    """True when the Pi's V4L2 H.264 encoder can actually be opened.

    Checked by running a tiny real encode rather than by grepping
    ``ffmpeg -encoders``: the encoder is listed on builds where the kernel
    device is missing or busy, and a failed encode at sync time would be far
    worse than a slightly slower software one.
    """
    ffmpeg = find_ffmpeg()
    if not ffmpeg:
        return False
    code, _ = _run([
        ffmpeg, '-hide_banner', '-loglevel', 'error',
        '-f', 'lavfi', '-i', 'testsrc=size=320x240:rate=10:duration=0.5',
        '-c:v', HW_ENCODER, '-f', 'null', '-',
    ], timeout=60)
    return code == 0


def needs_normalization(path, max_width=DEFAULT_MAX_WIDTH,
                        max_height=DEFAULT_MAX_HEIGHT):
    """(bool, info) — True when ``path`` exceeds the playback ceiling.

    Only resolution is considered. A file at or below the ceiling is returned
    untouched so content that already plays is never re-encoded.
    """
    info = probe_video(path)
    if info is None:
        return False, None
    width, height = info.get('width'), info.get('height')
    if not width or not height:
        return False, info
    return (width > max_width or height > max_height), info


def _target_size(width, height, max_width, max_height):
    """Scale ``width``x``height`` to fit the ceiling, preserving aspect ratio.

    Dimensions are forced even: H.264 4:2:0 requires even width and height, and
    an odd value makes the encoder fail outright.
    """
    scale = min(max_width / width, max_height / height)
    new_w = int(width * scale)
    new_h = int(height * scale)
    # Round down to even numbers (never up — that could exceed the ceiling).
    new_w -= new_w % 2
    new_h -= new_h % 2
    return max(2, new_w), max(2, new_h)


def normalized_path(original_path, max_width=DEFAULT_MAX_WIDTH,
                    max_height=DEFAULT_MAX_HEIGHT):
    """Deterministic output path for the normalised version of a file.

    Kept next to the original (not in a cache dir) so the existing media
    clean-up logic, which prunes unreferenced files under ``media/``, can do its
    normal job: the playlist ends up referencing the normalised file and the
    oversized original is pruned automatically.
    """
    directory, name = os.path.split(original_path)
    stem, ext = os.path.splitext(name)
    return os.path.join(
        directory, f'{stem}_kiwy{max_height}p{ext or ".mp4"}'
    )


def normalize_video(path, max_width=DEFAULT_MAX_WIDTH,
                    max_height=DEFAULT_MAX_HEIGHT, force=False, dry_run=False):
    """Downscale ``path`` if it exceeds the ceiling.

    Returns a dict:
        {'status': ..., 'output': ..., 'info': ..., 'elapsed_s': ...}

    ``status`` is one of:
        ``within_limit`` — already at or below the ceiling, untouched
        ``reused``       — a previous conversion exists and was verified
        ``normalized``   — a new file was produced
        ``failed``       — could not normalise (caller should use the original)
        ``would``        — dry_run only

    The three success states are kept distinct on purpose: reporting a
    converted file as "ok" alongside its *original* 4K dimensions reads as
    "nothing to do" and would hide a missing conversion.

    Never raises: a failure leaves the original file untouched and the caller
    keeps playing it, because a large-but-playable video is better than none.
    """
    result = {'status': 'within_limit', 'output': None, 'info': None,
              'elapsed_s': 0.0}

    if not os.path.isfile(path):
        result['status'] = 'failed'
        result['info'] = 'file not found'
        return result

    oversized, info = needs_normalization(path, max_width, max_height)
    result['info'] = info
    if not oversized:
        return result

    output = normalized_path(path, max_width, max_height)

    # Already converted in a previous sync: reuse it. ``media_state`` owns this
    # decision so the player resolves the same file, by the same rules.
    existing = media_state.normalized_file(path, max_height)
    if not force and existing:
        result['status'] = 'reused'
        result['output'] = existing
        _log(f'{os.path.basename(path)} already normalised -> '
             f'{os.path.basename(existing)}')
        return result

    new_w, new_h = _target_size(info['width'], info['height'],
                               max_width, max_height)
    source_mb = os.path.getsize(path) / 1024 / 1024
    _log(f'{os.path.basename(path)} is {info["width"]}x{info["height"]} '
         f'({source_mb:.1f} MB) — above {max_width}x{max_height}; '
         f'normalising to {new_w}x{new_h}')

    if dry_run:
        result['status'] = 'would'
        result['output'] = output
        return result

    ffmpeg = find_ffmpeg()
    if not ffmpeg:
        _log('ffmpeg not found — cannot normalise', 'warning')
        result['status'] = 'failed'
        result['info'] = 'ffmpeg not found'
        return result

    # Write to a temp file and rename on success, so a partial or failed
    # conversion can never be picked up as a valid video.
    #
    # The temp name keeps the real extension: ffmpeg infers the output muxer
    # from the filename, so a bare ".part" fails with "Unable to choose an
    # output format". The file is only ever moved to ``output`` after a
    # successful encode, so the temp name is not user-visible.
    root, ext = os.path.splitext(output)
    tmp_out = f'{root}.tmp{ext or ".mp4"}'
    try:
        if os.path.exists(tmp_out):
            os.remove(tmp_out)
    except OSError:
        pass

    audio_args = ['-c:a', 'copy'] if info.get('has_audio') else ['-an']

    # The scale filter runs on the CPU (the slow part for 4K); the encoder is
    # hardware when possible.
    vf = f'scale={new_w}:{new_h}:flags=fast_bilinear,format=yuv420p'

    def build(codec_args):
        """Assemble one ffmpeg command.

        ``-map`` is explicit so extra streams (subtitles, cover art, a second
        audio track) cannot change the output shape between runs.
        """
        cmd = [ffmpeg, '-hide_banner', '-loglevel', 'error', '-i', path,
               '-map', '0:v:0']
        if info.get('has_audio'):
            cmd += ['-map', '0:a:0?']
        cmd += ['-vf', vf] + codec_args + audio_args + [
            '-movflags', '+faststart',
            '-pix_fmt', 'yuv420p',
            '-y', tmp_out,
        ]
        return cmd

    # Try hardware first, then fall back to software. A hardware encoder that
    # is listed but unusable (device busy, kernel mismatch) fails here rather
    # than producing a broken file.
    attempts = []
    if hardware_encoder_available():
        attempts.append((
            f'hardware ({HW_ENCODER})',
            build(['-c:v', HW_ENCODER, '-b:v', TARGET_BITRATE,
                   '-maxrate', MAX_BITRATE, '-bufsize', BUFSIZE]),
        ))
    attempts.append((
        f'software ({SW_ENCODER})',
        build(['-c:v', SW_ENCODER, '-preset', 'ultrafast', '-crf', '23',
               '-maxrate', MAX_BITRATE, '-bufsize', BUFSIZE]),
    ))

    started = time.monotonic()

    # Publish the "converting" flag BEFORE the first encoder runs. The player
    # watches this marker and skips the item while it exists, so an item must be
    # flagged for the entire window in which its file is being rewritten.
    marker_written = media_state.begin_conversion(
        path, note=f'{info["width"]}x{info["height"]} -> {new_w}x{new_h}')
    if not marker_written:
        _log(f'could not write the conversion marker for '
             f'{os.path.basename(path)}; the player may try to show the file '
             f'while it is being rewritten', 'warning')

    try:
        return _run_encoders(
            attempts, path, output, tmp_out, info, source_mb, result,
            started, f'{new_w}x{new_h}')
    finally:
        # Always clear the flag, including on failure: leaving it set would
        # make the item unplayable until it went stale.
        media_state.end_conversion(path)


def _run_encoders(attempts, path, output, tmp_out, info, source_mb,
                  result, started, size_note):
    """Try each encoder in turn; move the output into place on success."""
    for label, cmd in attempts:
        code, out = _run(cmd, timeout=1800)
        if code == 0 and os.path.isfile(tmp_out) and os.path.getsize(tmp_out) > 0:
            try:
                os.replace(tmp_out, output)
            except OSError as exc:
                _log(f'could not move normalised file into place: {exc}',
                     'warning')
                result['status'] = 'failed'
                return result
            result['elapsed_s'] = time.monotonic() - started
            result['status'] = 'normalized'
            result['output'] = output

            out_mb = os.path.getsize(output) / 1024 / 1024
            _log(f'normalised with {label} in {result["elapsed_s"]:.0f}s: '
                 f'{source_mb:.1f} MB -> {out_mb:.1f} MB '
                 f'({os.path.basename(output)}) now {size_note}')

            # Metadata last: its presence is what marks the conversion complete
            # for media_state.normalized_file(), so it must never exist for a
            # half-written output.
            try:
                with open(output + MARKER_SUFFIX, 'w') as fh:
                    json.dump({
                        'source': os.path.basename(path),
                        'source_size': os.path.getsize(path),
                        'width': info['width'],
                        'height': info['height'],
                        'output_width': int(size_note.split('x')[0]),
                        'output_height': int(size_note.split('x')[1]),
                        'encoder': label,
                        'normalized_at': time.strftime('%Y-%m-%dT%H:%M:%S'),
                    }, fh, indent=2)
            except Exception:
                pass
            return result

        _log(f'{label} failed (rc={code}): {out.strip()[:300]}', 'warning')

    try:
        if os.path.exists(tmp_out):
            os.remove(tmp_out)
    except OSError:
        pass
    result['status'] = 'failed'
    result['info'] = 'all encoders failed'
    _log(f'could not normalise {os.path.basename(path)}; '
         f'the original will be used', 'warning')
    return result


def normalize_media_dirs(media_dirs, max_width=DEFAULT_MAX_WIDTH,
                         max_height=DEFAULT_MAX_HEIGHT, dry_run=False):
    """Normalise every oversized video under the given directories.

    Intended for bulk/offline use (``--all``) and for verifying an install.
    Returns a list of per-file result dicts.
    """
    results = []
    seen = set()
    for directory in media_dirs:
        if not os.path.isdir(directory):
            continue
        for root, _dirs, files in os.walk(directory):
            for name in sorted(files):
                if not name.lower().endswith(
                        ('.mp4', '.mkv', '.mov', '.webm', '.avi', '.m4v')):
                    continue
                if '_kiwy' in name or name.endswith(MARKER_SUFFIX):
                    continue  # already an output, never re-process
                path = os.path.join(root, name)
                if path in seen:
                    continue
                seen.add(path)
                size = os.path.getsize(path)
                key = (path, size)
                if key in seen:
                    continue
                results.append(normalize_video(
                    path, max_width, max_height, dry_run=dry_run))
    return results


# ── CLI ──────────────────────────────────────────────────────────────
def main():
    import argparse

    parser = argparse.ArgumentParser(
        description='Downscale oversized videos for Raspberry Pi playback.')
    parser.add_argument('paths', nargs='+',
                        help='video files or directories to inspect')
    parser.add_argument('--max-width', type=int, default=DEFAULT_MAX_WIDTH)
    parser.add_argument('--max-height', type=int, default=DEFAULT_MAX_HEIGHT)
    parser.add_argument('--dry-run', action='store_true',
                        help='report what would change, convert nothing')
    parser.add_argument('--force', action='store_true',
                        help='re-convert even if a normalised copy exists')
    args = parser.parse_args()

    print(f'max size : {args.max_width}x{args.max_height}')
    print(f'ffmpeg   : {find_ffmpeg()}')
    print(f'ffprobe  : {find_ffprobe()}')
    if not args.dry_run:
        hw = 'available' if hardware_encoder_available() else 'unavailable (will use libx264)'
        print(f'hw enc   : {hw}')
    print()

    changed = 0
    for target in args.paths:
        if os.path.isdir(target):
            results = normalize_media_dirs(
                [target], args.max_width, args.max_height, args.dry_run)
            changed += sum(1 for r in results
                           if r['status'] in ('normalized', 'would', 'reused'))
            continue

        if args.force:
            oversized, info = needs_normalization(
                target, args.max_width, args.max_height)
            if info:
                out = normalized_path(target, args.max_width, args.max_height)
                for stale in (out, out + MARKER_SUFFIX):
                    try:
                        os.remove(stale)
                    except OSError:
                        pass

        result = normalize_video(target, args.max_width, args.max_height,
                                 force=args.force, dry_run=args.dry_run)
        info = result.get('info')
        name = os.path.basename(target)
        if result['status'] == 'within_limit':
            if isinstance(info, dict) and info.get('width'):
                print(f'  ok       {name}  {info["width"]}x{info["height"]} '
                      f'(within limit)')
            else:
                print(f'  ok       {name}')
        elif result['status'] == 'reused':
            print(f'  reuse    {name}  {info["width"]}x{info["height"]} '
                  f'-> {os.path.basename(result["output"])}')
            changed += 1
        elif result['status'] == 'normalized':
            print(f'  DONE     {name} -> {os.path.basename(result["output"])} '
                  f'({result["elapsed_s"]:.0f}s)')
            changed += 1
        elif result['status'] == 'would':
            print(f'  WOULD    {name}  {info["width"]}x{info["height"]} '
                  f'-> {os.path.basename(result["output"])}')
            changed += 1
        else:
            print(f'  FAILED   {name}  {result.get("info")}')

    print(f'\n{changed} file(s) need normalisation')
    return 0


if __name__ == '__main__':
    raise SystemExit(main())