Keep shared microphone PCM continuous across device changes

tools/chatgui-android/app/src/main/java/com/utun/chat/data/MediaCaptureTimeline.kt

@@ -0,0 +1,65 @@
+package com.utun.chat.data
+
+import java.util.ArrayDeque
+import kotlin.math.roundToInt
+
+/** Общая сетка PCM 48k stereo. Аппаратные timestamps определяют положение отсчётов;
+ * отсутствующие участки — тишина. Никакого переноса старого звука на новое время.
+ * Все методы вызываются под lock владельца. */
+internal class MediaCaptureTimeline(val originNs: Long) {
+    companion object {
+        const val FRAMES = 960
+        const val PERIOD_NS = 20_000_000L
+        const val RESERVE_NS = 200_000_000L
+    }
+    private val pending = ArrayDeque<MicrophoneCapture.Frame>()
+    private var position = 0L
+    private var lastInputNs = -1L
+    var generation = 0L
+        private set
+    var lateFrames = 0L
+        private set
+    var silentFrames = 0L
+        private set
+    val nextTimeNs: Long get() = originNs + position * 1_000_000_000 / 48000
+    val deadlineNs: Long get() = nextTimeNs + PERIOD_NS + RESERVE_NS
+
+    fun reset(epoch: Long) {
+        generation = epoch; pending.clear(); lastInputNs = -1
+    }
+
+    fun offer(frame: MicrophoneCapture.Frame): Boolean {
+        require(frame.pcm.size == FRAMES * 2)
+        if (frame.generation != generation || frame.timeNs <= lastInputNs) return false
+        lastInputNs = frame.timeNs
+        if (frame.timeNs + PERIOD_NS <= nextTimeNs) { lateFrames++; return false }
+        if (pending.size >= 24) { lateFrames++; return false }
+        pending.addLast(frame)
+        return true
+    }
+
+    fun pull(nowNs: Long): MicrophoneCapture.Frame {
+        check(nowNs >= deadlineNs)
+        val start = nextTimeNs
+        val pcm = ShortArray(FRAMES * 2)
+        var filled = 0
+        while (pending.isNotEmpty()) {
+            val frame = pending.first
+            val offset = ((frame.timeNs - start).toDouble() * 48000 / 1_000_000_000).roundToInt()
+            if (offset >= FRAMES) break
+            val source = maxOf(0, -offset)
+            val target = maxOf(0, offset, filled)
+            val skip = target - maxOf(0, offset)
+            val count = minOf(FRAMES - source - skip, FRAMES - target)
+            if (count > 0) {
+                frame.pcm.copyInto(pcm, target * 2, (source + skip) * 2, (source + skip + count) * 2)
+                silentFrames += target - filled
+                filled = target + count
+            }
+            if (offset + FRAMES > FRAMES) break
+            pending.removeFirst()
+        }
+        silentFrames += FRAMES - filled
+        return MicrophoneCapture.Frame(pcm, start, position, nowNs, generation).also { position += FRAMES }
+    }
+}

tools/chatgui-android/app/src/main/java/com/utun/chat/data/MicrophoneCapture.kt

@@ -6,7 +6,8 @@ import java.util.concurrent.TimeUnit
 /** Один аппаратный reader, независимые ограниченные очереди потребителей. PCM каждого потребителя — своя копия.
  * Закрытие подписки не останавливает соседей; последний владелец останавливает read, делает join и release.
  * PTT попадает в ту же очередь, что PCM, поэтому native TX всегда имеет одного последовательного владельца. */
