Add kiosk desktop app
Tauri app for the Pi. Media runs through the native LiveKit SDK because WebKitGTK has no WebRTC, with frames encoded in Rust and drawn to a canvas under the annotation overlay.
This commit is contained in:
SirBlob committed
1 parent
18384c0cbc
commit
7d0a973638
85 files changed
+8330
No files matched your search
@@ -0,0 +1,180 @@
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::sync::Arc;
|
||||
|
||||
use livekit::track::{RemoteTrack, TrackSource};
|
||||
use livekit::{Room, RoomEvent, RoomOptions};
|
||||
use tauri::{AppHandle, Emitter};
|
||||
use tokio::sync::Mutex;
|
||||
|
||||
use super::events::{
|
||||
DataPayload, ParticipantView, ParticipantsPayload, StatusPayload, DATA_EVENT,
|
||||
PARTICIPANTS_EVENT, STATUS_EVENT,
|
||||
};
|
||||
use super::video;
|
||||
|
||||
/// Bumped on every connect attempt. Tasks belonging to a superseded connection compare
|
||||
/// against this before emitting anything, so a room we deliberately replaced cannot
|
||||
/// report itself as disconnected and trigger a pointless reconnect.
|
||||
static GENERATION: AtomicU64 = AtomicU64::new(0);
|
||||
|
||||
pub fn is_current(generation: u64) -> bool {
|
||||
GENERATION.load(Ordering::SeqCst) == generation
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
pub struct RoomHandle {
|
||||
room: Mutex<Option<Arc<Room>>>,
|
||||
}
|
||||
|
||||
impl RoomHandle {
|
||||
pub async fn connect(&self, app: AppHandle, url: String, token: String) -> Result<(), String> {
|
||||
let generation = GENERATION.fetch_add(1, Ordering::SeqCst) + 1;
|
||||
|
||||
self.close_current().await;
|
||||
video::stop(&app);
|
||||
|
||||
emit_status(&app, generation, StatusPayload::new("connecting"));
|
||||
|
||||
let (room, mut events) = Room::connect(&url, &token, RoomOptions::default())
|
||||
.await
|
||||
.map_err(|error| error.to_string())?;
|
||||
|
||||
if !is_current(generation) {
|
||||
let _ = room.close().await;
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
*self.room.lock().await = Some(Arc::new(room));
|
||||
emit_status(&app, generation, StatusPayload::new("connected"));
|
||||
|
||||
let room_for_task = self.room.lock().await.clone();
|
||||
|
||||
tauri::async_runtime::spawn(async move {
|
||||
if let Some(room) = room_for_task.as_ref() {
|
||||
emit_participants(&app, generation, room);
|
||||
}
|
||||
|
||||
while let Some(event) = events.recv().await {
|
||||
if !is_current(generation) {
|
||||
return;
|
||||
}
|
||||
|
||||
if matches!(
|
||||
event,
|
||||
RoomEvent::ParticipantConnected(_) | RoomEvent::ParticipantDisconnected(_)
|
||||
) {
|
||||
if let Some(room) = room_for_task.as_ref() {
|
||||
emit_participants(&app, generation, room);
|
||||
}
|
||||
}
|
||||
|
||||
handle_event(&app, generation, event);
|
||||
}
|
||||
|
||||
if is_current(generation) {
|
||||
video::stop(&app);
|
||||
emit_status(&app, generation, StatusPayload::new("disconnected"));
|
||||
}
|
||||
});
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn disconnect(&self) {
|
||||
GENERATION.fetch_add(1, Ordering::SeqCst);
|
||||
self.close_current().await;
|
||||
}
|
||||
|
||||
async fn close_current(&self) {
|
||||
let existing = self.room.lock().await.take();
|
||||
if let Some(room) = existing {
|
||||
let _ = room.close().await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn handle_event(app: &AppHandle, generation: u64, event: RoomEvent) {
|
||||
match event {
|
||||
RoomEvent::Connected { .. } => emit_status(app, generation, StatusPayload::new("connected")),
|
||||
|
||||
RoomEvent::Reconnecting => {
|
||||
emit_status(app, generation, StatusPayload::new("reconnecting"))
|
||||
}
|
||||
|
||||
RoomEvent::Reconnected => emit_status(app, generation, StatusPayload::new("connected")),
|
||||
|
||||
RoomEvent::Disconnected { reason } => {
|
||||
video::stop(app);
|
||||
emit_status(
|
||||
app,
|
||||
generation,
|
||||
StatusPayload::with_detail("disconnected", format!("{reason:?}")),
|
||||
);
|
||||
}
|
||||
|
||||
RoomEvent::DataReceived { payload, .. } => {
|
||||
if let Ok(envelope) = String::from_utf8(payload.to_vec()) {
|
||||
let _ = app.emit(DATA_EVENT, DataPayload { envelope });
|
||||
}
|
||||
}
|
||||
|
||||
RoomEvent::TrackSubscribed {
|
||||
track, publication, ..
|
||||
} => {
|
||||
let priority = match publication.source() {
|
||||
TrackSource::Screenshare => video::PRIORITY_SCREEN,
|
||||
TrackSource::Camera => video::PRIORITY_CAMERA,
|
||||
_ => return,
|
||||
};
|
||||
|
||||
if let RemoteTrack::Video(video_track) = track {
|
||||
video::start(
|
||||
app.clone(),
|
||||
generation,
|
||||
video_track,
|
||||
priority,
|
||||
publication.sid().to_string(),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
RoomEvent::TrackUnsubscribed { publication, .. } => {
|
||||
video::stop_track(app, &publication.sid().to_string());
|
||||
}
|
||||
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
|
||||
fn emit_participants(app: &AppHandle, generation: u64, room: &Room) {
|
||||
if !is_current(generation) {
|
||||
return;
|
||||
}
|
||||
|
||||
let participants = room
|
||||
.remote_participants()
|
||||
.values()
|
||||
.map(|participant| {
|
||||
let identity = participant.identity().to_string();
|
||||
let name = participant.name();
|
||||
|
||||
ParticipantView {
|
||||
display_name: if name.is_empty() {
|
||||
identity.clone()
|
||||
} else {
|
||||
name
|
||||
},
|
||||
identity,
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
|
||||
let _ = app.emit(PARTICIPANTS_EVENT, ParticipantsPayload { participants });
|
||||
}
|
||||
|
||||
fn emit_status(app: &AppHandle, generation: u64, payload: StatusPayload) {
|
||||
if !is_current(generation) {
|
||||
return;
|
||||
}
|
||||
let _ = app.emit(STATUS_EVENT, payload);
|
||||
}
|
||||
@@ -0,0 +1,56 @@
|
||||
use serde::Serialize;
|
||||
|
||||
pub const STATUS_EVENT: &str = "room://status";
|
||||
pub const DATA_EVENT: &str = "room://data";
|
||||
pub const VIDEO_EVENT: &str = "room://video";
|
||||
pub const PARTICIPANTS_EVENT: &str = "room://participants";
|
||||
|
||||
#[derive(Clone, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct ParticipantView {
|
||||
pub identity: String,
|
||||
pub display_name: String,
|
||||
}
|
||||
|
||||
#[derive(Clone, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct ParticipantsPayload {
|
||||
pub participants: Vec<ParticipantView>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct StatusPayload {
|
||||
pub status: &'static str,
|
||||
pub detail: Option<String>,
|
||||
}
|
||||
|
||||
impl StatusPayload {
|
||||
pub fn new(status: &'static str) -> Self {
|
||||
Self {
|
||||
status,
|
||||
detail: None,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn with_detail(status: &'static str, detail: String) -> Self {
|
||||
Self {
|
||||
status,
|
||||
detail: Some(detail),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct DataPayload {
|
||||
pub envelope: String,
|
||||
}
|
||||
|
||||
#[derive(Clone, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct VideoPayload {
|
||||
pub active: bool,
|
||||
pub width: u32,
|
||||
pub height: u32,
|
||||
}
|
||||
@@ -0,0 +1,131 @@
|
||||
use std::sync::Mutex;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use image::codecs::jpeg::JpegEncoder;
|
||||
use image::ExtendedColorType;
|
||||
use tauri::ipc::{Channel, InvokeResponseBody};
|
||||
|
||||
/// Where encoded frames go. The webview opens this channel once and keeps it for the life
|
||||
/// of the app, so video never has to travel as an event payload.
|
||||
static SINK: Mutex<Option<Channel<InvokeResponseBody>>> = Mutex::new(None);
|
||||
static LAST_SENT: Mutex<Option<Instant>> = Mutex::new(None);
|
||||
|
||||
const DEFAULT_MAX_WIDTH: u32 = 1920;
|
||||
const DEFAULT_FPS: u32 = 60;
|
||||
const DEFAULT_QUALITY: u8 = 85;
|
||||
|
||||
/// Read once at startup. A Pi will not keep up with the desktop defaults, so these are
|
||||
/// tunable without a rebuild: PISTATION_VIDEO_MAX_WIDTH, _FPS and _QUALITY.
|
||||
pub struct VideoSettings {
|
||||
pub max_width: u32,
|
||||
pub frame_interval: Duration,
|
||||
pub quality: u8,
|
||||
}
|
||||
|
||||
fn settings() -> &'static VideoSettings {
|
||||
static SETTINGS: std::sync::OnceLock<VideoSettings> = std::sync::OnceLock::new();
|
||||
|
||||
SETTINGS.get_or_init(|| {
|
||||
let max_width = env_number("PISTATION_VIDEO_MAX_WIDTH", DEFAULT_MAX_WIDTH as u64) as u32;
|
||||
let fps = (env_number("PISTATION_VIDEO_FPS", DEFAULT_FPS as u64) as u32).clamp(1, 120);
|
||||
let quality = env_number("PISTATION_VIDEO_QUALITY", DEFAULT_QUALITY as u64).clamp(1, 100);
|
||||
|
||||
VideoSettings {
|
||||
max_width: max_width.max(160),
|
||||
frame_interval: Duration::from_secs_f64(1.0 / fps as f64),
|
||||
quality: quality as u8,
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
fn env_number(key: &str, fallback: u64) -> u64 {
|
||||
std::env::var(key)
|
||||
.ok()
|
||||
.and_then(|value| value.parse().ok())
|
||||
.unwrap_or(fallback)
|
||||
}
|
||||
|
||||
pub fn set_sink(channel: Option<Channel<InvokeResponseBody>>) {
|
||||
if let Ok(mut guard) = SINK.lock() {
|
||||
*guard = channel;
|
||||
}
|
||||
if let Ok(mut guard) = LAST_SENT.lock() {
|
||||
*guard = None;
|
||||
}
|
||||
}
|
||||
|
||||
pub fn has_sink() -> bool {
|
||||
SINK.lock().map(|guard| guard.is_some()).unwrap_or(false)
|
||||
}
|
||||
|
||||
/// Returns the size this frame should be encoded at, or None when it should be dropped
|
||||
/// because the previous one went out too recently.
|
||||
pub fn next_target(width: u32, height: u32) -> Option<(u32, u32)> {
|
||||
if width == 0 || height == 0 || !has_sink() {
|
||||
return None;
|
||||
}
|
||||
|
||||
let config = settings();
|
||||
let mut guard = LAST_SENT.lock().ok()?;
|
||||
|
||||
if let Some(last) = *guard {
|
||||
if last.elapsed() < config.frame_interval {
|
||||
return None;
|
||||
}
|
||||
}
|
||||
|
||||
*guard = Some(Instant::now());
|
||||
|
||||
if width <= config.max_width {
|
||||
return Some((width, height));
|
||||
}
|
||||
|
||||
let scaled_height = ((height as f32) * (config.max_width as f32 / width as f32)).round() as u32;
|
||||
Some((config.max_width, scaled_height.max(1)))
|
||||
}
|
||||
|
||||
/// Packs RGB into `rgb` from an R,G,B,A source and encodes into `encoded`.
|
||||
///
|
||||
/// Both buffers are owned by the caller and reused across frames. At sixty frames a second
|
||||
/// the allocations alone were costing more than the encode.
|
||||
pub fn send(source: &[u8], rgb: &mut Vec<u8>, encoded: &mut Vec<u8>, width: u32, height: u32) {
|
||||
let pixels = (width as usize) * (height as usize);
|
||||
rgb.clear();
|
||||
rgb.reserve(pixels * 3);
|
||||
|
||||
for pixel in source.chunks_exact(4) {
|
||||
rgb.extend_from_slice(&pixel[..3]);
|
||||
}
|
||||
|
||||
encoded.clear();
|
||||
let mut encoder = JpegEncoder::new_with_quality(&mut *encoded, settings().quality);
|
||||
|
||||
if let Err(error) = encoder.encode(rgb, width, height, ExtendedColorType::Rgb8) {
|
||||
report_once(&format!("jpeg encode failed: {error}"));
|
||||
return;
|
||||
}
|
||||
|
||||
let Ok(guard) = SINK.lock() else {
|
||||
return;
|
||||
};
|
||||
|
||||
if let Some(channel) = guard.as_ref() {
|
||||
if let Err(error) = channel.send(InvokeResponseBody::Raw(std::mem::take(encoded))) {
|
||||
report_once(&format!("could not push a frame to the webview: {error}"));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Frames arrive many times a second, so a failure that repeats must not flood the log.
|
||||
fn report_once(message: &str) {
|
||||
static REPORTED: Mutex<Option<String>> = Mutex::new(None);
|
||||
|
||||
if let Ok(mut guard) = REPORTED.lock() {
|
||||
if guard.as_deref() == Some(message) {
|
||||
return;
|
||||
}
|
||||
*guard = Some(message.to_string());
|
||||
}
|
||||
|
||||
eprintln!("[pistation] {message}");
|
||||
}
|
||||
@@ -0,0 +1,38 @@
|
||||
pub mod client;
|
||||
mod events;
|
||||
mod frames;
|
||||
mod video;
|
||||
|
||||
pub use client::RoomHandle;
|
||||
|
||||
use tauri::ipc::{Channel, InvokeResponseBody};
|
||||
use tauri::{AppHandle, State};
|
||||
|
||||
/// The webview opens one channel for decoded video and keeps it open. Frames are JPEG
|
||||
/// encoded in Rust and drawn onto a canvas, which keeps the annotation overlay working
|
||||
/// exactly as it does in the browser.
|
||||
#[tauri::command]
|
||||
pub fn video_subscribe(channel: Channel<InvokeResponseBody>) {
|
||||
frames::set_sink(Some(channel));
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
pub fn video_unsubscribe() {
|
||||
frames::set_sink(None);
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
pub async fn room_connect(
|
||||
app: AppHandle,
|
||||
handle: State<'_, RoomHandle>,
|
||||
url: String,
|
||||
token: String,
|
||||
) -> Result<(), String> {
|
||||
handle.connect(app, url, token).await
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
pub async fn room_disconnect(handle: State<'_, RoomHandle>) -> Result<(), String> {
|
||||
handle.disconnect().await;
|
||||
Ok(())
|
||||
}
|
||||
@@ -0,0 +1,156 @@
|
||||
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicU8, Ordering};
|
||||
use std::sync::Mutex;
|
||||
|
||||
use futures_util::StreamExt;
|
||||
use livekit::track::RemoteVideoTrack;
|
||||
use livekit::webrtc::video_frame::native::VideoFrameBufferExt;
|
||||
use livekit::webrtc::video_frame::VideoFormatType;
|
||||
use livekit::webrtc::video_stream::native::NativeVideoStream;
|
||||
use tauri::{AppHandle, Emitter};
|
||||
|
||||
use super::client::is_current;
|
||||
use super::events::{VideoPayload, VIDEO_EVENT};
|
||||
use super::frames;
|
||||
|
||||
static IS_STREAMING: AtomicBool = AtomicBool::new(false);
|
||||
|
||||
/// Bumped whenever the displayed track changes, so a frame loop for a track we have moved
|
||||
/// off stops without having to be cancelled directly.
|
||||
static STREAM_EPOCH: AtomicU64 = AtomicU64::new(0);
|
||||
static CURRENT_PRIORITY: AtomicU8 = AtomicU8::new(0);
|
||||
static CURRENT_SID: Mutex<Option<String>> = Mutex::new(None);
|
||||
|
||||
/// A shared screen always wins over a camera, so plugging in a presentation takes the
|
||||
/// display back from whoever was pointing a phone at the room.
|
||||
pub const PRIORITY_CAMERA: u8 = 1;
|
||||
pub const PRIORITY_SCREEN: u8 = 2;
|
||||
|
||||
/// Consumes decoded frames from the subscribed screen share.
|
||||
///
|
||||
/// Frames are converted to BGRA, which on a little endian machine is the same layout as
|
||||
/// cairo's ARgb32, and handed to the native drawing surface sitting under the webview.
|
||||
pub fn start(
|
||||
app: AppHandle,
|
||||
generation: u64,
|
||||
track: RemoteVideoTrack,
|
||||
priority: u8,
|
||||
sid: String,
|
||||
) {
|
||||
// A camera must not displace a screen share that is already on the display.
|
||||
if IS_STREAMING.load(Ordering::SeqCst) && CURRENT_PRIORITY.load(Ordering::SeqCst) > priority {
|
||||
return;
|
||||
}
|
||||
|
||||
let epoch = STREAM_EPOCH.fetch_add(1, Ordering::SeqCst) + 1;
|
||||
CURRENT_PRIORITY.store(priority, Ordering::SeqCst);
|
||||
IS_STREAMING.store(true, Ordering::SeqCst);
|
||||
|
||||
if let Ok(mut guard) = CURRENT_SID.lock() {
|
||||
*guard = Some(sid);
|
||||
}
|
||||
|
||||
tauri::async_runtime::spawn(async move {
|
||||
let mut stream = NativeVideoStream::new(track.rtc_track());
|
||||
let mut last_size = (0u32, 0u32);
|
||||
|
||||
// Reused across frames so a sixty frame per second stream does not allocate
|
||||
// several megabytes per frame.
|
||||
let mut rgba = Vec::new();
|
||||
let mut rgb = Vec::new();
|
||||
let mut encoded = Vec::new();
|
||||
|
||||
while let Some(frame) = stream.next().await {
|
||||
if STREAM_EPOCH.load(Ordering::SeqCst) != epoch || !is_current(generation) {
|
||||
return;
|
||||
}
|
||||
|
||||
let width = frame.buffer.width();
|
||||
let height = frame.buffer.height();
|
||||
if width == 0 || height == 0 {
|
||||
continue;
|
||||
}
|
||||
|
||||
if (width, height) != last_size {
|
||||
last_size = (width, height);
|
||||
let _ = app.emit(
|
||||
VIDEO_EVENT,
|
||||
VideoPayload {
|
||||
active: true,
|
||||
width,
|
||||
height,
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
let Some((target_width, target_height)) = frames::next_target(width, height) else {
|
||||
continue;
|
||||
};
|
||||
|
||||
// to_argb converts, it does not resample, so the buffer has to be scaled
|
||||
// first. Doing it in I420 is cheaper than scaling four channels of RGBA.
|
||||
let mut i420 = frame.buffer.to_i420();
|
||||
let source = if (target_width, target_height) == (width, height) {
|
||||
i420
|
||||
} else {
|
||||
i420.scale(target_width as i32, target_height as i32)
|
||||
};
|
||||
|
||||
let stride = target_width * 4;
|
||||
rgba.resize((stride * target_height) as usize, 0);
|
||||
|
||||
// libyuv names formats after the 32 bit word, not the byte order. On little
|
||||
// endian its "RGBA" lands in memory as A,B,G,R, so taking the first three
|
||||
// bytes as red, green and blue picked up alpha as red and tinted everything.
|
||||
// "ABGR" is the one that lays out as R,G,B,A in memory.
|
||||
source.to_argb(
|
||||
VideoFormatType::ABGR,
|
||||
&mut rgba,
|
||||
stride,
|
||||
target_width as i32,
|
||||
target_height as i32,
|
||||
);
|
||||
|
||||
frames::send(&rgba, &mut rgb, &mut encoded, target_width, target_height);
|
||||
}
|
||||
|
||||
// The stream ended on its own rather than being replaced.
|
||||
if STREAM_EPOCH.load(Ordering::SeqCst) == epoch && is_current(generation) {
|
||||
stop(&app);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/// Clears the display only if the track that went away is the one being shown. A camera
|
||||
/// leaving must not blank a screen share that took over from it.
|
||||
pub fn stop_track(app: &AppHandle, sid: &str) {
|
||||
let is_current_track = CURRENT_SID
|
||||
.lock()
|
||||
.map(|guard| guard.as_deref() == Some(sid))
|
||||
.unwrap_or(false);
|
||||
|
||||
if is_current_track {
|
||||
stop(app);
|
||||
}
|
||||
}
|
||||
|
||||
pub fn stop(app: &AppHandle) {
|
||||
STREAM_EPOCH.fetch_add(1, Ordering::SeqCst);
|
||||
CURRENT_PRIORITY.store(0, Ordering::SeqCst);
|
||||
|
||||
if let Ok(mut guard) = CURRENT_SID.lock() {
|
||||
*guard = None;
|
||||
}
|
||||
|
||||
if !IS_STREAMING.swap(false, Ordering::SeqCst) {
|
||||
return;
|
||||
}
|
||||
|
||||
let _ = app.emit(
|
||||
VIDEO_EVENT,
|
||||
VideoPayload {
|
||||
active: false,
|
||||
width: 0,
|
||||
height: 0,
|
||||
},
|
||||
);
|
||||
}
|
||||
Reference in new issue
Block a user