kobo_core/audio/
a2dp.rs

1// SPDX-License-Identifier: MIT
2// Copyright (c) 2026 Nayeem Bin Ahsan
3//! `AospA2dpSink` - drives the Android `audio.a2dp.default` HAL on the Kobo.
4//!
5//! Proven path (Spike A1): open the A2DP control Unix socket, send the AOSP
6//! `audio_a2dp_hw` commands (`CHECK_READY`/`START`/`STOP`), then write S16LE
7//! PCM to the data socket. `btservice` (separate from nickel) owns the sockets
8//! and serves the paired BT headset - no nickel needed at runtime (Spike A2).
9//!
10//! Gap #5 (backpressure/pacing): the data socket is written with `write_all`,
11//! which **blocks when the socket buffer is full**. Since the BT sink drains at
12//! real time (44.1 kHz), this yields natural sample-clock pacing. Continuous
13//! speech that doesn't underrun/garble is the C-play gate.
14use std::time::Duration;
15use thiserror::Error;
16use tokio::io::{AsyncReadExt, AsyncWriteExt};
17use tokio::net::UnixStream;
18
19pub const CTRL_PATH: &str = "/tmp/audio.a2dp_ctrl";
20pub const DATA_PATH: &str = "/tmp/audio.a2dp_data";
21
22// AOSP audio_a2dp_hw command opcodes (audio_a2dp_hw.h)
23const CMD_CHECK_READY: u8 = 0x01;
24const CMD_START: u8 = 0x02;
25const CMD_STOP: u8 = 0x03;
26// ack values
27const ACK_SUCCESS: u8 = 0x00;
28// Max wait for btservice to respond (ctrl-socket ack) or drain (data-socket
29// write) before treating the BT stack as wedged (gap #5 / review #2).
30const BTSERVICE_TIMEOUT_SECS: u64 = 3;
31
32#[derive(Debug, Error)]
33pub enum A2dpError {
34    #[error("connect {path}: {source}")]
35    Connect {
36        path: String,
37        source: std::io::Error,
38    },
39    #[error("ctrl command {cmd:#04x} ack {ack:#04x} (expected {expected:#04x})")]
40    BadAck { cmd: u8, ack: u8, expected: u8 },
41    #[error("io: {0}")]
42    Io(#[from] std::io::Error),
43}
44
45/// The sample format the A1 proof validated on the Havit headset.
46/// (Edge-TTS PCM is 24 kHz mono -> resampled to this by [`crate::pipeline`].)
47pub const TARGET_RATE: usize = 44_100;
48pub const TARGET_CHANNELS: usize = 2;
49
50pub struct AospA2dpSink {
51    ctrl: UnixStream,
52    data: Option<UnixStream>,
53    started: bool,
54}
55
56impl AospA2dpSink {
57    /// Open the control socket and verify the sink is ready.
58    /// (The data socket is NOT connected here - it only exists after `start()`.)
59    pub async fn open() -> Result<Self, A2dpError> {
60        let ctrl = UnixStream::connect(CTRL_PATH)
61            .await
62            .map_err(|e| A2dpError::Connect {
63                path: CTRL_PATH.into(),
64                source: e,
65            })?;
66        let mut s = AospA2dpSink {
67            ctrl,
68            data: None,
69            started: false,
70        };
71        s.ctrl_cmd(CMD_CHECK_READY, "CHECK_READY").await?;
72        Ok(s)
73    }
74
75    async fn ctrl_cmd(&mut self, cmd: u8, _name: &str) -> Result<u8, A2dpError> {
76        self.ctrl.write_all(&[cmd]).await?;
77        let mut ack = [0u8; 1];
78        // Distinguish timeout from a real read error (review #2): previously the
79        // inner io::Result was dropped, so a read error left ack=[0x00] = false success.
80        match tokio::time::timeout(
81            Duration::from_secs(BTSERVICE_TIMEOUT_SECS),
82            self.ctrl.read_exact(&mut ack),
83        )
84        .await
85        {
86            Ok(Ok(_)) => {}
87            Ok(Err(e)) => return Err(A2dpError::Io(e)),
88            Err(_) => {
89                return Err(A2dpError::Io(std::io::Error::new(
90                    std::io::ErrorKind::TimedOut,
91                    "ctrl ack timeout",
92                )))
93            }
94        }
95        if (cmd == CMD_CHECK_READY || cmd == CMD_START) && ack[0] != ACK_SUCCESS {
96            return Err(A2dpError::BadAck {
97                cmd,
98                ack: ack[0],
99                expected: ACK_SUCCESS,
100            });
101        }
102        Ok(ack[0])
103    }
104
105    /// START the stream and open the data socket (created by the HAL on START).
106    pub async fn start(&mut self) -> Result<(), A2dpError> {
107        self.ctrl_cmd(CMD_START, "START").await?;
108        self.data = Some(
109            UnixStream::connect(DATA_PATH)
110                .await
111                .map_err(|e| A2dpError::Connect {
112                    path: DATA_PATH.into(),
113                    source: e,
114                })?,
115        );
116        self.started = true;
117        Ok(())
118    }
119
120    /// Write PCM bytes (S16LE interleaved) to the A2DP data socket.
121    /// Blocks on a full buffer -> real-time pacing (gap #5).
122    pub async fn write_pcm(&mut self, pcm: &[u8]) -> Result<(), A2dpError> {
123        match self.data.as_mut() {
124            Some(d) => tokio::time::timeout(
125                Duration::from_secs(BTSERVICE_TIMEOUT_SECS),
126                d.write_all(pcm),
127            )
128            .await
129            .map_err(|_| {
130                A2dpError::Io(std::io::Error::new(
131                    std::io::ErrorKind::TimedOut,
132                    "a2dp data write timeout (btservice not draining)",
133                ))
134            })?
135            .map_err(A2dpError::Io),
136            None => Err(A2dpError::Io(std::io::Error::new(
137                std::io::ErrorKind::NotConnected,
138                "a2dp data socket not open (call start() first)",
139            ))),
140        }
141    }
142
143    /// Query the kernel socket send buffer for unsent bytes (TIOCOUTQ).
144    /// This gives the TRUE buffered amount - no wall-clock drift.
145    /// Returns bytes not yet consumed by the A2DP HAL.
146    pub fn unsent_bytes(&self) -> usize {
147        use std::os::unix::io::AsRawFd;
148        if let Some(ref data) = self.data {
149            let fd = data.as_raw_fd();
150            let mut value: std::ffi::c_int = 0;
151            // SAFETY: fd is a live socket descriptor (data owns it for this
152            // call); TIOCOUTQ writes one int into &mut value (C-compatible),
153            // no aliasing. A failing ioctl returns <0 and value stays 0.
154            unsafe {
155                libc::ioctl(fd, libc::TIOCOUTQ, &mut value);
156            }
157            value.max(0) as usize
158        } else {
159            0
160        }
161    }
162
163    /// STOP the stream.
164    pub async fn stop(&mut self) -> Result<(), A2dpError> {
165        if self.started {
166            // best-effort: STOP on an already-closing stream; Drop will clean up
167            log::warn!("a2dp: best-effort STOP (error ignored if stream already closing)");
168            let _ = self.ctrl_cmd(CMD_STOP, "STOP").await;
169            self.started = false;
170        }
171        Ok(())
172    }
173}
174
175impl Drop for AospA2dpSink {
176    fn drop(&mut self) {
177        // best-effort synchronous stop (the async stop() should be awaited first)
178        use std::os::unix::net::UnixStream as StdStream;
179        if self.started {
180            if let Ok(mut c) = StdStream::connect(CTRL_PATH) {
181                use std::io::Write;
182                // best-effort: Drop runs during teardown; a write failure here
183                // means the control socket is already gone (btservice stopped).
184                if let Err(e) = c.write_all(&[CMD_STOP]) {
185                    log::warn!("a2dp drop STOP write failed: {e}");
186                }
187            }
188        }
189    }
190}
kobo_core/audio/a2dp.rs

kobo_core/audio/
a2dp.rs
