denki 0.2.0

//! RTSP MJPEG snapshot helper.
//!
//! Captures a single JPEG frame from a Tapo camera's MJPEG RTSP stream.
//!
//! Tapo PTZ cameras (verified on the C220) advertise an MJPEG profile but
//! deviate from [RFC 2435](https://www.rfc-editor.org/rfc/rfc2435.txt): the
//! payload of every RTP packet (after the 8-byte RTP/JPEG main header and any
//! optional sub-headers) carries a slice of an already-complete JFIF file
//! (`FF D8` ... `FF D9`, including DQT/SOF/DHT) rather than the bare entropy-
//! coded scan that the RFC requires. Retina's `Depacketizer` faithfully
//! reconstructs JPEG headers from the RTP/JPEG main header fields and prepends
//! them to the body, which here produces a malformed double-headered JPEG that
//! decoders reject (the symptom is a near-black image with a thin band of
//! valid pixels at the top).
//!
//! To avoid that, this module subscribes to raw RTP packets and concatenates
//! the post-header bytes of all fragments belonging to one frame. The result
//! is the camera's original JPEG, byte-for-byte.

use std::time::Duration;

use anyhow::{Context, anyhow};
use retina::client::{
    Credentials, PacketItem, PlayOptions, Session, SessionOptions, SetupOptions,
    TcpTransportOptions, Transport,
};
use tokio_stream::StreamExt;

use crate::tapo_vendor::error::Error;

/// Capture a single JPEG frame from an MJPEG-encoded RTSP stream.
///
/// `timeout` covers the whole capture, from the RTSP DESCRIBE to the last
/// fragment of the frame. Retina sets no timeouts of its own.
pub(crate) async fn grab_mjpeg_frame(
    url: &str,
    creds: Credentials,
    timeout: Duration,
) -> Result<Vec<u8>, Error> {
    tokio::time::timeout(timeout, grab_frame(url, creds))
        .await
        .map_err(|_| anyhow!("RTSP snapshot timed out after {timeout:?}"))?
        .map_err(Into::into)
}

async fn grab_frame(url: &str, creds: Credentials) -> anyhow::Result<Vec<u8>> {
    let parsed_url = reqwest::Url::parse(url).context("parse RTSP URL")?;

    let opts = SessionOptions::default()
        .creds(Some(creds))
        .user_agent("tapo".into());

    let mut session = Session::describe(parsed_url, opts)
        .await
        .context("RTSP DESCRIBE")?;

    let video_idx = session
        .streams()
        .iter()
        .position(|s| (s.media() == "video" || s.media() == "image") && s.encoding_name() == "jpeg")
        .ok_or_else(|| anyhow!("no MJPEG video stream in RTSP SDP"))?;

    session
        .setup(
            video_idx,
            SetupOptions::default().transport(Transport::Tcp(TcpTransportOptions::default())),
        )
        .await
        .context("RTSP SETUP")?;

    let mut session = session
        .play(PlayOptions::default())
        .await
        .context("RTSP PLAY")?;

    let mut frame: Vec<u8> = Vec::with_capacity(64 * 1024);
    let mut have_start = false;

    loop {
        let item = session
            .next()
            .await
            .ok_or_else(|| anyhow!("RTSP stream ended before a frame arrived"))?
            .context("RTSP packet")?;

        let pkt = match item {
            PacketItem::Rtp(p) => p,
            _ => continue,
        };

        let mark = pkt.mark();
        let payload = pkt.payload();
        // RFC 2435 main header is 8 bytes:
        //   0: type-specific
        //   1..4: fragment offset (24-bit big-endian)
        //   4: type
        //   5: Q
        //   6: width / 8
        //   7: height / 8
        if payload.len() < 8 {
            continue;
        }
        let frag_offset = u32::from_be_bytes([0, payload[1], payload[2], payload[3]]);
        let type_field = payload[4];
        let q = payload[5];
        let mut body_offset = 8usize;
        // Restart-marker header (4 bytes), present when type >= 64.
        if type_field >= 64 {
            if payload.len() < body_offset + 4 {
                continue;
            }
            body_offset += 4;
        }
        // Quantization-table header on the first fragment when Q >= 128.
        if frag_offset == 0 && q >= 128 {
            if payload.len() < body_offset + 4 {
                continue;
            }
            let length =
                ((payload[body_offset + 2] as usize) << 8) | (payload[body_offset + 3] as usize);
            body_offset += 4 + length;
            if payload.len() < body_offset {
                continue;
            }
        }

        if frag_offset == 0 {
            frame.clear();
            have_start = true;
        } else if !have_start {
            // Joined mid-frame; wait for the next first fragment.
            continue;
        }

        frame.extend_from_slice(&payload[body_offset..]);

        if mark && have_start {
            // Tapo cameras embed a full JFIF (SOI..EOI) in the RTP body,
            // so `frame` is already a complete JPEG. Verify the SOI/EOI
            // bookends and emit it; otherwise wait for the next frame.
            let n = frame.len();
            let has_soi = n >= 4 && frame[0] == 0xff && frame[1] == 0xd8;
            let has_eoi = n >= 4 && frame[n - 2] == 0xff && frame[n - 1] == 0xd9;
            if has_soi && has_eoi {
                return Ok(frame);
            }
            frame.clear();
            have_start = false;
        }
    }
}