gled-core 2.32.0

use crate::storage::asset::project::Project;
pub(crate) use crate::storage::asset::project::io::{ArtnetConfig, ArtnetControlConfig};
use crate::{
    pipeline::output_sender::{ArtnetInput, ArtnetInputs},
    storage::collections::Collections,
};
use artnet_protocol::{ArtCommand, PollReply, PortAddress};
use chrono::Utc;
use gled_gpu::constants::UNIVERSE_BUFFER_SIZE;
use kanal::{Receiver, unbounded};
use once_cell::sync::Lazy;
use parking_lot::Mutex;
use std::net::IpAddr;
use std::ops::Shl;
use std::{
    net::{Ipv4Addr, UdpSocket},
    sync::Arc,
    thread,
    time::Duration,
};
use sysinfo::Networks;
use tracing::{debug, trace, warn};

static ARTNET_PORT: u16 = 6454;
pub(crate) static ARTNET_SOCKET: Lazy<Arc<UdpSocket>> = Lazy::new(|| {
    // Another Art-Net program on this machine may hold the port. Sending
    // works from any port, so gled keeps its output and only loses Art-Net
    // input then.
    let socket = UdpSocket::bind(("0.0.0.0", ARTNET_PORT)).unwrap_or_else(|err| {
        warn!(
            "Art-Net port {ARTNET_PORT} is taken ({err}), probably by another Art-Net program: \
             sending from another port, Art-Net input is off"
        );
        UdpSocket::bind(("0.0.0.0", 0)).expect("Could not bind any UDP port for Art-Net")
    });
    // Enlarge the kernel send buffer. At hundreds of fps × dozens of universes
    // the default ~200 KB buffer overflows instantly, forcing every send into a
    // WouldBlock retry spin that throttled output throughput to ~20 Mbit.
    if let Err(e) = socket2::SockRef::from(&socket).set_send_buffer_size(8 * 1024 * 1024) {
        warn!("Could not enlarge artnet socket send buffer: {e}");
    }
    Arc::new(socket)
});

#[derive(Debug)]
pub(crate) struct ArtnetEvent {
    pub(crate) channel: u16,
    pub(crate) value: u8,
}

pub(crate) static ARTNET_CONFIG: Lazy<Mutex<ArtnetConfig>> =
    Lazy::new(|| Mutex::new(Default::default()));

/// Whether data on `port_address` is for the control universe. A control
/// universe past what a port address holds (set over OSC, say) matches nothing.
fn is_control_universe(port_address: PortAddress, control_universe: u16) -> bool {
    PortAddress::try_from(control_universe).is_ok_and(|control| port_address == control)
}

/// The Art-Net port address of the bridged input universe.
fn port_address(config: &ArtnetConfig) -> PortAddress {
    PortAddress::try_from(config.universe).unwrap_or_else(|_| PortAddress::from(0))
}

/// Changes the project's Art-Net control settings and applies them to the
/// running Art-Net input.
pub(crate) fn set_artnet_control_config(
    project: &mut Project,
    apply: impl FnOnce(&mut ArtnetControlConfig),
) {
    apply(&mut project.artnet_control_config);
    ARTNET_CONFIG.lock().artnet_control_config = project.artnet_control_config;
}

