audiofox 1.1.0

// SPDX-License-Identifier: GPL-2.0-or-later WITH Classpath-exception-2.0
// Audiofox, the audio capturer!
// Copyright (c) 2026 foxberries

use std::cell::OnceCell;
use std::io::Cursor;

use anyhow::{Context, anyhow};
use log::{info, trace};
use pw::keys;
use pw::link::Link;
use pw::properties::properties;
use pw::spa::param::ParamType;
use pw::spa::param::audio::{AudioFormat, AudioInfoRaw};
use pw::spa::pod::serialize::PodSerializer;
use pw::spa::pod::{Object, Pod, Value};
use pw::spa::sys::{SPA_AUDIO_CHANNEL_FL, SPA_AUDIO_CHANNEL_FR, SPA_AUDIO_MAX_CHANNELS};
use pw::spa::utils::{Direction, SpaTypes};
use pw::stream::StreamState::Unconnected;
use pw::stream::{Stream, StreamFlags, StreamListener, StreamRc};
use uuid::Uuid;

use crate::common::audiofox::AUDIOFOX;
use crate::common::callback::AudioCallback;
use crate::common::chunk::Chunk;
use crate::common::result::Result;
use crate::common::stream::{CHANNELS, SAMPLE_RATE};
use crate::platform::linux::registry::{PortName, RegId, RegName};
use crate::platform::platform::command::CmdListener;

/// Values used to initialize and maintain a stream.
#[derive(Default)]
pub struct StreamState {
    // Audiofox node
    pub id: RegId,
    pub in_left: PortName,
    pub in_right: PortName,

    // Link targets
    pub names: Vec<RegName>,
    pub ids: Vec<RegId>,

    // Keep-alive
    pub links: Vec<Link>,
    listener: OnceCell<StreamListener<AudioInfoRaw>>,
    stream: OnceCell<StreamRc>,
}

impl StreamState {
    /// Initialize a new stream.
    pub fn start(&self, listener: &CmdListener, uuid: Uuid, callback: AudioCallback, frames: usize) -> Result<()> {
        // Define a new Stream object
        let stream = StreamRc::new(
            listener.core.clone(), AUDIOFOX, properties! {
                *keys::NODE_ID => uuid.to_string(), // Must be string!
                *keys::NODE_NAME => AUDIOFOX,
                *keys::MEDIA_CATEGORY => "Capture",
                *keys::MEDIA_ROLE => "Music",
                *keys::MEDIA_TYPE => "Audio",
                *keys::NODE_LATENCY => format!("{frames}/{SAMPLE_RATE}"),
            }).context("Failed to create audio stream")?;

        // Configure this stream's event handlers
        let stream_listener = stream
            .add_local_listener_with_user_data(AudioInfoRaw::default())
            .param_changed(Self::on_update())
            .process(Self::on_buffer(callback))
            .state_changed(Self::on_state_change(listener.clone()))
            .register()
            .context("Failed to register stream listener")?;

        // Finalize the stream configuration
        Self::finalize(&stream)?;

        // Store the stream and its listener
        self.stream.set(stream).map_err(|_| anyhow!("Stream already initialized"))?;
        self.listener.set(stream_listener).map_err(|_| anyhow!("Stream listener already initialized"))?;

        Ok(())
    }

    /// Called when an audio parameter is updated.
    fn on_update() -> impl Fn(&Stream, &mut AudioInfoRaw, u32, Option<&Pod>) {
        move |_stream: &Stream, user_data, id, param| {
            if let Some(param) = param && id == ParamType::Format.as_raw() && user_data.parse(param).is_ok() {
                info!(
                    "capturing at rate={}, channels={}",
                    user_data.rate(), user_data.channels()
                );
            }
        }
    }

    /// Called when the stream receives a new audio buffer.
    fn on_buffer(callback: AudioCallback) -> impl FnMut(&Stream, &mut AudioInfoRaw) {
        move |stream, _user_data: &mut AudioInfoRaw| match stream.dequeue_buffer() {
            None => trace!("no buffer available"),
            Some(mut buffer) => {
                let datas = buffer.datas_mut();

                // Skip empty buffers
                if datas.is_empty() {
                    return;
                }

                let data = &mut datas[0];
                let size = data.chunk().size() as usize;

                // Partition the buffer with the given chunk size
                if let Some(sample_slice) = data.data() {
                    callback(Chunk::from(&sample_slice[..size]));
                }
            }
        }
    }

    /// Called when the state of the stream is changed.
    fn on_state_change(listener: CmdListener) -> impl FnMut(&Stream, &mut AudioInfoRaw, pw::stream::StreamState, pw::stream::StreamState) {
        move |stream, _user_data, _old, new| {
            // If the stream node is destroyed, clean up
            if new == Unconnected {
                let uuid: Uuid = stream.properties().get(*keys::NODE_SESSION)
                    .expect("Stream should always have a UUID")
                    .parse()
                    .expect("Stream UUID should never be invalid");
                listener.on_end(uuid);
            }
        }
    }

    /// Finalize a new stream by configuring its low-level parameters and establishing a connection.
    fn finalize(stream: &StreamRc) -> Result<()> {
        // Define the stream's parameters
        let mut audio_info = AudioInfoRaw::new();

        // Set the stream format
        audio_info.set_format(AudioFormat::F32LE);

        // Set the audio positions
        let mut positions = [0u32; SPA_AUDIO_MAX_CHANNELS as usize];
        positions[0] = SPA_AUDIO_CHANNEL_FL;
        positions[1] = SPA_AUDIO_CHANNEL_FR;
        audio_info.set_position(positions);

        // Set the channel count
        audio_info.set_channels(CHANNELS as u32);

        // Serialize into a Pod
        let props = PodSerializer::serialize(
            Cursor::new(Vec::new()),
            &Value::Object(Object {
                type_: SpaTypes::ObjectParamFormat.as_raw(),
                id: ParamType::EnumFormat.as_raw(),
                properties: audio_info.into(),
            })).context("Failed to serialize audio properties")?
            .0.into_inner();

        // Connect to the stream, which will automatically start once linking completes
        stream.connect(
            Direction::Input,
            None,
            StreamFlags::MAP_BUFFERS | StreamFlags::RT_PROCESS,
            &mut [Pod::from_bytes(&props).context("Failed to create props Pod")?],
        ).context("Failed to connect the stream")?;

        Ok(())
    }
}