gled_core/engine/

mod.rs

//! The engine: the one place that owns the running show. It holds the
//! authoritative project, applies actions from the user interfaces, OSC, MIDI
//! and Art-Net, evaluates the inputs, keeps time with Ableton Link, drives the
//! render thread and the audio analysis, and publishes its state to every
//! connected user interface.
//!
//! It runs on its own thread at a fixed tick rate, independent of any window.
//! The desktop app talks to it through an `EngineHandle`; a remote client will
//! talk to it the same way over the network.
//!
//! User interfaces change the show only through actions (see
//! `ui::project_edits` for edits made directly in widgets). Their actions are
//! numbered, and the engine reports the newest one it applied in
//! `EngineState::applied_sequence`, so an interface knows when the engine's
//! project includes its own edits.

mod actions;
mod applied_output;
mod devices;
mod event_log;
#[cfg(any(test, feature = "test-util"))]
mod fake;
mod midi_tools;
mod outbox;
mod session;
mod status;
mod svg_layout;
mod svg_templates;

pub(crate) use event_log::{EventLog, InputLog, LearnLog, MidiLog};
#[cfg(any(test, feature = "test-util"))]
pub use fake::{FakeEngine, fake_engine};
use gled_model::sound::trigger_ref::SoundLevels;
use gled_protocol::bytes::SharedBytes;
pub use gled_protocol::state::{
    ClientInput, Devices, EngineEvent, EngineState, RenderStats, ServerSettings, ServerStatus,
    SoundSnapshot, StorageCommand,
};
pub use outbox::Outbox;
pub use session::{ClientConnection, ClientId, ClientState, EngineConnector};

use std::{
    collections::{BTreeSet, HashMap},
    sync::Arc,
    time::{Duration, Instant},
};

use kanal::{Receiver, unbounded};
use tracing::debug;

use crate::{
    action::UiAction,
    audio::{
        AudioPool,
        sound_data::SoundData,
        sound_trigger_ref::{CollectSoundTriggers, SoundTriggerSet},
    },
    input::{
        Input, external_control::ExternalControlState, osc::OSCHandler, osc::OscConfig,
        project_inputs::ProjectInputs,
    },
    output_state::ProjectState,
    pipeline::extract_output::{ExtractOutput, OutputConfig, PreviewOutput},
    pipeline::output_sender::HOVERED_OUTPUT_ROUTING,
    runtime::{
        persistent_state::PersistentState,
        render_thread::{RenderFeedback, RenderSnapshot, RenderThread},
        timing::Timing,
    },
    storage::{
        action::StorageAction,
        asset::{Asset, project::Project, scene::Scene},
        asset_id::AssetId,
        collections::Collections,
        is_loading,
        store::{AssetStore, delete_asset},
    },
};
use applied_output::AppliedOutput;
use gled_model::{
    asset::{AssetTrait, output_device::routing::OutputRouting, scene::grid::GridLocation},
    collection::Collections as AssetCollections,
};
use gled_protocol::{
    assets::{AssetIdVisitor, AssetVisitor},
    midi::{MidiMonitorEvent, TestCommand},
};
use svg_layout::SvgLayout;
use uuid::Uuid;

/// The engine's tick rate. Inputs are evaluated and state is published once
/// per tick; the LED output runs at its own rate on the render thread.
///
/// 8333 us per tick is about 120 updates per second.
const TICK: Duration = Duration::from_micros(8_333);
/// Sound levels are published every this many ticks, 30 times a second.
const SOUND_EVERY_TICKS: u32 = 4;

/// What the engine needs from its creator.
#[derive(Debug)]
pub(crate) struct EngineConfig {
    /// GPU adapter for the render thread's output device.
    pub(crate) adapter: wgpu::Adapter,
    pub(crate) extract_output: ExtractOutput,
    pub(crate) artnet_control_receiver: Receiver<Vec<u8>>,
    /// The rendered output, read back for the preview.
    pub(crate) preview_output: Receiver<PreviewOutput>,
    /// Art-Net and DMX traffic in Mbit/s, in and out.
    pub(crate) network_stats: Receiver<(f64, f64)>,
    /// The machine's hottest temperature sensor in °C, once a second.
    pub(crate) temperature: Receiver<Option<f32>>,
    pub(crate) midi_monitor: Receiver<MidiMonitorEvent>,
    pub(crate) midi_learn: Receiver<[u8; 3]>,
    pub(crate) midi_test: kanal::Sender<TestCommand>,
    pub(crate) audio_pool: AudioPool,
}