pub(crate) fn start_thread(
    output_sender: ArtnetInputs,
) -> (Receiver<ArtnetEvent>, Receiver<Vec<u8>>) {
    let (bridge_sender, bridge_receiver) = unbounded();
    let (control_sender, control_receiver) = unbounded();

    trace!("Opening udp sockets on artnet port");

    let bridge_sender = bridge_sender.clone();
    let control_sender = control_sender.clone();
    thread::Builder::new()
        .name("gled:artnet:rx".to_string())
        .spawn(move || {
            #[cfg(feature = "profiling")]
            profiling::register_thread!("artnet:rx");

            let mut collections = Collections::default();

            trace!("Starting artnet receiving");
            trace!("Setting broadcast");
            ARTNET_SOCKET
                .set_broadcast(true)
                .expect("Could not set broadcast");
            let mut buf = [0; 4096];
            loop {
                collections.update();

                trace!("Receiving artnet package");
                let Ok((size, src)) = ARTNET_SOCKET.recv_from(&mut buf) else {
                    debug!("Could not receive on artnet");
                    std::thread::sleep(Duration::from_millis(10));
                    continue;
                };
                trace!("received on artnet");

                crate::network_stats::add_incoming_bytes(size);

                let output = match ArtCommand::from_buffer(&buf[..size]) {
                    Err(err) => {
                        warn!("Could not parse artnet package: {err:?}");
                        continue;
                    }
                    Ok(ArtCommand::Output(output)) => output,
                    Ok(ArtCommand::Poll(_poll)) => {
                        debug!("Artnet poll from {src:?}");

                        let networks = Networks::new_with_refreshed_list();
                        let bind_addr = &networks
                            .values()
                            .flat_map(|network| network.ip_networks().iter())
                            .find_map(|ip_network| match (ip_network.addr, src.ip()) {
                                (IpAddr::V4(interface_addr), IpAddr::V4(src_addr)) => {
                                    let netmask = u32::MAX.shl(32 - ip_network.prefix);
                                    ((interface_addr.to_bits() & netmask)
                                        == (src_addr.to_bits() & netmask))
                                        .then_some(interface_addr)
                                }
                                _ => None,
                            });

                        if let Some(bind_addr) = bind_addr {
                            let poll_reply = PollReply {
                                address: match src.ip() {
                                    std::net::IpAddr::V4(ip) => ip,
                                    _ => Ipv4Addr::LOCALHOST,
                                },
                                port: ARTNET_PORT,
                                version: [0, 0],
                                port_address: [0, 0],
                                oem: [0, 0],
                                ubea_version: 0,
                                status_1: 210,
                                esta_code: 31344,
                                short_name: {
                                    let mut bytes = [0; 18];
                                    bytes[..5].clone_from_slice("Gled2".as_bytes());
                                    bytes
                                },
                                long_name: {
                                    let mut bytes = [0; 64];
                                    bytes[..18].clone_from_slice("Gled2 Artnet Input".as_bytes());
                                    bytes
                                },
                                node_report: [0; 64],
                                num_ports: [0, 4],
                                port_types: [192; 4],
                                good_input: [0, 8, 8, 8],
                                good_output: [0, 0, 0, 0],
                                swin: [0; 4],
                                swout: [0; 4],
                                sw_video: 0,
                                sw_macro: 0,
                                sw_remote: 0,
                                spare: [0; 3],
                                style: 0,
                                mac: [0; 6],
                                bind_ip: bind_addr.octets(),
                                bind_index: 0,
                                status_2: 0,
                                filler: [0; 26],
                            };

                            let Ok(data) =
                                ArtCommand::PollReply(Box::new(poll_reply)).write_to_buffer()
                            else {
                                warn!("Could not create poll reply");
                                continue;
                            };

                            crate::network_stats::add_outgoing_bytes(data.len());
                            match ARTNET_SOCKET.send_to(&data, src) {
                                Err(err) => {
                                    warn!("Could not send PollReply: {err:?}");
                                }
                                Ok(count) => {
                                    crate::network_stats::add_outgoing_bytes(count);
                                }
                            }
                        }

                        continue;
                    }
                    Ok(artnet) => {
                        debug!("Unhandeled ArtCommand: {artnet:?}");
                        continue;
                    }
                };
                trace!("parsed artnet");

                let mut config = ARTNET_CONFIG.lock();
                // dispatch trigger packages
                if output.port_address == port_address(&config) {
                    trace!("artnet data on input universe");
                    let data = output.data.as_ref();
                    data.iter().enumerate().for_each(|(channel, value)| {
                        bridge_sender
                            .send(ArtnetEvent {
                                channel: channel as u16,
                                value: *value,
                            })
                            .expect("Could not send event")
                    });
                }
                // dispatch control packages
                if config.artnet_control_config.active
                    && is_control_universe(
                        output.port_address,
                        config.artnet_control_config.universe,
                    )
                {
                    trace!("artnet data on control universe");
                    control_sender
                        .send(output.data.as_ref().to_vec())
                        .expect("Could not send event");
                }
                // dispatch all artnet data to the bridge
                let bridge = config.bridge.entry(output.port_address.into()).or_default();
                bridge.last_data_at = Utc::now();
                bridge.updates += 1;

                if let Some(recipient) = bridge.output_routing.recipient(&collections) {
                    trace!("artnet bridge has output routing");

                    let mut data = [0u8; UNIVERSE_BUFFER_SIZE as usize];
                    data.copy_from_slice(output.data.as_ref());

                    output_sender.send(ArtnetInput {
                        recipient,
                        data,
                        htp: bridge.htp,
                    });
                }
            }
        })
        .expect("Could not spawn artnet receive thread");

    (bridge_receiver, control_receiver)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn a_control_universe_past_the_port_addresses_matches_nothing() {
        let port = PortAddress::from(7);
        assert!(is_control_universe(port, 7));
        assert!(!is_control_universe(port, 8));
        assert!(!is_control_universe(PortAddress::from(0), u16::MAX));
    }
}