//! Where a frame's time goes in the essence path, stage by stage, on this host: the RFC
//! 4175 reader, the 4:2:2 to 4:2:0 step, the encoder, and packaging with its hashes and
//! signatures. A profile to read before changing any of them, not a throughput claim: the
//! frames are fed as fast as each stage takes them, one stage at a time.
//!
//! ```text
//! cargo run --release -p joule-moq-ingest --features st2110 --example st2110_profile -- [encoder] [WxH:frames ...]
//! ```
//!
//! `encoder` is `auto`, `va`, `svt-av1` or `rav1e` (default `auto`); `[st2110]`'s other
//! settings are the defaults, but `WAI_SVT_AV1_LIBRARY` and `WAI_VA_DEVICE` where set.

use joule_moq_ingest::config::St2110Config;
use joule_moq_ingest::frame::{Codec, GrantRef};
use joule_moq_ingest::package::{Packager, Signer, TrackOut};
use joule_moq_ingest::st2110::encode::{self, Job, Settings, Stage};
use joule_moq_ingest::st2110::sdp::Sampling;
use joule_moq_ingest::st2110::send::{self, video_format, video_packets, Out};
use joule_moq_ingest::st2110::video::VideoReader;
use std::sync::Arc;
use std::time::Instant;

/// The process's CPU time so far, in seconds (Linux; 0 elsewhere).
fn cpu_s() -> f64 {
    let s = std::fs::read_to_string("/proc/self/stat").unwrap_or_default();
    let f: Vec<&str> = s.rsplit(") ").next().unwrap_or("").split(' ').collect();
    let t = |i: usize| f.get(i).and_then(|x| x.parse::<f64>().ok()).unwrap_or(0.0);
    (t(11) + t(12)) / 100.0
}

fn main() {
    let args: Vec<String> = std::env::args().skip(1).collect();
    let mut c = St2110Config { encoder: args.first().cloned().unwrap_or_else(|| "auto".into()), ..St2110Config::default() };
    encode::from_env(&mut c);
    let sizes: Vec<(u32, u32, u32)> = if args.len() > 1 {
        args[1..].iter().map(|a| {
            let (wh, n) = a.split_once(':').expect("WxH:frames");
            let (w, h) = wh.split_once('x').expect("WxH:frames");
            (w.parse().unwrap(), h.parse().unwrap(), n.parse().unwrap())
        }).collect()
    } else {
        vec![(640, 360, 60), (1280, 720, 60), (1920, 1080, 60)]
    };
    println!("build: optimised {}; load {}", !cfg!(debug_assertions), std::fs::read_to_string("/proc/loadavg").unwrap_or_default().trim());
    let chosen = encode::choose(&c).expect("an encoder");
    for l in chosen.lines() {
        println!("{l}");
    }
    for (w, h, n) in sizes {
        let f = video_format(Sampling::YCbCr422, 10, w, h, (30, 1));
        let s = Settings::of(&f, &c);
        // The sender's side, outside what is timed: four pictures, packed.
        let mut out = Out::new(96, 1, 0);
        let packed: Vec<Vec<Vec<u8>>> = (0..4).map(|k| video_packets(&f, &send::picture(&f, k), 0, &mut out, 1_440)).collect();
        let per_frame = packed[0].len();
        // The reader, as the session runs it: one packet at a time.
        let mut frames = Vec::new();
        let t = Instant::now();
        for k in 0..n {
            let mut reader = VideoReader::new(f.clone(), 96);
            for p in &packed[k as usize % 4] {
                frames.extend(reader.push(p).frames);
            }
            if let Some(fr) = reader.finish() {
                frames.push(fr);
            }
        }
        let unpack = t.elapsed().as_secs_f64();
        assert!(frames.len() == n as usize && frames.iter().all(|f| f.complete));
        // The stage: conversion and encode on its own thread, every frame taken.
        let backend = encode::open(chosen.identity.kind, &s, &c).expect("the encoder opens");
        let mut stage = Stage::start(backend, s.clone(), f.sampling, n as usize, c.key_interval_ms * 90);
        let (c0, t) = (cpu_s(), Instant::now());
        for (k, fr) in frames.into_iter().enumerate() {
            assert!(stage.submit(Job { planes: fr.planes, rtp: k as u64 * 3_000, arrival: k as u64 * 33_333, capture_ms: 0 }));
        }
        stage.finish();
        let mut units = Vec::new();
        while let Some(u) = stage.coded.blocking_recv() {
            units.push(u.expect("the encoder holds the floor"));
        }
        let (wall, cpu) = (t.elapsed().as_secs_f64(), cpu_s() - c0);
        // The first temporal units, for a reader of bitstreams: `WAI_PROFILE_DUMP=<directory>`.
        if let Ok(dir) = std::env::var("WAI_PROFILE_DUMP") {
            for (k, u) in units.iter().take(3).enumerate() {
                let _ = std::fs::write(format!("{dir}/{}-{w}x{h}-{k}.obu", chosen.identity.kind.name()), &u.unit);
            }
        }
        let st = stage.stats();
        // Packaging: the framing, a hash an object, a signature a group.
        let signer = Signer { key: Arc::new(ed25519_dalek::SigningKey::from_bytes(&[3; 32])), signer_id: "profile".into(), origin_label: "wai-ingest".into() };
        let mut pk = Packager::new(TrackOut { media: joule_moq_ingest::room::media::VIDEO, media_alias: 4, receipts_alias: 5, receipt_alias: 7, priority: 1 }, 90_000, 250_000, GrantRef { grant_hash: [9; 32], joule_ceiling_micro: u64::MAX, funds_ceiling: None }, signer, 0, 200_000, 2_000_000, 8 << 20);
        let t = Instant::now();
        let mut emits = Vec::new();
        for u in &units {
            pk.on_frame(Codec::Av1, &u.unit, u.rtp, u.arrival, u.key, &mut emits);
        }
        pk.end_group(&mut emits);
        let package = t.elapsed().as_secs_f64();
        let ms = |x: f64| x * 1000.0 / n as f64;
        let (convert, enc) = (st.convert_us as f64 / 1e6, st.encode_us as f64 / 1e6);
        println!(
            "{w} x {h}, 4:2:2 at 10 bits, {n} frames of {per_frame} packets: ms a frame: reader {:.2}, 4:2:2 to 4:2:0 {:.2}, encode {:.2} (max {:.1}), package, hash and sign {:.3}; the encoder alone {:.1} frames a second, its CPU time {:.1} times its wall time; {} octets a frame; rungs {} to {} of {}; {}",
            ms(unpack),
            ms(convert),
            ms(enc),
            st.encode_us_max as f64 / 1000.0,
            ms(package),
            n as f64 / enc,
            cpu / wall,
            st.octets / n as u64,
            st.rung_first,
            st.rung_last,
            st.rungs,
            st.detail
        );
    }
}