/// Starts the engine thread. It runs until every connector and connection is
/// dropped.
pub(crate) fn spawn(
    config: EngineConfig,
    actions: Receiver<UiAction>,
) -> anyhow::Result<EngineConnector> {
    let timing = Timing::default();
    let persistent_state = PersistentState::default();
    let render_thread = RenderThread::spawn(
        &config.adapter,
        timing.link(),
        config.extract_output,
        persistent_state.fps_limit(),
    )?;

    let (input_sender, input_receiver) = unbounded();

    let engine = Engine {
        startup: true,
        timing,
        project: None,
        project_id: None,
        selected_scene_instance: GridLocation::default(),
        blackout: true,
        blackout_hold: false,
        midi_output_active: false,
        actions,
        inputs: input_receiver,
        sessions: HashMap::new(),
        collections: Collections::default(),
        persistent_state,
        output_config: OutputConfig::default(),
        render_thread,
        external_control_state: ExternalControlState::new(config.artnet_control_receiver),
        audio_pool: config.audio_pool,
        sound_data: SoundData::default(),
        osc_handler: None,
        osc_config: None,
        applied_output: AppliedOutput::default(),
        hovered_routing: None,
        input_log: Arc::default(),
        device_scan: devices::DeviceScan::default(),
        svg_templates: svg_templates::SvgTemplates::default(),
        watched_triggers: HashMap::new(),
        midi: midi_tools::MidiTools::new(config.midi_monitor, config.midi_learn, config.midi_test),
        svg_layout: SvgLayout::default(),
        render_stats: RenderStats::default(),
        published_assets: (u64::MAX, Arc::default()),
        preview_output: config.preview_output,
        network_stats_receiver: config.network_stats,
        network_stats: (0.0, 0.0),
        temperature_receiver: config.temperature,
        temperature: None,
        host_name: gethostname::gethostname().to_string_lossy().into_owned(),
        sound: Arc::default(),
        ticks_since_sound: 0,
        preview: None,
    };

    std::thread::Builder::new()
        .name("gled:engine".to_string())
        .spawn(move || engine.run())?;

    Ok(EngineConnector {
        inputs: input_sender,
        next_id: Arc::default(),
    })
}

pub(crate) struct Engine {
    /// Load the last project once storage has finished loading.
    startup: bool,
    timing: Timing,
    project: Option<Project>,
    project_id: Option<AssetId<Project>>,
    selected_scene_instance: GridLocation,
    blackout: bool,
    blackout_hold: bool,
    midi_output_active: bool,
    actions: Receiver<UiAction>,
    inputs: Receiver<session::EngineInput>,
    sessions: HashMap<ClientId, session::Session>,
    collections: Collections,
    persistent_state: PersistentState,
    output_config: OutputConfig,
    render_thread: RenderThread,
    external_control_state: ExternalControlState,
    audio_pool: AudioPool,
    sound_data: SoundData,
    osc_handler: Option<Arc<OSCHandler>>,
    /// The OSC settings the handler runs with.
    osc_config: Option<OscConfig>,
    /// The project's output settings as last applied to the running output.
    applied_output: AppliedOutput,
    device_scan: devices::DeviceScan,
    svg_templates: svg_templates::SvgTemplates,
    /// The sound triggers each interface edits.
    watched_triggers: HashMap<ClientId, BTreeSet<Uuid>>,
    midi: midi_tools::MidiTools,
    /// The input events of the last ticks, for assigning inputs.
    input_log: Arc<InputLog>,
    /// The output routing an interface points at, and which one.
    hovered_routing: Option<(ClientId, OutputRouting)>,
    svg_layout: SvgLayout,
    render_stats: RenderStats,
    preview_output: Receiver<PreviewOutput>,
    network_stats_receiver: Receiver<(f64, f64)>,
    network_stats: (f64, f64),
    temperature_receiver: Receiver<Option<f32>>,
    temperature: Option<f32>,
    /// This machine's name, which interfaces on others show it under.
    host_name: String,
    sound: Arc<SoundSnapshot>,
    ticks_since_sound: u32,
    /// The newest preview frame; `None` while no project renders.
    preview: Option<PreviewOutput>,
    /// The assets as last published, and the replica's generation then.
    published_assets: (u64, Arc<AssetCollections>),
}

