AI Flow: Video

@@ -29,6 +29,13 @@   .takes a { color: var(--teal); }   .dot { display: inline-block; width: 10px; height: 10px; border-radius: 50%; background: var(--red); margin-right: 6px; animation: pulse 1s infinite; }   @keyframes pulse { 50% { opacity: .25; } }+  #toast { position: fixed; left: 50%; top: 64px; transform: translateX(-50%); background: var(--panel); border: 1px solid var(--teal); border-radius: 10px; padding: 10px 16px; opacity: 0; transition: opacity .25s; pointer-events: none; max-width: 70vw; z-index: 20; }+  #toast.on { opacity: 1; }+  #aicur { position: fixed; left: 0; top: 0; z-index: 30; pointer-events: none; opacity: 0; transition: transform .35s cubic-bezier(.3,.7,.2,1), opacity .2s; }+  #aicur.on { opacity: 1; }+  #aicur i { display: block; width: 16px; height: 16px; border-radius: 50%; background: var(--teal); box-shadow: 0 0 0 4px rgba(20,196,184,.25); }+  #aicur span { position: absolute; left: 20px; top: -4px; white-space: nowrap; font-size: 13px; background: var(--panel); border: 1px solid var(--line); border-radius: 6px; padding: 2px 8px; }+  .ai { color: var(--teal); font-size: 13px; } </style> </head> <body>@@ -46,11 +53,14 @@   <span class="spacer"></span>   <div class="takes" id="takes">No takes yet.</div> </div>+<div id="toast"></div>+<div id="aicur"><i></i><span id="aicurlbl"></span></div> <script> const $ = id => document.getElementById(id); const v = $('v'), state = $('state');+let say, open; let stream = null, rec = null, chunks = [], ctx = null, raf = 0, t = {}, peak = 0, takes = [];-const say = html => state.innerHTML = html;+say = html => state.innerHTML = html;  async function mics() {   const devs = (await navigator.mediaDevices.enumerateDevices()).filter(d => d.kind === 'audioinput');@@ -58,7 +68,7 @@ async function mics() {   $('mic').innerHTML = devs.map((d, i) => `<option value="${d.deviceId}">${d.label || 'Microphone ' + (i + 1)}</option>`).join('');   if (keep) $('mic').value = keep; }-async function open() {+open = async function () {   if (stream) stream.getTracks().forEach(x => x.stop());   const id = $('mic').value;   stream = await navigator.mediaDevices.getUserMedia({ audio: { deviceId: id ? { exact: id } : undefined, echoCancellation: false, noiseSuppression: false, autoGainControl: false, channelCount: 1 } });@@ -76,7 +86,7 @@ async function open() {     $('lvl').classList.toggle('hot', m > 0.95);     raf = requestAnimationFrame(loop);   })();-}+}; $('mic').onchange = () => open().catch(e => say('Could not open that microphone: ' + e.message)); $('test').onclick = () => open().then(() => say('Speak normally: the bar should reach about two thirds. Red means too loud.')).catch(e => say('The browser did not give access to a microphone (' + e.message + '). Allow it in the address bar and try again.')); @@ -112,6 +122,41 @@ async function save() { } navigator.mediaDevices.addEventListener('devicechange', mics); mics();++// ---- the live channel: the AI drives this page through the app's CLI, and the page reports back+let level = 0, micAccess = 'not asked yet', lastStatus = '';+const _say = say; say = html => { _say(html); lastStatus = state.textContent; report(); };+function report() {+  const body = { mics: [...$('mic').options].map(o => o.textContent), mic: $('mic').selectedOptions[0]?.textContent || '', micAccess, recording: !!rec && rec.state === 'recording', t: +v.currentTime.toFixed(2), duration: +(v.duration || 0).toFixed(2), level: +level.toFixed(3), peak: +peak.toFixed(3), takes: takes.length, status: lastStatus };+  fetch('/state', { method: 'POST', body: JSON.stringify(body) }).catch(() => {});+}+setInterval(report, 1000);+const _open = open; open = async () => { try { await _open(); micAccess = 'allowed'; } catch (e) { micAccess = 'blocked: ' + e.message; report(); throw e; } report(); };+(function meter() { const w = parseFloat($('lvl').style.width) || 0; level = w / 140; requestAnimationFrame(meter); })();++function toast(text) { const el = $('toast'); el.textContent = text; el.classList.add('on'); clearTimeout(toast.t); toast.t = setTimeout(() => el.classList.remove('on'), 4000); }+async function cursorTo(el, label) {+  const c = $('aicur'), r = el.getBoundingClientRect();+  $('aicurlbl').textContent = label || '';+  c.classList.add('on'); c.style.transform = `translate(${r.left + r.width / 2 - 8}px, ${r.top + r.height / 2 - 8}px)`;+  await new Promise(r => setTimeout(r, 420));+}+async function aiClick(el, label) { if (!el || el.disabled) return; await cursorTo(el, label); el.click(); setTimeout(() => $('aicur').classList.remove('on'), 900); }+const es = new EventSource('/events');+const on = (name, fn) => es.addEventListener(name, e => { let d = {}; try { d = JSON.parse(e.data); } catch (_) {} fn(d); });+on('record', d => aiClick($('rec'), (d.by || 'AI') + ': start recording'));+on('stop', d => aiClick($('stop'), (d.by || 'AI') + ': stop and save'));+on('test', d => aiClick($('test'), (d.by || 'AI') + ': test the microphone'));+on('mic', async d => { const o = [...$('mic').options].find(o => o.textContent.toLowerCase().includes(String(d.name || '').toLowerCase())); if (!o) return toast('No microphone matching "' + d.name + '"'); await cursorTo($('mic'), (d.by || 'AI') + ': pick ' + o.textContent); $('mic').value = o.value; $('mic').onchange(); setTimeout(() => $('aicur').classList.remove('on'), 900); });+on('seek', d => { v.currentTime = Math.max(0, +d.seconds || 0); report(); });+on('play', () => v.play());+on('pause', () => v.pause());+on('toast', d => toast(d.text || ''));+on('click', d => aiClick(document.querySelector(d.selector), d.by || 'AI'));+on('reveal', d => document.querySelector(d.selector)?.scrollIntoView({ block: 'center', behavior: 'smooth' }));+on('cursor', d => { const el = document.querySelector(d.selector); el && cursorTo(el, d.by || 'AI'); });+on('reload', () => location.reload());+on('state', d => { if (d.takes && d.takes.length > takes.length) report(); }); </script> </body> </html>

