path: root/packages/meshbay-node/src/meshbay_node/media_probe.py

"""
ffprobe wrapper shared by stream-time codec detection (transport/webrtc_server.py)
and index-time technical-field enrichment (indexer/enrich.py).

Split out of webrtc_server.py so the indexer package (which webrtc_server.py
already imports from) can call it too without a circular import.
"""

import asyncio
import json
from dataclasses import dataclass, field

_H264_PROFILES = {"Baseline": "42", "Main": "4d", "High": "64", "High 10": "6e"}

# Source video codecs whose MSE codec string is real but which no mainstream
# browser can actually decode via MediaSource on most desktop platforms — HEVC
# has no royalty-free decoder in Chrome/Firefox on Linux (and is spotty even
# on platforms with one). Found live: a real HEVC/EAC3 WEB-DL reported
# "Codec not supported for streaming: hev1.1.6.L93.B0,mp4a.40.2" from
# MediaSource.isTypeSupported, even though ffprobe/VLC play it fine. VP9/AV1
# are not in this set — those decode natively in every mainstream browser.
BROWSER_INCOMPATIBLE_VIDEO_CODECS = frozenset({"hevc"})


@dataclass(frozen=True)
class AudioTrack:
    """
    One selectable audio track.

    **`ordinal` is the position among the audio streams, not the container
    stream index**, because that is what `-map 0:a:<n>` takes. A file whose
    audio sits at container indices 1, 2 and 3 has ordinals 0, 1 and 2, and
    mapping `0:a:1` on the container index would silently serve the third
    track — the failure this field's name exists to prevent.
    """
    ordinal: int
    language: str | None
    title: str | None
    codec_name: str | None
    channels: int | None


# Subtitle codecs ffmpeg can convert to WebVTT, which is the only thing MSE
# can be given. An allow-list rather than a bitmap deny-list: the cost of
# wrongly excluding an exotic text codec is a track nobody can pick, and the
# cost of wrongly including a bitmap one is a track that is picked and then
# displays nothing, with no error to lead anyone back here.
TEXT_SUBTITLE_CODECS = frozenset({
    "subrip", "srt", "ass", "ssa", "mov_text", "webvtt", "text",
    "subviewer", "subviewer1", "sami", "realtext", "stl", "jacosub",
    "microdvd", "mpl2", "vplayer", "pjs",
})


@dataclass(frozen=True)
class SubtitleTrack:
    """
    One selectable subtitle track, guaranteed convertible to WebVTT.

    **`ordinal` counts every subtitle stream, including the bitmap ones this
    list does not carry**, because that is what `-map 0:s:<n>` counts. The
    same trap as `AudioTrack.ordinal` one level deeper: filtering the list and
    numbering the survivors would give a file whose streams are PGS, SRT, SRT
    the ordinals 0 and 1 for its two text tracks, and `-map 0:s:0` would then
    extract the PGS stream — which produces an empty WebVTT rather than an
    error, so the viewer gets a subtitle track with no subtitles in it.
    """
    ordinal: int
    language: str | None
    title: str | None
    codec_name: str | None
    # **A forced track is not a shorter version of the full one.** It carries
    # only signage and the lines spoken in another language — measured on a
    # real film: 30 cues and 77 seconds of text across 2h32, against 1559 cues
    # and 41% of the running time for the full track beside it. Picking it and
    # seeing nothing for ten minutes is the ordinary outcome, not a fault, and
    # nothing in the menu let a viewer tell those two apart. The disposition is
    # what says which it is; the title tag that would also say it ("Forced",
    # "SDH") is absent as often as it is present.
    forced: bool = False
    hearing_impaired: bool = False


@dataclass
class VideoProbe:
    """
    What one ffprobe call says about a video file.

    A dataclass rather than the tuple this used to return: the tuple had six
    positional fields, `has_audio` sat third and `raw_codec_name` sixth, and
    adding a seventh for the track list would have made every call site a
    counting exercise.
    """
    codec: str | None
    duration: float
    has_audio: bool
    width: int | None
    height: int | None
    raw_codec_name: str | None
    audio_tracks: list[AudioTrack] = field(default_factory=list)
    subtitle_tracks: list[SubtitleTrack] = field(default_factory=list)