impl Engine {
    fn run(mut self) {
        let mut next_tick = Instant::now();
        loop {
            if !self.tick() {
                debug!("All user interfaces are gone, stopping the engine");
                return;
            }
            next_tick += TICK;
            let now = Instant::now();
            match next_tick.checked_duration_since(now) {
                Some(wait) => std::thread::sleep(wait),
                // Behind schedule, e.g. after loading a project: start over
                // instead of running a burst of ticks.
                None => next_tick = now,
            }
        }
    }

    /// One engine step. Returns `false` once no user interface is connected.
    #[cfg_attr(feature = "profiling", profiling::function)]
    fn tick(&mut self) -> bool {
        self.receive_render_feedback();
        self.persistent_state.update();
        self.sound_data.update();
        self.timing.tick(&self.persistent_state);
        self.collections.update();
        self.sync_sound_triggers();

        if !is_loading() && self.startup {
            self.startup = false;
            if let Some(project) = self.persistent_state.last_project_id() {
                UiAction::SetProject(project).enqueue();
            }
        }

        if !self.receive_client_inputs() {
            return false;
        }
        Input::tick();
        self.midi.tick();
        let new_events = Input::new_events();
        if !new_events.is_empty() {
            let log = Arc::make_mut(&mut self.input_log);
            for event in new_events {
                log.push(event);
            }
        }
        self.handle_actions();

        self.external_control_state
            .process_events(&mut self.project, &self.collections);
        self.evaluate_triggers();
        if let Some(project) = &mut self.project {
            project.update_input_state(self.timing.beat_progression());
        }
        self.apply_project_settings();

        self.render_thread.send_snapshot(RenderSnapshot {
            project: self.project.clone(),
            blackout: self.blackout || self.blackout_hold,
            output_config: self.output_config.clone(),
            fps_limit: self.persistent_state.fps_limit(),
        });

        ProjectState {
            project: self.project.clone(),
            selected_scene_instance: self.selected_scene_instance,
            blackout: self.blackout || self.blackout_hold,
            beats_per_minute: self.timing.beats_per_minute(),
            beat_progression: self.timing.beat_progression(),
        }
        .enqueue();

        self.publish_state();
        true
    }

    /// Applies what the user interfaces sent. Returns `false` once no
    /// connector or connection is left.
    fn receive_client_inputs(&mut self) -> bool {
        loop {
            match self.inputs.try_recv() {
                Ok(Some(session::EngineInput::Connect { id, state, events })) => {
                    debug!(id, "client connected");
                    self.sessions.insert(
                        id,
                        session::Session {
                            state,
                            events,
                            applied_sequence: 0,
                            keys: BTreeSet::new(),
                        },
                    );
                }
                Ok(Some(session::EngineInput::Client { id, input })) => {
                    self.apply_client_input(id, input)
                }
                Ok(Some(session::EngineInput::Disconnect(id))) => self.disconnect(id),
                Ok(None) => return true,
                Err(_) => return false,
            }
        }
    }

    fn apply_client_input(&mut self, id: ClientId, input: ClientInput) {
        match input {
            ClientInput::Action { sequence, action } => {
                if let Some(session) = self.sessions.get_mut(&id) {
                    session.applied_sequence = session.applied_sequence.max(sequence);
                }
                self.apply_action(action);
            }
            ClientInput::Keys(keys) => {
                debug!(id, ?keys, "keys held in the interface");
                if let Some(session) = self.sessions.get_mut(&id) {
                    session.keys = keys;
                }
                self.update_keys();
            }
            ClientInput::SetBeatsPerMinute(bpm) => self.timing.set_beats_per_minute(bpm),
            ClientInput::SaveAsset(asset) => asset.accept(&mut SaveInto(&mut self.collections)),
            ClientInput::DeleteAsset(id) => id.accept(&mut DeleteFrom(&mut self.collections)),
            ClientInput::Storage(command) => match command {
                StorageCommand::Commit { message } => StorageAction::Commit { message }.enqueue(),
                StorageCommand::Pull => StorageAction::Pull.enqueue(),
                StorageCommand::Push => StorageAction::Push.enqueue(),
                StorageCommand::SwitchBranch(branch) => {
                    StorageAction::SwitchBranch(branch).enqueue()
                }
                StorageCommand::Reclone => StorageAction::Nuke.enqueue(),
                StorageCommand::SetGitPassphrase(passphrase) => self
                    .persistent_state
                    .git_credentials_mut()
                    .set_passphrase(passphrase),
            },
            ClientInput::SetServerSettings(settings) => self.apply_server_settings(settings),
            ClientInput::WatchDevices(watch) => self.device_scan.watch(id, watch),
            ClientInput::WatchMidi(watch) => self.midi.watch(id, watch),
            ClientInput::MidiLearning(learning) => self.midi.learning(id, learning),
            ClientInput::MidiTest(command) => self.midi.test(command),
            ClientInput::UseSvgTemplate(name) => match svg_templates::read(&name) {
                Ok(contents) => self.apply_action(UiAction::SetSvg(Some(
                    gled_model::asset::svg::Svg::new(contents),
                ))),
                Err(err) => self.send_event(EngineEvent::Error(err)),
            },
            ClientInput::WatchSoundTriggers(triggers) => {
                self.watched_triggers.insert(id, triggers);
            }
            ClientInput::HoverOutputRouting(routing) => {
                let owned = self
                    .hovered_routing
                    .as_ref()
                    .is_some_and(|(owner, _)| *owner == id);
                match routing {
                    Some(routing) => self.set_hovered_routing(Some((id, routing))),
                    None if owned => self.set_hovered_routing(None),
                    None => {}
                }
            }
        }
    }

