Capture Android AVC and Opus on a shared monotonic timeline

2026-10-12 02:01:09 +02:00
1 parent 2f707366f3
commit c50d5066a9
@@ -0,0 +1,62 @@
+package com.utun.chat.data
+
+import java.io.ByteArrayOutputStream
+import java.nio.ByteBuffer
+
+/** Единый wire-формат AVC: Annex B. Никакого поиска кадра по произвольным байтам после потери блока. */
+object H264AnnexB {
+    const val MAX_FRAME = 2 * 1024 * 1024
+    private val start = byteArrayOf(0, 0, 0, 1)
+
+    fun normalize(bytes: ByteArray): ByteArray {
+        require(bytes.isNotEmpty() && bytes.size <= MAX_FRAME)
+        if (startSize(bytes, 0) != 0) {
+            require(nals(bytes).isNotEmpty())
+            return bytes.copyOf()
+        }
+        val out = ByteArrayOutputStream()
+        val b = ByteBuffer.wrap(bytes)
+        while (b.hasRemaining()) {
+            require(b.remaining() >= 4)
+            val length = b.int
+            require(length > 0 && length <= b.remaining())
+            out.write(start); out.write(bytes, b.position(), length); b.position(b.position() + length)
+        }
+        require(out.size() <= MAX_FRAME)
+        return out.toByteArray()
+    }
+
+    private fun startSize(data: ByteArray, at: Int): Int {
+        if (at + 3 <= data.size && data[at] == 0.toByte() && data[at + 1] == 0.toByte()) {
+            if (data[at + 2] == 1.toByte()) return 3
+            if (at + 4 <= data.size && data[at + 2] == 0.toByte() && data[at + 3] == 1.toByte()) return 4
+        }
+        return 0
+    }
+
+    /** Полные NAL вместе со start code; SPS/PPS можно использовать как MediaCodec csd-0/csd-1. */
+    fun nals(bytes: ByteArray): List<Pair<Int, ByteArray>> {
+        val result = mutableListOf<Pair<Int, ByteArray>>()
+        var at = 0
+        while (at < bytes.size) {
+            val prefix = startSize(bytes, at)
+            require(prefix != 0 && at + prefix < bytes.size)
+            var end = at + prefix + 1
+            while (end < bytes.size && startSize(bytes, end) == 0) end++
+            require((bytes[at + prefix].toInt() and 0x80) == 0)
+            result += (bytes[at + prefix].toInt() and 31) to bytes.copyOfRange(at, end)
+            at = end
+        }
+        return result
+    }
+
+    fun keyFrame(frame: ByteArray, config: ByteArray): ByteArray {
+        val normalized = normalize(frame)
+        val types = nals(normalized).map { it.first }.toSet()
+        require(5 in types)
+        val output = if (7 in types && 8 in types) normalized else config + normalized
+        val all = nals(output).map { it.first }.toSet()
+        require(7 in all && 8 in all && output.size <= MAX_FRAME)
+        return output
+    }
+}
@@ -0,0 +1,256 @@
+package com.utun.chat.data
+
+import android.annotation.SuppressLint
+import android.content.Context
+import android.hardware.camera2.CameraCaptureSession
+import android.hardware.camera2.CameraCharacteristics
+import android.hardware.camera2.CameraDevice
+import android.hardware.camera2.CameraManager
+import android.hardware.camera2.CaptureRequest
+import android.media.AudioFormat
+import android.media.AudioRecord
+import android.media.AudioTimestamp
+import android.media.MediaCodec
+import android.media.MediaCodecInfo
+import android.media.MediaCodecList
+import android.media.MediaFormat
+import android.media.MediaRecorder
+import android.os.Bundle
+import android.os.Handler
+import android.os.HandlerThread
+import android.util.Range
+import android.view.Surface
+import java.io.ByteArrayOutputStream
+
+/** Сервис владеет камерой/кодировщиком; локальный preview — обычный reader собственного стрима.
+ * Камера MediaCodec Surface и AudioTimestamp используют CLOCK_MONOTONIC; PTS относительно общего originNs.
+ * Кодек/камера — один HandlerThread, Opus — отдельный audio thread. UI не владеет ресурсами захвата. */
+class StreamCaptureEngine(
+    private val context: Context, private val id: Long, private val group: Long,
+    private val cameraId: String, private val microphone: Boolean, private val displayRotation: Int,
+    private val failure: (String) -> Unit
+) {
+    private val thread = HandlerThread("stream-camera").apply { start() }
+    private val handler = Handler(thread.looper)
+    private val manager = context.getSystemService(CameraManager::class.java)
+    private var camera: CameraDevice? = null
+    private var session: CameraCaptureSession? = null
+    private var encoder: MediaCodec? = null
+    private var input: Surface? = null
+    private var record: AudioRecord? = null
+    private var audioThread: Thread? = null
+    @Volatile private var audioRunning = false
+    private var stopped = false
+    private var stopping = false
+    private var stopDone: (() -> Unit)? = null
+    private var originNs = 0L
+    private var csd = byteArrayOf()
+    private val partial = ByteArrayOutputStream()
+    private var partialPts = 0L
+    private var partialKey = false
+    private var requireKey = true
+    private var lastKeyRequest = 0L
+    private var frames = 0L
+    private var bytesSent = 0L
+    private var reportNs = 0L
+    private var lastPts = -1L
+
+    fun start() { handler.post { try { configure() } catch (e: Exception) { failed("setup: ${e.message}") } } }
+
+    @SuppressLint("MissingPermission")
+    private fun configure() {
+        if (stopping || stopped) return
+        val characteristics = manager.getCameraCharacteristics(cameraId)
+        val sizes = characteristics.get(CameraCharacteristics.SCALER_STREAM_CONFIGURATION_MAP)
+            ?.getOutputSizes(MediaCodec::class.java)?.toList().orEmpty()
+        val info = MediaCodecList(MediaCodecList.REGULAR_CODECS).codecInfos.firstOrNull { codec ->
+            codec.isEncoder && codec.supportedTypes.any { it.equals(MediaFormat.MIMETYPE_VIDEO_AVC, true) } &&
+                codec.getCapabilitiesForType(MediaFormat.MIMETYPE_VIDEO_AVC).profileLevels.any {
+                    it.profile == MediaCodecInfo.CodecProfileLevel.AVCProfileBaseline
+                } && sizes.any { s -> s.width <= 1280 && s.height <= 720 &&
+                    codec.getCapabilitiesForType(MediaFormat.MIMETYPE_VIDEO_AVC).videoCapabilities.areSizeAndRateSupported(s.width, s.height, 30.0) }
+        } ?: error("No AVC baseline camera/encoder combination at 30 fps")
+        val size = sizes.filter { it.width <= 1280 && it.height <= 720 &&
+            info.getCapabilitiesForType(MediaFormat.MIMETYPE_VIDEO_AVC).videoCapabilities.areSizeAndRateSupported(it.width, it.height, 30.0) }
+            .maxByOrNull { it.width * it.height } ?: error("No compatible camera size")
+        val front = characteristics.get(CameraCharacteristics.LENS_FACING) == CameraCharacteristics.LENS_FACING_FRONT
+        val sensor = characteristics.get(CameraCharacteristics.SENSOR_ORIENTATION) ?: 0
+        val rotation = (sensor + if (front) displayRotation else -displayRotation + 360) % 360
+        val format = MediaFormat.createVideoFormat(MediaFormat.MIMETYPE_VIDEO_AVC, size.width, size.height).apply {
+            setInteger(MediaFormat.KEY_COLOR_FORMAT, MediaCodecInfo.CodecCapabilities.COLOR_FormatSurface)
+            setInteger(MediaFormat.KEY_PROFILE, MediaCodecInfo.CodecProfileLevel.AVCProfileBaseline)
+            setInteger(MediaFormat.KEY_BIT_RATE, if (size.width * size.height >= 1280 * 720) 2_000_000 else 1_000_000)
+            setInteger(MediaFormat.KEY_FRAME_RATE, 30); setInteger(MediaFormat.KEY_I_FRAME_INTERVAL, 1)
+        }
+        check(NativeLib.streamOpen(id, group, width = size.width, height = size.height, rotation = rotation, audio = microphone))
+        originNs = System.nanoTime()
+        encoder = MediaCodec.createByCodecName(info.name).also { codec ->
+            codec.setCallback(object : MediaCodec.Callback() {
+                override fun onInputBufferAvailable(codec: MediaCodec, index: Int) {}
+                override fun onError(codec: MediaCodec, error: MediaCodec.CodecException) { failed("encoder: ${error.diagnosticInfo}") }
+                override fun onOutputFormatChanged(codec: MediaCodec, format: MediaFormat) {
+                    try {
+                        val parts = listOf("csd-0", "csd-1").mapNotNull { name -> format.getByteBuffer(name)?.duplicate()?.let { b ->
+                            ByteArray(b.remaining()).also { b.get(it) }.let(H264AnnexB::normalize)
+                        } }
+                        csd = parts.fold(byteArrayOf()) { a, b -> a + b }
+                        val types = H264AnnexB.nals(csd).map { it.first }.toSet()
+                        check(7 in types && 8 in types) { "AVC format missing SPS/PPS" }
+                        LogManager.addLog("INFO", "video", "encoder format id=$id csd=${csd.size} name=${codec.name}")
+                    } catch (e: Exception) { failed("encoder format: ${e.message}") }
+                }
+                override fun onOutputBufferAvailable(codec: MediaCodec, index: Int, info: MediaCodec.BufferInfo) {
+                    if (stopped) return
+                    try {
+                        if (info.size > 0) {
+                            val buffer = codec.getOutputBuffer(index) ?: error("Missing AVC output")
+                            buffer.position(info.offset); buffer.limit(info.offset + info.size)
+                            val bytes = ByteArray(info.size).also { buffer.get(it) }
+                            if (info.flags and MediaCodec.BUFFER_FLAG_CODEC_CONFIG != 0) {
+                                val config = H264AnnexB.normalize(bytes)
+                                val types = H264AnnexB.nals(config).map { it.first }.toSet()
+                                if (7 in types && 8 in types) csd = config
+                            } else encoded(bytes, info)
+                        }
+                    } catch (e: Exception) { failed("encoded frame: ${e.message}") }
+                    finally { if (!stopped) codec.releaseOutputBuffer(index, false) }
+                    if (info.flags and MediaCodec.BUFFER_FLAG_END_OF_STREAM != 0) finishStop()
+                }
+            }, handler)
+            codec.configure(format, null, null, MediaCodec.CONFIGURE_FLAG_ENCODE)
+            input = codec.createInputSurface(); codec.start()
+        }
+        if (microphone) startAudio()
+        manager.openCamera(cameraId, object : CameraDevice.StateCallback() {
+            override fun onOpened(device: CameraDevice) {
+                if (stopping || stopped) { device.close(); return }
+                camera = device
+                device.createCaptureSession(listOf(input!!), object : CameraCaptureSession.StateCallback() {
+                    override fun onConfigured(configured: CameraCaptureSession) {
+                        if (stopping || stopped) { configured.close(); return }
+                        session = configured
+                        try {
+                            val request = device.createCaptureRequest(CameraDevice.TEMPLATE_RECORD).apply {
+                                addTarget(input!!)
+                                val rates = characteristics.get(CameraCharacteristics.CONTROL_AE_AVAILABLE_TARGET_FPS_RANGES).orEmpty()
+                                rates.filter { it.upper == 30 }.minByOrNull { it.upper - it.lower }?.let {
+                                    set(CaptureRequest.CONTROL_AE_TARGET_FPS_RANGE, Range(it.lower, it.upper))
+                                }
+                            }.build()
+                            configured.setRepeatingRequest(request, null, handler)
+                            LogManager.addLog("INFO", "video", "capture started id=$id camera=$cameraId ${size.width}x${size.height} rotation=$rotation " +
+                                "codec=${info.name} audio=$microphone timebase=MONOTONIC origin=$originNs")
+                        } catch (e: Exception) { failed("camera request: ${e.message}") }
+                    }
+                    override fun onConfigureFailed(session: CameraCaptureSession) { session.close(); failed("camera session configuration") }
+                }, handler)
+            }
+            override fun onDisconnected(device: CameraDevice) { device.close(); failed("camera disconnected") }
+            override fun onError(device: CameraDevice, error: Int) { device.close(); failed("camera error=$error") }
+        }, handler)
+    }
+
+    private fun encoded(bytes: ByteArray, info: MediaCodec.BufferInfo) {
+        if (partial.size() == 0) { partialPts = info.presentationTimeUs; partialKey = false }
+        check(info.presentationTimeUs == partialPts) { "Partial AVC frame PTS changed" }
+        check(partial.size() + bytes.size <= H264AnnexB.MAX_FRAME) { "AVC frame exceeds limit" }
+        partial.write(bytes); partialKey = partialKey || info.flags and MediaCodec.BUFFER_FLAG_KEY_FRAME != 0
+        if (info.flags and MediaCodec.BUFFER_FLAG_PARTIAL_FRAME != 0) return
+        val data = H264AnnexB.normalize(partial.toByteArray()); partial.reset()
+        val key = partialKey && H264AnnexB.nals(data).any { it.first == 5 }
+        if (requireKey && !key) { requestKey(); return }
+        val output = if (key) H264AnnexB.keyFrame(data, csd) else data
+        val pts = partialPts - originNs / 1000
+        check(pts >= 0 && pts > lastPts) { "AVC PTS outside monotonic timeline pts=$pts last=$lastPts" }
+        lastPts = pts
+        when (NativeLib.streamFeed(id, true, pts, key, output)) {
+            1 -> { requireKey = false; frames++; bytesSent += output.size }
+            0 -> { requireKey = true; requestKey() }
+            else -> failed("stream publisher detached")
+        }
+        val now = System.nanoTime()
+        if (now - reportNs >= 1_000_000_000) {
+            reportNs = now
+            LogManager.addLog("DEBUG", "video", "capture id=$id frames=$frames bytes=$bytesSent pts=$pts key=$key waitingKey=$requireKey")
+        }
+    }
+
+    private fun requestKey() {
+        val now = System.nanoTime()
+        if (now - lastKeyRequest < 500_000_000) return
+        lastKeyRequest = now
+        encoder?.setParameters(Bundle().apply { putInt(MediaCodec.PARAMETER_KEY_REQUEST_SYNC_FRAME, 0) })
+        LogManager.addLog("INFO", "video", "requested AVC keyframe id=$id")
+    }
+
+    @SuppressLint("MissingPermission")
+    private fun startAudio() {
+        val min = AudioRecord.getMinBufferSize(48000, AudioFormat.CHANNEL_IN_MONO, AudioFormat.ENCODING_PCM_16BIT)
+        check(min > 0)
+        val rec = AudioRecord(MediaRecorder.AudioSource.MIC, 48000, AudioFormat.CHANNEL_IN_MONO,
+            AudioFormat.ENCODING_PCM_16BIT, (min * 2).coerceAtLeast(960 * 8))
+        record = rec; check(rec.state == AudioRecord.STATE_INITIALIZED)
+        rec.startRecording(); audioRunning = true
+        audioThread = Thread({
+            var opus = 0L
+            try {
+                opus = NativeLib.streamOpusCreate(true); check(opus != 0L)
+                val pcm = ShortArray(960); val stamp = AudioTimestamp()
+                var total = 0L; var waitingStamp = true; var last = -1L
+                while (audioRunning) {
+                    var filled = 0
+                    while (audioRunning && filled < pcm.size) {
+                        val n = rec.read(pcm, filled, pcm.size - filled, AudioRecord.READ_BLOCKING)
+                        check(n > 0 || !audioRunning) { "AudioRecord read=$n" }
+                        if (n > 0) filled += n
+                    }
+                    if (!audioRunning || filled != pcm.size) break
+                    val first = total; total += filled
+                    if (rec.getTimestamp(stamp, AudioTimestamp.TIMEBASE_MONOTONIC) != AudioRecord.SUCCESS) {
+                        if (waitingStamp) LogManager.addLog("DEBUG", "stream", "audio timestamp pending id=$id samples=$total")
+                        continue
+                    }
+                    waitingStamp = false
+                    val pts = (stamp.nanoTime - originNs) / 1000 - (stamp.framePosition - first) * 1_000_000 / 48000
+                    check(pts >= 0 && pts > last) { "Audio timestamp discontinuity pts=$pts last=$last" }; last = pts
+                    val packet = NativeLib.streamOpusEncode(opus, pcm) ?: error("Opus encode failed")
+                    check(NativeLib.streamFeed(id, false, pts, false, packet) >= 0) { "Stream audio publisher detached" }
+                }
+            } catch (e: Exception) { if (audioRunning) handler.post { failed("audio capture: ${e.message}") } }
+            finally { if (opus != 0L) NativeLib.streamOpusClose(opus) }
+        }, "stream-audio-capture").apply { start() }
+    }
+
+    private fun failed(detail: String) {
+        if (stopping || stopped) return
+        LogManager.addLog("ERROR", "stream", "capture failed id=$id $detail")
+        failure(detail)
+    }
+
+    /** Сначала прекращаем producers, затем EOS; completion разрешает native FIN. */
+    fun stop(done: () -> Unit) {
+        handler.post {
+            if (stopped) { done(); return@post }
+            if (stopping) { LogManager.addLog("DEBUG", "stream", "duplicate capture stop id=$id"); return@post }
+            stopping = true; stopDone = done; audioRunning = false
+            try { record?.stop() } catch (e: Exception) { LogManager.addLog("WARN", "stream", "audio stop id=$id: ${e.message}") }
+            audioThread?.join(); audioThread = null
+            record?.release(); record = null
+            session?.close(); session = null; camera?.close(); camera = null
+            try {
+                if (encoder != null) { encoder!!.signalEndOfInputStream(); handler.postDelayed({ finishStop() }, 1000) }
+                else finishStop()
+            } catch (e: Exception) { LogManager.addLog("WARN", "stream", "encoder EOS id=$id: ${e.message}"); finishStop() }
+        }
+    }
+
+    private fun finishStop() {
+        if (stopped) return
+        stopped = true
+        try { encoder?.stop() } catch (e: Exception) { LogManager.addLog("WARN", "video", "encoder stop id=$id: ${e.message}") }
+        try { encoder?.release() } catch (e: Exception) { LogManager.addLog("WARN", "video", "encoder release id=$id: ${e.message}") }
+        encoder = null; input?.release(); input = null; partial.reset()
+        LogManager.addLog("INFO", "stream", "capture released id=$id frames=$frames bytes=$bytesSent")
+        stopDone?.invoke(); stopDone = null; thread.quitSafely()
+    }
+}