diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/H264AnnexB.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/H264AnnexB.kt new file mode 100644 index 00000000..d12aa63c --- /dev/null +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/H264AnnexB.kt @@ -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> { + val result = mutableListOf>() + 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 + } +} diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamCaptureEngine.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamCaptureEngine.kt new file mode 100644 index 00000000..5fb43b99 --- /dev/null +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamCaptureEngine.kt @@ -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() + } +}