    fn disconnect(&mut self, id: ClientId) {
        if self.sessions.remove(&id).is_some() {
            debug!(id, "client disconnected");
            // Keys it held are released, and a routing it pointed at.
            self.update_keys();
            self.device_scan.disconnect(id);
            self.midi.disconnect(id);
            self.watched_triggers.remove(&id);
            if self
                .hovered_routing
                .as_ref()
                .is_some_and(|(owner, _)| *owner == id)
            {
                self.set_hovered_routing(None);
            }
        }
    }

    fn set_hovered_routing(&mut self, hovered: Option<(ClientId, OutputRouting)>) {
        *HOVERED_OUTPUT_ROUTING.lock() = hovered.as_ref().map(|(_, routing)| routing.clone());
        self.hovered_routing = hovered;
    }

    /// A key counts as held while any interface holds it.
    fn update_keys(&self) {
        let keys = self
            .sessions
            .values()
            .flat_map(|session| session.keys.iter().copied())
            .collect();
        Input::set_keys(keys);
    }

    /// The project's own trigger inputs: tap, blackout and tempo.
    fn evaluate_triggers(&mut self) {
        let Some(project) = &self.project else {
            self.blackout_hold = false;
            return;
        };
        if project.blackout_input_is_new() {
            self.blackout = !self.blackout;
        }
        self.blackout_hold = project.blackout_hold_input_is_live();
        if project.tap_input_is_new() {
            self.timing.tap(&self.persistent_state);
        }
        if project.double_input_is_new() {
            self.timing.multiply_speed(2.0, &self.persistent_state);
        }
        if project.half_input_is_new() {
            self.timing.multiply_speed(0.5, &self.persistent_state);
        }
    }

    /// Brings everything that follows project settings in line with the
    /// project, whichever way it changed: the SVG's lamp layout and OSC.
    fn apply_project_settings(&mut self) {
        let svg = self
            .project
            .as_ref()
            .and_then(|project| project.svg.as_ref());
        if self.svg_layout.update(svg, &mut self.output_config) {
            self.render_thread
                .send_command(crate::runtime::render_thread::RenderCommand::SendPositions);
        }

        if let Some(project) = &self.project {
            self.applied_output.apply(project);
        }

        let osc_config = self.project.as_ref().map(|project| project.osc_config);
        if osc_config != self.osc_config {
            self.osc_config = osc_config;
            self.apply_osc_config(osc_config.unwrap_or_default());
        }
    }

    fn apply_osc_config(&mut self, config: OscConfig) {
        if let Some(handler) = self.osc_handler.take() {
            handler.stop();
        }
        if config.active {
            self.osc_handler = OSCHandler::start(config.port)
                .map_err(|err| {
                    UiAction::Error(format!(
                        "Could not start OSC on port {}: {err}",
                        config.port
                    ))
                    .enqueue();
                })
                .ok();
        }
    }

