From 4a49c830d147dc82c699540d7efa0577b568fa27 Mon Sep 17 00:00:00 2001 From: evgeny Date: Fri, 16 Oct 2026 01:13:23 +0200 Subject: [PATCH] Add bounded Android microphone subscriptions with independent ownership --- .../chat/data/AndroidMicrophoneCapture.kt | 67 ++++++++++++++ .../com/utun/chat/data/MicrophoneCapture.kt | 88 +++++++++++++++++++ .../utun/chat/data/MicrophoneCaptureTest.kt | 69 +++++++++++++++ 3 files changed, 224 insertions(+) create mode 100644 tools/chatgui-android/app/src/main/java/com/utun/chat/data/AndroidMicrophoneCapture.kt create mode 100644 tools/chatgui-android/app/src/main/java/com/utun/chat/data/MicrophoneCapture.kt create mode 100644 tools/chatgui-android/app/src/test/java/com/utun/chat/data/MicrophoneCaptureTest.kt diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/AndroidMicrophoneCapture.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/AndroidMicrophoneCapture.kt new file mode 100644 index 00000000..d103b35e --- /dev/null +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/AndroidMicrophoneCapture.kt @@ -0,0 +1,67 @@ +package com.utun.chat.data + +import android.media.AudioDeviceInfo +import android.media.AudioFormat +import android.media.AudioRecord +import android.media.AudioTimestamp +import android.media.MediaRecorder + +/** Реестр трансляционного захвата. Рация заимствует его только при совпадении выбранного входа. + * Системный default при отсутствии явного выбора означает тот же вход, который уже использует трансляция. */ +object AndroidMicrophoneCapture { + private var stream: MicrophoneCapture? = null + + private class Record(private val record: AudioRecord) : MicrophoneCapture.Device { + override val channels = record.channelCount + override val inputId: Int? get() = record.routedDevice?.id + private val stamp = AudioTimestamp() + private var total = 0L + @Volatile private var running = true + override fun read(): MicrophoneCapture.Frame { + val pcm = ShortArray(960 * channels) + var used = 0 + while (running && used < pcm.size) { + val count = record.read(pcm, used, pcm.size - used, AudioRecord.READ_BLOCKING) + check(count > 0) { "AudioRecord read=$count input=$inputId" } + used += count + } + check(running) { "AudioRecord stopped" } + val first = total; total += 960 + val measured = record.getTimestamp(stamp, AudioTimestamp.TIMEBASE_MONOTONIC) == AudioRecord.SUCCESS + val time = if (measured) stamp.nanoTime - (stamp.framePosition - first) * 1_000_000_000 / 48000 else 0L + return MicrophoneCapture.Frame(pcm, time, first) + } + override fun stop() { running = false; record.stop() } + override fun close() { record.release() } + } + + fun own(record: AudioRecord, released: () -> Unit = {}): MicrophoneCapture { + try { + record.startRecording() + check(record.recordingState == AudioRecord.RECORDSTATE_RECORDING) { "Microphone did not start" } + return MicrophoneCapture(Record(record), { LogManager.addLog("WARN", "radio", it) }, released) + } catch (e: Exception) { record.release(); throw e } + } + + @Synchronized fun openStream(released: () -> Unit): MicrophoneCapture.Subscription { + check(stream == null || stream!!.closed) { "Stream microphone already owned" } + val min = AudioRecord.getMinBufferSize(48000, AudioFormat.CHANNEL_IN_STEREO, AudioFormat.ENCODING_PCM_16BIT) + check(min > 0) { "CAMCORDER stereo format unavailable" } + val record = AudioRecord(MediaRecorder.AudioSource.CAMCORDER, 48000, AudioFormat.CHANNEL_IN_STEREO, + AudioFormat.ENCODING_PCM_16BIT, (min * 2).coerceAtLeast(1920 * 8)) + if (record.state != AudioRecord.STATE_INITIALIZED || record.channelCount != 2) { + record.release(); error("CAMCORDER stereo capture unavailable") + } + val capture = own(record, released) + stream = capture + LogManager.addLog("INFO", "stream", "capture opened source=CAMCORDER channels=2 rate=48000 input=${capture.inputId}") + return capture.subscribe("stream") + } + + @Synchronized fun borrowRadio(input: AudioDeviceInfo?): MicrophoneCapture.Subscription? { + val capture = stream ?: return null + if (capture.closed || (input != null && input.id != capture.inputId)) return null + LogManager.addLog("INFO", "radio", "shared stream capture input=${capture.inputId} selected=${input?.id} channels=${capture.channels}") + return capture.subscribe("radio") + } +} diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/MicrophoneCapture.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/MicrophoneCapture.kt new file mode 100644 index 00000000..db1cd62d --- /dev/null +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/MicrophoneCapture.kt @@ -0,0 +1,88 @@ +package com.utun.chat.data + +import java.util.concurrent.ArrayBlockingQueue +import java.util.concurrent.TimeUnit + +/** Один аппаратный reader, независимые ограниченные очереди потребителей. PCM каждого потребителя — своя копия. + * Закрытие подписки не останавливает соседей; последний владелец останавливает read, делает join и release. + * PTT попадает в ту же очередь, что PCM, поэтому native TX всегда имеет одного последовательного владельца. */ +class MicrophoneCapture(private val device: Device, private val log: (String) -> Unit, private val released: () -> Unit = {}) { + interface Device { + val channels: Int + val inputId: Int? + fun read(): Frame + fun stop() + fun close() + } + sealed interface Event + data class Frame(val pcm: ShortArray, val timeNs: Long, val sequence: Long) : Event + data class Ptt(val pressed: Boolean) : Event + private val consumers = mutableListOf() + @Volatile private var reading = true + @Volatile private var failure: Exception? = null + private var worker: Thread? = null + val closed: Boolean get() = !reading + val inputId: Int? get() = device.inputId + val channels: Int get() = device.channels + + inner class Subscription internal constructor(val name: String) : AutoCloseable { + private val events = ArrayBlockingQueue(32) + @Volatile private var attached = true + private var dropped = 0L + val channels: Int get() = device.channels + + @Synchronized internal fun deliver(frame: Frame) { + if (!attached) return + if (events.size >= 24 || !events.offer(frame.copy(pcm = frame.pcm.copyOf()))) { + dropped++ + if (dropped == 1L || dropped % 50L == 0L) log("capture queue overflow consumer=$name dropped=$dropped") + } + } + /** Только управляющий поток; восемь мест зарезервированы под команды. */ + @Synchronized fun ptt(pressed: Boolean): Boolean { + if (!attached) return false + if (events.offer(Ptt(pressed))) return true + log("capture control queue full consumer=$name pressed=$pressed") + return false + } + fun poll(): Event? { + failure?.let { throw IllegalStateException("Microphone read failed", it) } + return if (attached) events.poll(20, TimeUnit.MILLISECONDS) else null + } + override fun close() { + synchronized(this) { if (!attached) return; attached = false; events.clear() } + detach(this) + } + } + + @Synchronized fun subscribe(name: String): Subscription { + check(reading && failure == null) { "Microphone is closed or failed" } + val subscription = Subscription(name) + consumers.add(subscription) + if (worker == null) worker = Thread({ + try { + while (reading) { + val frame = device.read() + val targets = synchronized(this) { consumers.toList() } + for (target in targets) target.deliver(frame) + } + } catch (e: Exception) { + if (reading) { failure = e; log("capture reader failed: ${e.message}") } + } + }, "microphone-capture").apply { start() } + log("capture attached consumer=$name channels=$channels consumers=${consumers.size}") + return subscription + } + + private fun detach(subscription: Subscription) { + val last = synchronized(this) { + consumers.remove(subscription) + log("capture detached consumer=${subscription.name} remaining=${consumers.size}") + consumers.isEmpty().also { if (it) reading = false } + } + if (!last) return + try { device.stop() } catch (e: Exception) { log("capture stop failed: ${e.message}") } + worker?.join() + try { device.close() } finally { released() } + } +} diff --git a/tools/chatgui-android/app/src/test/java/com/utun/chat/data/MicrophoneCaptureTest.kt b/tools/chatgui-android/app/src/test/java/com/utun/chat/data/MicrophoneCaptureTest.kt new file mode 100644 index 00000000..4b340a28 --- /dev/null +++ b/tools/chatgui-android/app/src/test/java/com/utun/chat/data/MicrophoneCaptureTest.kt @@ -0,0 +1,69 @@ +package com.utun.chat.data + +import org.junit.Assert.* +import org.junit.Test +import java.util.concurrent.LinkedBlockingQueue +import java.util.concurrent.atomic.AtomicInteger + +class MicrophoneCaptureTest { + private class Device : MicrophoneCapture.Device { + override val channels = 2 + override val inputId = 7 + val input = LinkedBlockingQueue() + val reads = AtomicInteger() + var closes = 0 + override fun read(): MicrophoneCapture.Frame = input.take().also { reads.incrementAndGet() } + override fun stop() { input.offer(MicrophoneCapture.Frame(shortArrayOf(), 0, 0)) } + override fun close() { closes++ } + } + private fun frame(subscription: MicrophoneCapture.Subscription): MicrophoneCapture.Frame { + repeat(50) { (subscription.poll() as? MicrophoneCapture.Frame)?.let { return it } } + error("Capture frame timeout") + } + + @Test fun consumersShareCaptureFormatAndTimeButOwnTheirSamples() { + val device = Device(); val capture = MicrophoneCapture(device, {}) + val stream = capture.subscribe("stream"); val radio = capture.subscribe("radio") + try { + device.input.add(MicrophoneCapture.Frame(shortArrayOf(10, 20), 987654321, 960)) + val first = frame(stream); val second = frame(radio) + assertEquals(2, radio.channels); assertEquals(first.timeNs, second.timeNs); assertEquals(first.sequence, second.sequence) + first.pcm[0] = 99 + assertArrayEquals(shortArrayOf(10, 20), second.pcm) + stream.close() + device.input.add(MicrophoneCapture.Frame(shortArrayOf(30, 40), 987674321, 1920)) + assertArrayEquals(shortArrayOf(30, 40), frame(radio).pcm); assertEquals(0, device.closes) + } finally { stream.close(); radio.close() } + assertEquals(1, device.closes); assertTrue(capture.closed) + } + + @Test fun stoppingRadioKeepsStreamAndPttIsOrderedWithPcm() { + val device = Device(); val capture = MicrophoneCapture(device, {}) + val stream = capture.subscribe("stream"); val radio = capture.subscribe("radio") + try { + assertTrue(radio.ptt(true)) + device.input.add(MicrophoneCapture.Frame(shortArrayOf(1, 2), 1000, 0)) + assertEquals(MicrophoneCapture.Ptt(true), radio.poll()) + assertArrayEquals(shortArrayOf(1, 2), frame(radio).pcm) + assertTrue(radio.ptt(false)); assertEquals(MicrophoneCapture.Ptt(false), radio.poll()) + radio.close(); assertEquals(0, device.closes) + assertArrayEquals(shortArrayOf(1, 2), frame(stream).pcm) + } finally { radio.close(); stream.close() } + assertEquals(1, device.closes) + } + + @Test fun slowConsumerHasBoundedQueueAndReservedControlCapacity() { + val device = Device(); val logs = LinkedBlockingQueue(); val capture = MicrophoneCapture(device, { logs.add(it) }) + val stream = capture.subscribe("stream"); val radio = capture.subscribe("radio") + try { + repeat(40) { + device.input.add(MicrophoneCapture.Frame(shortArrayOf(1, 2), it * 20000000L, it * 960L)) + frame(stream) // Трансляция успевает, рация намеренно не читает. + } + assertTrue(logs.any { it.contains("consumer=radio dropped=") }) + assertTrue(radio.ptt(true)); assertTrue(radio.ptt(false)) + repeat(24) { assertTrue(radio.poll() is MicrophoneCapture.Frame) } + assertEquals(MicrophoneCapture.Ptt(true), radio.poll()); assertEquals(MicrophoneCapture.Ptt(false), radio.poll()) + } finally { radio.close(); stream.close() } + } +}