kobo_core/audio/
a2dp.rs

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

kobo_core/audio/
a2dp.rs
