gled-core 2.31.1

//! Send data to output devices: each rendered frame becomes one Art-Net
//! packet per routed universe (or a DMX frame for an Enttec interface),
//! merged with Art-Net arriving for the same recipient (the bridge).
//!
//! One thread, `gled:output:prepare`, turns a frame into packets: colour
//! order, channel overwrites, the bridge, the packet bytes. A few sender
//! threads make the system calls (see `senders`). A frame crosses between
//! them once, not once per packet.

mod artdmx;
#[cfg(test)]
mod bench;
pub(crate) mod enttec_usb_pro;
mod merge;
mod routes;
mod senders;

pub(crate) use crate::storage::asset::output_device::routing::Recipient;

use anyhow::{Context, Result};
use kanal::{Receiver, Sender, bounded};
use std::{
    net::UdpSocket,
    sync::Arc,
    thread,
    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
};
use tracing::debug;

use crate::{
    input::artnet::ARTNET_SOCKET,
    pipeline::extract_output::{OutputFrame, OutputFrames},
    storage::collections::Collections,
};
use gled_gpu::constants::UNIVERSE_BUFFER_SIZE;
use gled_model::asset::{output_device::routing::OutputRouting, project::io::ChannelOverwrites};
use merge::Merge;
use once_cell::sync::Lazy;
use parking_lot::Mutex;
use routes::Routes;
use senders::{Batch, Dropped, Senders, batch_for};

/// Art-Net for the bridge queued while a frame is prepared.
const INPUT_CAPACITY: usize = 1024;
/// How often bridged Art-Net nothing sends any more is forgotten.
const EXPIRY_INTERVAL: Duration = Duration::from_millis(250);

type Channels = [u8; UNIVERSE_BUFFER_SIZE as usize];

/// Channel values the user overrides, sent on top of the output.
static CHANNEL_OVERWRITES: Lazy<Mutex<ChannelOverwrites>> =
    Lazy::new(|| Mutex::new(ChannelOverwrites::default()));

/// The routing the output routings window points at. Its universe is left
/// out of the output, so the installation shows which lamps it drives.
pub(crate) static HOVERED_OUTPUT_ROUTING: Lazy<Mutex<Option<OutputRouting>>> =
    Lazy::new(|| Mutex::new(None));

/// The channel overwrites the output sends right now.
pub(crate) fn current_channel_overwrites() -> ChannelOverwrites {
    CHANNEL_OVERWRITES.lock().clone()
}

/// Replaces the channel overwrites the output sends.
pub(crate) fn set_channel_overwrites(overwrites: ChannelOverwrites) {
    *CHANNEL_OVERWRITES.lock() = overwrites;
}

/// Art-Net that arrived for a universe the bridge routes on, to merge with
/// gled's output for the same recipient.
#[derive(Debug)]
pub(crate) struct ArtnetInput {
    pub(crate) recipient: Recipient,
    pub(crate) data: Channels,
    pub(crate) htp: bool,
}

/// Where the Art-Net bridge hands the output thread what arrived.
#[derive(Debug, Clone)]
pub(crate) struct ArtnetInputs {
    inputs: Sender<ArtnetInput>,
    bell: Sender<()>,
}

impl ArtnetInputs {
    /// Queues `input` and wakes the output thread. Waits while the output
    /// thread has a full queue.
    pub(crate) fn send(&self, input: ArtnetInput) {
        if self.inputs.send(input).is_ok() {
            // A ring already pending wakes the thread for this one too.
            let _ = self.bell.try_send(());
        }
    }
}

/// Starts the output threads. Returns where the Art-Net bridge sends what
/// arrives.
pub(crate) fn start(frames: OutputFrames) -> Result<ArtnetInputs> {
    let OutputFrames { frames, wake, bell } = frames;
    enttec_usb_pro::start();

    debug!("Spawning output threads");
    let socket = ARTNET_SOCKET.clone();
    // Before any thread uses the socket: the Art-Net input thread relies on
    // it not blocking.
    match socket.set_broadcast(true) {
        Ok(_) => debug!("Activated sending to broadcast"),
        Err(e) => debug!("Could not activate sending to broadcast: {e}"),
    }
    match socket.set_nonblocking(true) {
        Ok(_) => debug!("Activated non-blocking mode"),
        Err(e) => debug!("Could not activate non-blocking mode: {e}"),
    };
    let senders = Senders::start(&socket)?;
    let (input_sender, inputs) = bounded(INPUT_CAPACITY);
    thread::Builder::new()
        .name("gled:output:prepare".to_owned())
        .spawn(move || {
            #[cfg(feature = "profiling")]
            profiling::register_thread!("output:prepare");

            Output {
                frames,
                wake,
                inputs,
                socket,
                senders,
                collections: None,
                routes: Routes::default(),
                merge: Merge::default(),
                expired: Instant::now(),
                packet: Vec::new(),
                dropped: Dropped::default(),
            }
            .run();
        })
        .context("Could not spawn the output thread")?;
    Ok(ArtnetInputs {
        inputs: input_sender,
        bell,
    })
}