@@ -5,6 +5,25 @@ use std::io::{BufRead, BufReader, Read, Seek, SeekFrom, Write}; use std::net::{TcpListener, TcpStream}; use std::path::{Path, PathBuf};+use std::sync::{Arc, Mutex};+use serde_json::{json, Value};++/// What every connection shares: the open pages' event streams and the state they sync to.+pub struct Shared { clients: Vec<TcpStream>, state: Value }++/// Push one event to every open page; a page that has gone away is dropped.+fn broadcast(sh: &Arc<Mutex<Shared>>, event: &str, data: &Value) {+    let msg = format!("event: {event}\ndata: {data}\n\n");+    let mut g = sh.lock().unwrap();+    g.clients.retain_mut(|c| c.write_all(msg.as_bytes()).and_then(|_| c.flush()).is_ok());+}++fn takes_in(out: &Path) -> Vec<Value> {+    let mut v: Vec<Value> = std::fs::read_dir(out).map(|d| d.flatten().filter(|e| e.file_name().to_string_lossy().ends_with(".json") && e.file_name().to_string_lossy().starts_with("take-"))+        .filter_map(|e| std::fs::read_to_string(e.path()).ok().and_then(|t| serde_json::from_str(&t).ok())).collect()).unwrap_or_default();+    v.sort_by_key(|x: &Value| x["file"].as_str().unwrap_or("").to_string());+    v+}  const PAGE: &str = include_str!("../assets/narrate.html"); @@ -55,24 +74,64 @@ fn seconds_of(path: &Path) -> f64 {         .and_then(|o| String::from_utf8_lossy(&o.stdout).trim().parse().ok()).unwrap_or(0.0) } -fn handle(mut s: TcpStream, video: PathBuf, out: PathBuf) {+fn handle(mut s: TcpStream, video: PathBuf, out: PathBuf, sh: Arc<Mutex<Shared>>) {     let mut rd = BufReader::new(s.try_clone().expect("stream"));     let mut line = String::new();     if rd.read_line(&mut line).unwrap_or(0) == 0 { return; }     let mut it = line.split_whitespace();     let (method, target) = (it.next().unwrap_or("").to_string(), it.next().unwrap_or("/").to_string());-    let (mut len, mut range, mut meta) = (0usize, None, String::new());+    let (mut len, mut range, mut meta, mut who) = (0usize, None, String::new(), String::new());     loop {         let mut h = String::new();         if rd.read_line(&mut h).unwrap_or(0) == 0 || h == "\r\n" || h == "\n" { break; }         if let Some((k, v)) = h.split_once(':') {-            match k.to_ascii_lowercase().as_str() { "content-length" => len = v.trim().parse().unwrap_or(0), "range" => range = Some(v.trim().to_string()), "x-take-meta" => meta = pct_decode(v.trim()), _ => {} }+            match k.to_ascii_lowercase().as_str() { "content-length" => len = v.trim().parse().unwrap_or(0), "range" => range = Some(v.trim().to_string()), "x-take-meta" => meta = pct_decode(v.trim()), "x-ai-thread" => who = pct_decode(v.trim()), _ => {} }         }     }     let path = target.split('?').next().unwrap_or("/");     match (method.as_str(), path) {         ("GET", "/") => respond(&mut s, "200 OK", "text/html; charset=utf-8", "", PAGE.as_bytes()),         ("GET", "/video") => serve_file(&mut s, &video, range.as_deref()),+        // the live channel: the page opens this once and every directive arrives on it+        ("GET", "/events") => {+            let _ = write!(s, "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nCache-Control: no-cache\r\nConnection: keep-alive\r\nX-Accel-Buffering: no\r\n\r\n:{}\n\n", " ".repeat(2048));+            let mut g = sh.lock().unwrap();+            let _ = write!(s, "event: state\ndata: {}\n\n", g.state);+            let _ = s.flush();+            g.clients.push(s);+        }+        ("GET", "/state") => {+            let mut g = sh.lock().unwrap();+            g.state["takes"] = takes_in(&out).into();+            g.state["pagesOpen"] = g.clients.len().into();+            let body = g.state.to_string();+            drop(g);+            respond(&mut s, "200 OK", "application/json", "", body.as_bytes())+        }+        // the page reports what it is doing (mics, level, recording, where the video is)+        ("POST", "/state") => {+            let mut body = vec![0u8; len];+            let _ = rd.read_exact(&mut body);+            if let Ok(v) = serde_json::from_slice::<Value>(&body) { sh.lock().unwrap().state["page"] = v; }+            respond(&mut s, "200 OK", "application/json", "", b"{\"ok\":true}")+        }+        // drive the open page: record, stop, test, mic, seek, toast, click, reveal, cursor, reload+        ("POST", p) if p.starts_with("/ui/") => {+            let verb = p.trim_start_matches("/ui/").to_string();+            let mut body = vec![0u8; len];+            let _ = rd.read_exact(&mut body);+            let mut data: Value = serde_json::from_slice(&body).unwrap_or(json!({}));+            if who.is_empty() { who = data["aiThread"].as_str().unwrap_or("").to_string(); }+            if who.is_empty() { return respond(&mut s, "400 Bad Request", "application/json", "", json!({"error": "caller_identity_required", "hint": "say who is driving the page: pass --ai-thread \"<your thread name>\" (header X-AI-Thread)"}).to_string().as_bytes()); }+            if !["record", "stop", "test", "mic", "seek", "toast", "click", "reveal", "cursor", "reload", "play", "pause"].contains(&verb.as_str()) {+                return respond(&mut s, "404 Not Found", "application/json", "", json!({"error": "unknown directive", "hint": "record, stop, test, mic, seek, play, pause, toast, click, reveal, cursor, reload"}).to_string().as_bytes());+            }+            data["by"] = who.clone().into();+            let pages = sh.lock().unwrap().clients.len();+            broadcast(&sh, &verb, &data);+            sh.lock().unwrap().state["lastDirective"] = json!({"verb": verb, "by": who, "data": data});+            respond(&mut s, "200 OK", "application/json", "", json!({"ok": true, "verb": verb, "pagesOpen": pages}).to_string().as_bytes())+        }         ("GET", p) if p.starts_with("/takes/") => {             let name = p.trim_start_matches("/takes/");             if name.contains('/') || name.contains("..") { return respond(&mut s, "400 Bad Request", "text/plain", "", b""); }@@ -92,22 +151,56 @@ fn handle(mut s: TcpStream, video: PathBuf, out: PathBuf) {             let _ = std::fs::write(out.join(format!("take-{n}.json")), serde_json::to_string_pretty(&m).unwrap_or_default() + "\n");             println!("OK: take {n} saved: {} ({:.0} s, peak {})", out.join(&file).display(), secs, m["peak"]);             println!("Hint: listen to it before you cut (clipping, room noise, missed scenes); the video starts {} ms into the recording.", m["videoStartsAtMs"]);+            broadcast(&sh, "state", &json!({"takes": takes_in(&out)}));             respond(&mut s, "200 OK", "application/json", "", serde_json::json!({"file": file, "seconds": secs}).to_string().as_bytes())         }         _ => respond(&mut s, "404 Not Found", "text/plain", "", b"not found"),     } } +fn registry() -> PathBuf { PathBuf::from(std::env::var("HOME").unwrap_or_default()).join(".adom/instances/adom-aiflow-video") }++/// The port of a running narration page: the newest registered one that still answers.+pub fn find_port() -> Option<u16> {+    let mut ports: Vec<(std::time::SystemTime, u16)> = std::fs::read_dir(registry()).ok()?.flatten().filter_map(|e| {+        let p: u16 = e.path().file_stem()?.to_str()?.strip_prefix("narrate-")?.parse().ok()?;+        Some((e.metadata().ok()?.modified().ok()?, p))+    }).collect();+    ports.sort(); ports.reverse();+    ports.into_iter().map(|x| x.1).find(|p| TcpStream::connect(("127.0.0.1", *p)).is_ok())+}++/// One request to a running narration page; returns (status code, body).+pub fn call(port: u16, method: &str, path: &str, ai_thread: Option<&str>, body: &str) -> Result<(u16, String), String> {+    let mut s = TcpStream::connect(("127.0.0.1", port)).map_err(|e| format!("no narration page answers on port {port}: {e}"))?;+    let who = ai_thread.map(|t| format!("X-AI-Thread: {t}\r\n")).unwrap_or_default();+    write!(s, "{method} {path} HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: {}\r\n{who}Connection: close\r\n\r\n{body}", body.len()).map_err(|e| e.to_string())?;+    let mut t = String::new();+    s.read_to_string(&mut t).map_err(|e| e.to_string())?;+    let code = t.split_whitespace().nth(1).and_then(|c| c.parse().ok()).unwrap_or(0);+    Ok((code, t.split_once("\r\n\r\n").map(|x| x.1.to_string()).unwrap_or_default()))+}+ /// Serve until stopped. Returns only when the port cannot be bound. pub fn serve(video: &Path, out: &Path, port: u16) -> Result<(), String> {     let l = TcpListener::bind(("0.0.0.0", port)).map_err(|e| format!("could not listen on port {port}: {e}"))?;     let port = l.local_addr().map(|a| a.port()).unwrap_or(port);+    let sh = Arc::new(Mutex::new(Shared { clients: vec![], state: json!({"app": "adom-aiflow-video narrate", "port": port, "video": video.display().to_string(), "out": out.display().to_string(), "takes": takes_in(out), "page": {}}) }));+    let _ = std::fs::create_dir_all(registry());+    let _ = std::fs::write(registry().join(format!("narrate-{port}.json")), json!({"port": port, "pid": std::process::id(), "video": video.display().to_string(), "out": out.display().to_string()}).to_string());     println!("OK: narration page at http://localhost:{port}/ (teleprompter: {})", video.display());     println!("Hint: open it in the human's own browser on the machine with the microphone (a pup window or their default browser; a browser allows the microphone on localhost). They pick the mic, press Record and read along; each take is saved in {}.", out.display());+    println!("Hint: you can drive the open page yourself: `narrate status`, `narrate test`, `narrate record`, `narrate stop`, `narrate mic <name>`, `narrate seek <seconds>`, `narrate toast \"...\"` (each with --ai-thread).");     println!("Hint: leave this running while they record; stop it (Ctrl-C) when they are done.");+    let hb = sh.clone();+    std::thread::spawn(move || loop {+        std::thread::sleep(std::time::Duration::from_secs(15));+        let mut g = hb.lock().unwrap();+        g.clients.retain_mut(|c| c.write_all(b": hb\n\n").and_then(|_| c.flush()).is_ok());+    });     for s in l.incoming().flatten() {-        let (v, o) = (video.to_path_buf(), out.to_path_buf());-        std::thread::spawn(move || handle(s, v, o));+        let (v, o, h) = (video.to_path_buf(), out.to_path_buf(), sh.clone());+        std::thread::spawn(move || handle(s, v, o, h));     }     Ok(()) }