audiofox 1.0.2

// 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::builder::{CHUNK_SIZE_U8, ChunkBuilder};
use crate::common::callback::AudioCallback;
use crate::common::chunk::Chunk;
use crate::common::result::Result;
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) -> 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",
            }).context("Failed to create audio stream")?;

        // Configure this stream's event handlers
        let builder = ChunkBuilder::new_u8();
        let stream_listener = stream
            .add_local_listener_with_user_data(AudioInfoRaw::default())
            .param_changed(Self::on_update(uuid))
            .process(Self::on_buffer(callback, builder))
            .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(uuid: Uuid) -> impl Fn(&Stream, &mut AudioInfoRaw, u32, Option<&Pod>) {
        move |_, user_data, id, param| {
            let Some(param) = param else {
                trace!("stream uuid={uuid}: format reset");
                return;
            };
            if id == ParamType::Format.as_raw() && user_data.parse(param).is_ok() {
                let prefix = if log::log_enabled!(log::Level::Debug) {
                    format!("stream uuid={uuid}: ")
                } else {
                    String::new()
                };
                info!("{}capturing at rate:{}, channels:{}", prefix, user_data.rate(), user_data.channels());
            }
        }
    }

    /// Called when the stream receives a new audio buffer.
    fn on_buffer(callback: AudioCallback, mut builder: ChunkBuilder<u8, CHUNK_SIZE_U8>) -> impl FnMut(&Stream, &mut AudioInfoRaw) {
        move |stream, _user_data| 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;

                // Accumulate samples until we have enough to send a full chunk
                if let Some(sample_slice) = data.data() {
                    // Partition the buffer with the given chunk size
                    let samples = &mut sample_slice[..size];
                    // Bulk-copy the samples over to the ChunkBuilder
                    if let Some(data) = builder.extend_from_slice(samples) {
                        // If data is returned, we're at capacity
                        callback(Chunk::from(&builder));
                        // Reset the ChunkBuilder with the overflow
                        builder.reset_with_overflow(data);
                    }
                }
            }
        }
    }

    /// 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(2);

        // 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(())
    }
}