    /// Keeps the running sound triggers in step with the triggers the project
    /// and the scene assets use. Cheap when nothing changed.
    fn sync_sound_triggers(&mut self) {
        let mut wanted = SoundTriggerSet::new();
        self.project.collect_sound_triggers(&mut wanted);
        for scene in Asset::<Scene>::all(&self.collections) {
            scene.data.collect_sound_triggers(&mut wanted);
        }
        if self.sound_data.sync_triggers(&wanted) {
            self.sound_data.save();
        }
    }

    fn receive_render_feedback(&mut self) {
        while let Ok(Some(stats)) = self.network_stats_receiver.try_recv() {
            self.network_stats = stats;
        }
        while let Ok(Some(temperature)) = self.temperature_receiver.try_recv() {
            self.temperature = temperature;
        }
        while let Ok(Some(output)) = self.preview_output.try_recv() {
            self.preview = Some(output);
        }
        if self.project.is_none() {
            self.preview = None;
        }
        while let Some(feedback) = self.render_thread.try_recv_feedback() {
            match feedback {
                RenderFeedback::FrameStats(stats) => {
                    self.timing.set_framerate(stats.fps);
                    self.render_stats = stats;
                }
            }
        }
    }

    fn publish_state(&mut self) {
        let shared = Arc::new(EngineState {
            project: self.project.clone(),
            project_id: self.project_id,
            selected_scene_instance: self.selected_scene_instance,
            blackout: self.blackout,
            blackout_hold: self.blackout_hold,
            midi_output_active: self.midi_output_active,
            output_config: self.output_config.clone(),
            beat: self.timing.beat_state(),
            tap_count: self.timing.tap_count(),
            layout_generation: self.svg_layout.generation(),
            render: self.render_stats,
            status: self.server_status(),
        });
        self.ticks_since_sound += 1;
        if self.ticks_since_sound >= SOUND_EVERY_TICKS {
            self.ticks_since_sound = 0;
            let watched: BTreeSet<Uuid> =
                self.watched_triggers.values().flatten().copied().collect();
            self.sound = Arc::new(SoundSnapshot {
                levels: self.sound_data.trigger_levels().collect(),
                impulses: self
                    .sound_data
                    .triggers
                    .iter()
                    .map(|(id, trigger)| (*id, trigger.impulse))
                    .collect(),
                spectrum: SharedBytes(Arc::new(self.sound_data.spectrum_bytes().to_vec())),
                trigger_spectra: watched
                    .iter()
                    .filter_map(|id| {
                        let trigger = self.sound_data.triggers.get(id)?;
                        let bytes = trigger
                            .spectrum
                            .iter()
                            .flat_map(|value| value.to_le_bytes())
                            .collect();
                        Some((*id, SharedBytes(Arc::new(bytes))))
                    })
                    .collect(),
            });
        }
        if self.published_assets.0 != self.collections.generation() {
            self.published_assets = (
                self.collections.generation(),
                Arc::new(AssetCollections::clone(&self.collections)),
            );
        }
        let mut gone = Vec::new();
        for (id, session) in &self.sessions {
            // A full channel means the client has not taken the last states
            // yet; it only ever wants the newest, so this one is dropped.
            let state = ClientState {
                shared: shared.clone(),
                applied_sequence: session.applied_sequence,
                assets: self.published_assets.1.clone(),
                preview: self.preview.clone(),
                sound: self.sound.clone(),
                input_log: self.input_log.clone(),
                midi_log: self.midi.monitor_log.clone(),
                learn_log: self.midi.learn_log.clone(),
            };
            if session.state.try_send(state).is_err() {
                gone.push(*id);
            }
        }
        for id in gone {
            self.disconnect(id);
        }
    }

    fn send_event(&self, event: EngineEvent) {
        for session in self.sessions.values() {
            // A closed channel is noticed by the next `publish_state`.
            let _ = session.events.send(event.clone());
        }
    }
}

/// Saves an asset a client sent, of whatever kind.
struct SaveInto<'a>(&'a mut Collections);

impl AssetVisitor for SaveInto<'_> {
    fn asset<T: AssetTrait>(&mut self, asset: Arc<Asset<T>>) {
        Arc::unwrap_or_clone(asset).save(self.0);
    }
}

/// Deletes an asset a client named, of whatever kind.
struct DeleteFrom<'a>(&'a mut Collections);

impl AssetIdVisitor for DeleteFrom<'_> {
    fn id<T: AssetTrait>(&mut self, id: AssetId<T>) {
        delete_asset(id, self.0);
    }
}