-class MicrophoneCapture(private val device: Device, private val log: (String, String) -> Unit, private val released: () -> Unit = {}) {
+class MicrophoneCapture(private val device: Device, private val log: (String, String) -> Unit, private val released: () -> Unit = {},
+                        private val changed: () -> Unit = {}) {
     interface Device {
         val channels: Int
         val inputId: Int?
@@ -15,14 +16,15 @@ class MicrophoneCapture(private val device: Device, private val log: (String, St
         fun close()
     }
     sealed interface Event
-    data class Frame(val pcm: ShortArray, val timeNs: Long, val sequence: Long, val createdNs: Long = System.nanoTime()) : Event
+    data class Frame(val pcm: ShortArray, val timeNs: Long, val sequence: Long, val createdNs: Long = System.nanoTime(),
+                     val generation: Long = 0) : Event
     data class Ptt(val pressed: Boolean) : Event
     private val consumers = mutableListOf<Subscription>()
     @Volatile private var reading = true
-    @Volatile private var disposed = false
     @Volatile private var failure: Exception? = null
     private var worker: Thread? = null
-    val closed: Boolean get() = disposed
+    val closed: Boolean get() = !reading
+    val names: Set<String> get() = synchronized(this) { consumers.map { it.name }.toSet() }
     val inputId: Int? get() = device.inputId
     val channels: Int get() = device.channels
 
@@ -72,6 +74,7 @@ class MicrophoneCapture(private val device: Device, private val log: (String, St
             }
         }, "microphone-capture").apply { start() }
         log("INFO", "capture attached consumer=$name channels=$channels consumers=${consumers.size}")
+        changed()
         return subscription
     }
 
@@ -81,10 +84,11 @@ class MicrophoneCapture(private val device: Device, private val log: (String, St
             log("INFO", "capture detached consumer=${subscription.name} remaining=${consumers.size}")
             consumers.isEmpty().also { if (it) reading = false }
         }
+        changed()
         if (!last) return
         try { device.stop() } catch (e: Exception) { log("WARN", "capture stop failed: ${e.message}") }
         worker?.join()
         try { device.close() } catch (e: Exception) { log("ERROR", "capture release failed: ${e.message}") }
-        finally { disposed = true; released() }
+        finally { released() }
     }
 }

tools/chatgui-android/app/src/test/java/com/utun/chat/data/MediaCaptureTimelineTest.kt

@@ -0,0 +1,71 @@
+package com.utun.chat.data
+
+import org.junit.Assert.*
+import org.junit.Test
+
+class MediaCaptureTimelineTest {
+    private val origin = 1_000_000_000L
+    private fun frame(time: Long, value: Int, epoch: Long = 1) =
+        MicrophoneCapture.Frame(ShortArray(1920) { value.toShort() }, time, 0, generation = epoch)
+    private fun pull(timeline: MediaCaptureTimeline) = timeline.pull(timeline.deadlineNs)
+
+    @Test fun switchingKeepsTimeAndSequenceAndRejectsOldDevice() {
+        val timeline = MediaCaptureTimeline(origin)
+        timeline.reset(1)
+        assertTrue(timeline.offer(frame(origin, 11)))
+        val first = pull(timeline)
+        timeline.reset(2)
+        assertFalse(timeline.offer(frame(origin + 20_000_000, 99)))
+        val silence = pull(timeline)
+        assertTrue(silence.pcm.all { it == 0.toShort() })
+        assertTrue(timeline.offer(frame(origin + 40_000_000, 22, 2)))
+        val next = pull(timeline)
+        assertEquals(first.timeNs + 40_000_000, next.timeNs)
+        assertEquals(1920L, next.sequence)
+        assertTrue(next.pcm.all { it == 22.toShort() })
+    }
+
+    @Test fun differentlyAlignedMicrophoneFramesFillTheirActualTime() {
+        val timeline = MediaCaptureTimeline(origin)
+        timeline.reset(1)
+        assertTrue(timeline.offer(frame(origin + 10_000_000, 1)))
+        assertTrue(timeline.offer(frame(origin + 30_000_000, 2)))
+        val first = pull(timeline).pcm
+        val second = pull(timeline).pcm
+        assertTrue(first.take(960).all { it == 0.toShort() })
+        assertTrue(first.drop(960).all { it == 1.toShort() })
+        assertTrue(second.take(960).all { it == 1.toShort() })
+        assertTrue(second.drop(960).all { it == 2.toShort() })
+    }
+
+    @Test fun lateSoundIsNeverRetimestampedOverPreviouslySentSilence() {
+        val timeline = MediaCaptureTimeline(origin)
+        timeline.reset(1)
+        repeat(20) { pull(timeline) }
+        assertFalse(timeline.offer(frame(origin, 123)))
+        assertEquals(1L, timeline.lateFrames)
+        val result = pull(timeline)
+        assertEquals(origin + 400_000_000, result.timeNs)
+        assertTrue(result.pcm.all { it == 0.toShort() })
+    }
+
+    @Test fun deviceOpeningDelayProducesContinuousTwentyMillisecondPackets() {
+        val timeline = MediaCaptureTimeline(origin)
+        timeline.reset(1)
+        repeat(1500) { index ->
+            if (index % 50 == 0) timeline.reset(index.toLong() + 1)
+            val result = pull(timeline)
+            assertEquals(origin + index * 20_000_000L, result.timeNs)
+            assertEquals(index * 960L, result.sequence)
+            assertEquals(1920, result.pcm.size)
+        }
+    }
+
+    @Test fun stalePacketsCannotGrowTheCaptureQueueWithoutBound() {
+        val timeline = MediaCaptureTimeline(origin)
+        timeline.reset(1)
+        repeat(24) { assertTrue(timeline.offer(frame(origin + it * 20_000_000L, 1))) }
+        assertFalse(timeline.offer(frame(origin + 480_000_000L, 2)))
+        assertEquals(1L, timeline.lateFrames)
+    }
+}