diff --git a/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/audio/AudioDecoder.kt b/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/audio/AudioDecoder.kt index ab66c20..b202aac 100644 --- a/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/audio/AudioDecoder.kt +++ b/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/audio/AudioDecoder.kt @@ -41,7 +41,7 @@ object AudioDecoder { var srcRate = inFormat.getIntOr(MediaFormat.KEY_SAMPLE_RATE, TARGET_SR) var channels = inFormat.getIntOr(MediaFormat.KEY_CHANNEL_COUNT, 1) - val shorts = ArrayList(TARGET_SR * 30) + val pcm = java.io.ByteArrayOutputStream(TARGET_SR * 60) // 累积 PCM16 字节，不装箱 val info = MediaCodec.BufferInfo() var inEos = false var outEos = false @@ -72,9 +72,10 @@ object AudioDecoder { oi >= 0 -> { if (info.size > 0) { val ob = codec.getOutputBuffer(oi)!! - ob.position(info.offset); ob.limit(info.offset + info.size) - val sb = ob.order(ByteOrder.LITTLE_ENDIAN).asShortBuffer() - while (sb.hasRemaining()) shorts.add(sb.get()) + val bytes = ByteArray(info.size) + ob.position(info.offset) + ob.get(bytes) + pcm.write(bytes) } codec.releaseOutputBuffer(oi, false) if (info.flags and MediaCodec.BUFFER_FLAG_END_OF_STREAM != 0) outEos = true @@ -83,15 +84,18 @@ object AudioDecoder { } codec.stop(); codec.release() + val raw = pcm.toByteArray() + val sbuf = java.nio.ByteBuffer.wrap(raw).order(ByteOrder.LITTLE_ENDIAN).asShortBuffer() + val nShorts = sbuf.remaining() // 下混 stereo → mono var mono = if (channels >= 2) { - FloatArray(shorts.size / channels) { i -> + FloatArray(nShorts / channels) { i -> var s = 0f - for (c in 0 until channels) s += shorts[i * channels + c] / 32768f + for (c in 0 until channels) s += sbuf.get(i * channels + c) / 32768f s / channels } } else { - FloatArray(shorts.size) { shorts[it] / 32768f } + FloatArray(nShorts) { sbuf.get(it) / 32768f } } if (srcRate != TARGET_SR) mono = resample(mono, srcRate, TARGET_SR) mono diff --git a/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/diarize/DiarizationWorker.kt b/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/diarize/DiarizationWorker.kt index aa5df32..7e4a06d 100644 --- a/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/diarize/DiarizationWorker.kt +++ b/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/diarize/DiarizationWorker.kt @@ -8,6 +8,7 @@ import com.lemon.lifelog.asr.DiarizationEngine import com.lemon.lifelog.asr.ModelDownloader import com.lemon.lifelog.asr.SpeakerClusterer import com.lemon.lifelog.asr.SpeakerEmbedder +import com.lemon.lifelog.asr.toLeBytes import com.lemon.lifelog.audio.AudioDecoder import com.lemon.lifelog.data.Repository import com.lemon.lifelog.data.Segment @@ -35,7 +36,7 @@ class DiarizationWorker(ctx: Context, params: WorkerParameters) : CoroutineWorke private val isRunning = AtomicBoolean(false) const val RUN_BUDGET_MS = 7 * 60 * 1000L const val MIN_SESSION_SEGMENTS = 2 // 单段会话无需 diarize - const val MAX_CONCAT_SAMPLES = 18 * 60 * 16000 // ≤18min speech，防内存/超时 + const val MAX_CONCAT_SAMPLES = 10 * 60 * 16000 // ≤10min speech，防 OOM（与 photolog 共进程）/超时 private const val KEY_DONE = "diarized_sessions" // prefs StringSet of windowStartTs } @@ -105,6 +106,12 @@ class DiarizationWorker(ctx: Context, params: WorkerParameters) : CoroutineWorke embedder?.release() } + // 全局重聚类：用 diarization 产出的"干净"per-turn embedding 统一聚类 + //（reclusterAll 本就全局 + 保留人工命名锚点）。只在有新会话被处理时跑，处理完即稳定。 + if (processed > 0) { + val n = SpeakerClusterer.reclusterAll(repo, matchThreshold = threshold) + Log.i(TAG, "global recluster → $n speakers") + } prefs.edit().putStringSet(KEY_DONE, done).apply() Log.i(TAG, "done: ${sessions.size} sessions, $processed diarized") return Result.success() @@ -121,55 +128,52 @@ class DiarizationWorker(ctx: Context, params: WorkerParameters) : CoroutineWorke // 1. 解码 + 拼接，记录每段在拼接波形里的 [offset, offset+len]（采样） data class SegSpan(val seg: Segment, val from: Int, val to: Int) val spans = ArrayList(session.size) - val audio = ArrayList(sr * 60) + val chunks = ArrayList(session.size) // 不装箱：存 FloatArray 块，最后一次性拼 + var total = 0 for (seg in session) { val pcm = AudioDecoder.decodeToFloat(seg.audioPath) if (pcm.isEmpty()) continue - val from = audio.size - for (v in pcm) audio.add(v) - spans.add(SegSpan(seg, from, audio.size)) - if (audio.size >= MAX_CONCAT_SAMPLES) break + val from = total + chunks.add(pcm); total += pcm.size + spans.add(SegSpan(seg, from, total)) + if (total >= MAX_CONCAT_SAMPLES) break } - if (spans.size < MIN_SESSION_SEGMENTS || audio.isEmpty()) return@withContext false - val samples = FloatArray(audio.size) { audio[it] } + if (spans.size < MIN_SESSION_SEGMENTS || total == 0) return@withContext false + val samples = FloatArray(total) + var off = 0 + for (c in chunks) { c.copyInto(samples, off); off += c.size } + chunks.clear() // 2. diarize val turns = engine.diarize(samples) if (turns.isEmpty()) return@withContext false - // 3. 每个局部说话人 → 最长 turn 的 embedding → 桥接全局 - val byLocal = turns.groupBy { it.speaker } - val clusterer = SpeakerClusterer(repo, matchThreshold = threshold).also { it.load() } - val localToGlobal = HashMap>() // local → (globalSpeakerId, sim) - for ((local, ts) in byLocal) { - // 局部说话人总时长 <2s 视为噪声，不桥接（避免一堆短 turn 各建一个全局说话人）。 - val totalSec = ts.sumOf { (it.endSec - it.startSec).toDouble() } - if (totalSec < 2.0) continue - val longest = ts.maxByOrNull { it.endSec - it.startSec } ?: continue - val a = (longest.startSec * sr).toInt().coerceIn(0, samples.size) - val b = (longest.endSec * sr).toInt().coerceIn(a, samples.size) - if (b - a < sr / 2) continue // <0.5s 太短，跳过该局部说话人 - val emb = embedder.embed(samples.copyOfRange(a, b), sr) ?: continue - val durMs = ((longest.endSec - longest.startSec) * 1000).toLong() - val asg = clusterer.assign(emb, durMs) - if (asg.speakerId != null) localToGlobal[local] = asg.speakerId to (asg.sim ?: 1f) - } - if (localToGlobal.isEmpty()) return@withContext false - - // 4. 每段 → 重叠最大的 turn → 全局 speakerId + // 3. 每段 → 重叠最大的 turn → 取该 turn 的"干净"embedding（单说话人）覆盖段 embedding。 + // 不在这里建全局说话人——交给最后的全局 reclusterAll 统一聚类（避免每会话贪心爆炸）。 + val turnEmb = HashMap() // turn 下标 → embedding（缓存，多段共用一个 turn） + var embedded = 0 for (span in spans) { val s0 = span.from.toFloat() / sr val s1 = span.to.toFloat() / sr - var bestTurn: DiarizationEngine.Turn? = null + var bestIdx = -1 var bestOv = 0f - for (t in turns) { + turns.forEachIndexed { i, t -> val ov = minOf(s1, t.endSec) - maxOf(s0, t.startSec) - if (ov > bestOv) { bestOv = ov; bestTurn = t } + if (ov > bestOv) { bestOv = ov; bestIdx = i } + } + if (bestIdx < 0) continue + val emb = turnEmb.getOrPut(bestIdx) { + val t = turns[bestIdx] + val a = (t.startSec * sr).toInt().coerceIn(0, samples.size) + val b = (t.endSec * sr).toInt().coerceIn(a, samples.size) + if (b - a < sr / 4) null else embedder.embed(samples.copyOfRange(a, b), sr) + } + if (emb != null) { + withContext(Dispatchers.IO) { repo.segments.updateEmbedding(span.seg.id, emb.toLeBytes()) } + embedded++ } - val g = bestTurn?.let { localToGlobal[it.speaker] } ?: continue - withContext(Dispatchers.IO) { repo.segments.updateSpeaker(span.seg.id, g.first, g.second) } } - Log.i(TAG, "session ${session.first().startTs}: ${spans.size} segs, ${turns.size} turns, ${byLocal.size} local speakers") - true + Log.i(TAG, "session ${session.first().startTs}: ${spans.size} segs, ${turns.size} turns, $embedded clean-embedded") + embedded > 0 } }