async def probe_video(path: str) -> VideoProbe:
    """
    Probe a video file with ffprobe.

    The audio half of the codec string is always "mp4a.40.2" (AAC-LC) or
    absent — never the source's real audio codec — because the streaming
    path always transcodes audio to AAC and never copies it: MSE in every
    mainstream browser only decodes AAC/Opus, and a source codec outside
    that (AC-3, E-AC-3, DTS, ...) is at best silently unplayable and at
    worst, for E-AC-3 at least, makes ffmpeg itself refuse to write the
    fragmented MP4 header ("Cannot write moov atom before EAC3 packets
    parsed" — reproduced against a real 5.1 E-AC-3 WEB-DL). Video is copied
    whenever the browser can decode it directly; the raw codec name is
    returned alongside the MSE string so the caller can decide whether this
    source needs a real re-encode instead (BROWSER_INCOMPATIBLE_VIDEO_CODECS
    above) — the MSE string alone can't drive that decision, since it still
    faithfully reports "hev1..." for a source this pipeline cannot actually
    deliver copied.

    **A `None` codec string means "this must be re-encoded", not "this cannot
    be played".** Only the four codecs a browser can decode through MediaSource
    are mapped; everything else — MPEG-4 Part 2 (Xvid, DivX), MPEG-2, VC-1,
    WMV, Theora — has no string to report because there is no browser decoder
    to report it to, and the streaming path answers that by re-encoding to
    H264. It answered it by refusing until 2026-09-09, which read to the
    operator as a broken file rather than as an unwired code path.

    **Every audio track is reported, not just the first.** The streaming path
    transcodes audio unconditionally, so serving the second track costs exactly
    what serving the first costs and the choice is the viewer's to make; a
    library of dubbed films is one where the first track is a language half the
    group does not want. `has_audio` stays as the single question the muxing
    decisions ask, and is now `bool(audio_tracks)`.

    **Only text subtitle tracks are reported.** A library's embedded subtitles
    are roughly four-fifths text (subrip, ass) and one-fifth bitmap (PGS,
    VOBSUB); a bitmap track has no path to WebVTT without OCR, so listing one
    would offer a choice that silently displays nothing. A file whose only
    subtitles are bitmap therefore reports none at all and gets no selector,
    exactly like a file with no subtitles — which is a true statement about
    what this node can serve, not a concealed failure.

    width/height come from the same ffprobe call (one extra `-show_entries`
    field, no second process spawn) — resolution is deliberately never
    guessed from the filename (docs/MESHBAY_DESIGN.md §9.7).
    """
    from meshbay_node.platform import ffprobe_cmd
    proc = await asyncio.create_subprocess_exec(
        ffprobe_cmd(), "-v", "error",
        "-show_entries",
        "stream=codec_name,profile,level,codec_type,width,height,channels",
        "-show_entries", "stream_tags=language,title",
        "-show_entries", "stream_disposition=forced,hearing_impaired",
        "-show_entries", "format=duration",
        "-of", "json", path,
        stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
    )
    stdout, _ = await proc.communicate()
    info = json.loads(stdout)
    duration = float(info.get("format", {}).get("duration", 0))

    v_codec = ""
    raw_codec_name: str | None = None
    width: int | None = None
    height: int | None = None
    audio_tracks: list[AudioTrack] = []
    subtitle_tracks: list[SubtitleTrack] = []
    subtitle_streams_seen = 0
    for s in info.get("streams", []):
        if s.get("codec_type") == "video" and not v_codec:
            cn = s.get("codec_name", "")
            raw_codec_name = cn or None
            if cn == "h264":
                p = _H264_PROFILES.get(s.get("profile", "High"), "64")
                lvl = int(s.get("level", 40))
                v_codec = f"avc1.{p}00{lvl:02x}"
            elif cn == "hevc":
                v_codec = "hev1.1.6.L93.B0"
            elif cn == "vp9":
                v_codec = "vp09.00.10.08"
            elif cn == "av1":
                v_codec = "av01.0.01M.08"
            width = s.get("width")
            height = s.get("height")
        elif s.get("codec_type") == "audio":
            tags = s.get("tags") or {}
            audio_tracks.append(AudioTrack(
                # Counted here, never read from `s["index"]` — see AudioTrack.
                ordinal=len(audio_tracks),
                language=(tags.get("language") or "").strip() or None,
                title=(tags.get("title") or "").strip() or None,
                codec_name=s.get("codec_name") or None,
                channels=s.get("channels"),
            ))
        elif s.get("codec_type") == "subtitle":
            # Counted before the filter, never after — see SubtitleTrack.
            ordinal = subtitle_streams_seen
            subtitle_streams_seen += 1
            codec_name = (s.get("codec_name") or "").strip() or None
            if codec_name not in TEXT_SUBTITLE_CODECS:
                continue
            tags = s.get("tags") or {}
            disp = s.get("disposition") or {}
            subtitle_tracks.append(SubtitleTrack(
                ordinal=ordinal,
                language=(tags.get("language") or "").strip() or None,
                title=(tags.get("title") or "").strip() or None,
                codec_name=codec_name,
                forced=bool(disp.get("forced")),
                hearing_impaired=bool(disp.get("hearing_impaired")),
            ))

    has_audio = bool(audio_tracks)
    codec = None
    if v_codec:
        codec = f"{v_codec},mp4a.40.2" if has_audio else v_codec
    return VideoProbe(
        codec=codec,
        duration=duration,
        has_audio=has_audio,
        width=width,
        height=height,
        raw_codec_name=raw_codec_name,
        audio_tracks=audio_tracks,
        subtitle_tracks=subtitle_tracks,
    )