/// The output thread's state.
struct Output {
    frames: Receiver<OutputFrame>,
    /// Rung after a frame or bridged Art-Net was queued.
    wake: Receiver<()>,
    inputs: Receiver<ArtnetInput>,
    socket: Arc<UdpSocket>,
    senders: Senders,
    /// Created with the first frame, so starting does not wait for the
    /// assets.
    collections: Option<Collections>,
    routes: Routes,
    merge: Merge,
    expired: Instant,
    /// The packet bridged Art-Net is sent in.
    packet: Vec<u8>,
    dropped: Dropped,
}

impl Output {
    fn run(mut self) {
        // Everything queued before a ring is taken after it, so no ring is
        // missed.
        while self.wake.recv().is_ok() {
            self.bridge_inputs();
            while let Ok(Some(frame)) = self.frames.try_recv() {
                self.send_frame(frame);
                self.bridge_inputs();
            }
            let now = Instant::now();
            if now.saturating_duration_since(self.expired) >= EXPIRY_INTERVAL {
                self.merge.expire(now);
                self.expired = now;
            }
        }
        debug!("Nothing hands the output frames any more, stopping it");
    }

    /// Sends what arrived for the Art-Net bridge, merged with gled's output.
    fn bridge_inputs(&mut self) {
        while let Ok(Some(input)) = self.inputs.try_recv() {
            let slot = self.routes.slot_of(&input.recipient);
            let data = self.merge.input(
                input.recipient.clone(),
                input.data,
                input.htp,
                slot,
                Instant::now(),
            );
            match &input.recipient {
                Recipient::Artnet { addr, universe } => {
                    self.packet.clear();
                    if artdmx::write(&mut self.packet, *universe, &data) {
                        let bytes =
                            senders::send(&self.socket, &self.packet, *addr, &mut self.dropped);
                        crate::network_stats::add_outgoing_bytes(bytes);
                    }
                }
                Recipient::EnttecDmxUsbPro { serial_number } => {
                    enttec_usb_pro::send(serial_number.clone(), data);
                }
            }
        }
    }

    /// Turns a frame into packets and hands them to the senders.
    #[cfg_attr(feature = "profiling", profiling::function)]
    fn send_frame(&mut self, frame: OutputFrame) {
        let Self {
            senders,
            collections,
            routes,
            merge,
            ..
        } = self;
        let OutputFrame { mut data, config } = frame;
        let collections = collections.get_or_insert_with(Collections::default);
        collections.update();
        routes.update(&config, collections);
        let hovered = HOVERED_OUTPUT_ROUTING.lock().clone();
        let mut overwrites = current_channel_overwrites();
        let now = Instant::now();

        let universes = data.as_chunks_mut::<{ UNIVERSE_BUFFER_SIZE as usize }>().0;
        let slots = universes.len().min(routes.len());
        let mut batches = senders.take(slots);
        let count = batches.len();
        for (slot, values) in universes.iter_mut().take(slots).enumerate() {
            let Some(route) = routes.slot(slot) else {
                continue;
            };
            if Some(&route.routing) == hovered.as_ref() {
                continue;
            }
            // Lamps that are not RGB read the triple in their own order;
            // overwrites stay as the user typed them.
            config.color_orders.correct(route.universe, values);
            overwrites.overwrite_data(route.device, route.routing.universe, values);
            let Some(recipient) = &route.recipient else {
                continue;
            };
            let out = merge.slot(slot, recipient, values, now);
            deliver(&mut batches[batch_for(slot, slots, count)], recipient, &out);
        }

        let last = &mut batches[count - 1];
        if let Some(recipient) = hovered.and_then(|routing| routing.recipient(collections)) {
            deliver(last, &recipient, &[blink(); UNIVERSE_BUFFER_SIZE as usize]);
        }
        for (routing, values) in overwrites.overwritten_universes() {
            if let Some(recipient) = routing.recipient(collections) {
                let out = merge.extra(recipient.clone(), values, now);
                deliver(last, &recipient, &out);
            }
        }
        senders.send(batches);
        merge.keep_frame(data);
    }
}

/// Adds the universe's packet to `batch`, or hands it to the Enttec thread.
fn deliver(batch: &mut Batch, recipient: &Recipient, data: &Channels) {
    match recipient {
        Recipient::Artnet { addr, universe } => batch.push(*addr, *universe, data),
        Recipient::EnttecDmxUsbPro { serial_number } => {
            enttec_usb_pro::send(serial_number.clone(), *data);
        }
    }
}

/// The value every channel of the hovered universe blinks with: a triangle
/// wave over about half a second.
fn blink() -> u8 {
    let millis = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_or(0, |time| time.as_millis());
    (millis % 512).abs_diff(256) as u8
}