From 3512a84f758e79ab750db56c506e979f3820598d Mon Sep 17 00:00:00 2001 From: yibo Date: Sun, 7 Jun 2026 13:50:15 +0800 Subject: [PATCH] feat(lifelog): 录音改为 block-based diarization-first 管线（攒段→整块分轮转写） --- lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/LifelogApp.kt | 6 +++--- lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/MainActivity.kt | 3 +-- lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/asr/SpeakerClusterer.kt | 7 ++++++- lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/diarize/DiarizationScheduler.kt | 37 ------------------------------------- lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/diarize/DiarizationWorker.kt | 206 -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/service/BlockBuffer.kt | 145 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/service/BlockProcessor.kt | 105 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/service/RecorderService.kt | 137 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++-------------------------------------------------------------------------- 8 files changed, 323 insertions(+), 323 deletions(-) delete mode 100644 lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/diarize/DiarizationScheduler.kt delete mode 100644 lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/diarize/DiarizationWorker.kt create mode 100644 lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/service/BlockBuffer.kt create mode 100644 lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/service/BlockProcessor.kt diff --git a/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/LifelogApp.kt b/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/LifelogApp.kt index db44ede..b322759 100644 --- a/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/LifelogApp.kt +++ b/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/LifelogApp.kt @@ -3,7 +3,6 @@ package com.lemon.lifelog import android.app.Application import android.app.NotificationChannel import android.app.NotificationManager -import com.lemon.lifelog.diarize.DiarizationScheduler import com.lemon.lifelog.service.RecorderService import com.lemon.lifelog.summary.SummaryScheduler @@ -18,9 +17,10 @@ class LifelogApp : Application() { NotificationManager.IMPORTANCE_LOW, ) ) - // 纪要 + diarization periodic worker（各自 15min，UPDATE）。 + // 纪要 periodic worker（15min，UPDATE）。 + // 注：旧的周期 DiarizationWorker 已废弃——diarization-first 管线在「块」关闭时 + // 当场分轮+转写，不再需要事后整体重跑。声纹纠偏走「重新聚类」按钮（reclusterAll）。 SummaryScheduler.schedulePeriodic(this) - DiarizationScheduler.schedulePeriodic(this) // 相册：未处理照片的周期后台处理（人脸+VL），空闲时跑，不阻塞导入。 com.lemon.lifelog.vision.PhotoScheduler.schedulePeriodic(this) } diff --git a/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/MainActivity.kt b/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/MainActivity.kt index 21a3b4c..bd49322 100644 --- a/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/MainActivity.kt +++ b/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/MainActivity.kt @@ -83,7 +83,6 @@ import com.lemon.lifelog.asr.XfyunIatClient import com.lemon.lifelog.asr.transcribeWithStats import com.lemon.lifelog.audio.SegmentPlayer import com.lemon.lifelog.chat.ChatConfig -import com.lemon.lifelog.diarize.DiarizationScheduler import com.lemon.lifelog.summary.SummaryProgress import com.lemon.lifelog.summary.SummaryRunLog import com.lemon.lifelog.data.Repository @@ -905,7 +904,7 @@ private fun SummariesTab(repo: Repository, activity: Context, onJumpToTime: (Lon Column(modifier = Modifier.fillMaxSize().padding(16.dp)) { Row(verticalAlignment = Alignment.CenterVertically, horizontalArrangement = Arrangement.spacedBy(8.dp)) { Text("纪要（${summaries.size}）", style = MaterialTheme.typography.titleMedium) - Button(onClick = { DiarizationScheduler.runNow(activity); SummaryScheduler.runNow(activity) }, enabled = !running) { + Button(onClick = { SummaryScheduler.runNow(activity) }, enabled = !running) { if (running) { CircularProgressIndicator(modifier = Modifier.size(16.dp), strokeWidth = 2.dp) Spacer(Modifier.width(6.dp)) diff --git a/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/asr/SpeakerClusterer.kt b/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/asr/SpeakerClusterer.kt index 728ec92..48c3abc 100644 --- a/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/asr/SpeakerClusterer.kt +++ b/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/asr/SpeakerClusterer.kt @@ -59,7 +59,12 @@ class SpeakerClusterer( } if (best != null && bestSim >= matchThreshold) { - // capped running mean：上限保证质心保持自适应、不被早期样本钉死。 + if (autoIndexOf(best.name) == null) { + // 人工命名 anchor：只归属、**冻结质心不漂移**（同 reclusterAll 的 L1）。 + // 否则 anchor 吸入相邻说话人后会朝其漂移、再吞更多 → 雪球式串人。 + return@withLock Assignment(best.id, bestSim, best.name) + } + // 自动簇：capped running mean，质心保持自适应、不被早期样本钉死。 val n = minOf(best.count, maxSamplesPerCentroid) val merged = FloatArray(emb.size) { i -> (best.centroid[i] * n + emb[i]) / (n + 1) } l2normalize(merged) diff --git a/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/diarize/DiarizationScheduler.kt b/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/diarize/DiarizationScheduler.kt deleted file mode 100644 index 38ce850..0000000 --- a/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/diarize/DiarizationScheduler.kt +++ /dev/null @@ -1,37 +0,0 @@ -package com.lemon.lifelog.diarize - -import android.content.Context -import androidx.work.Constraints -import androidx.work.ExistingPeriodicWorkPolicy -import androidx.work.ExistingWorkPolicy -import androidx.work.NetworkType -import androidx.work.OneTimeWorkRequestBuilder -import androidx.work.PeriodicWorkRequestBuilder -import androidx.work.WorkManager -import java.util.concurrent.TimeUnit - -object DiarizationScheduler { - private const val PERIODIC_TAG = "lifelog-diarize-periodic" - private const val ONE_SHOT_TAG = "lifelog-diarize-once" - - fun schedulePeriodic(ctx: Context) { - val constraints = Constraints.Builder() - .setRequiredNetworkType(NetworkType.NOT_REQUIRED) - .setRequiresBatteryNotLow(true) - .build() - val request = PeriodicWorkRequestBuilder(15, TimeUnit.MINUTES) - .setConstraints(constraints) - .build() - WorkManager.getInstance(ctx).enqueueUniquePeriodicWork( - PERIODIC_TAG, - ExistingPeriodicWorkPolicy.UPDATE, - request, - ) - } - - /** 手动跑一次（KEEP：已在跑/排队则忽略）。 */ - fun runNow(ctx: Context) { - val request = OneTimeWorkRequestBuilder().addTag(ONE_SHOT_TAG).build() - WorkManager.getInstance(ctx).enqueueUniqueWork(ONE_SHOT_TAG, ExistingWorkPolicy.KEEP, request) - } -} 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 deleted file mode 100644 index dd97cce..0000000 --- a/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/diarize/DiarizationWorker.kt +++ /dev/null @@ -1,206 +0,0 @@ -package com.lemon.lifelog.diarize - -import android.content.Context -import android.util.Log -import androidx.work.CoroutineWorker -import androidx.work.WorkerParameters -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 -import com.lemon.lifelog.summary.SummaryWorker -import java.io.File -import java.util.concurrent.atomic.AtomicBoolean -import kotlinx.coroutines.Dispatchers -import kotlinx.coroutines.withContext - -/** - * 会话级 diarization 重跑（方案 A）。对每个**已结束、未 diarize**的会话： - * 1. 解码 + 按序拼接该会话所有段音频（speech-only 16k mono）。 - * 2. pyannote 分割 + CAMPPlus + FastClustering → 准确的 [start,end]→局部说话人 turns - * （能把「几秒多人」的段拆开）。 - * 3. 每个局部说话人 → 取其最长 turn 的 embedding → 经 SpeakerClusterer 桥接到 - * 全局 SpeakerProfile（跨会话/跨天「说话人N」一致）。 - * 4. 每个 VAD 段 → 找重叠最大的 turn → 重标 speakerId（段内多说话人先归主导者，v1）。 - * - * 与逐段实时聚类相比准确得多；实时标签只作即时占位，这里批处理纠正。 - */ -class DiarizationWorker(ctx: Context, params: WorkerParameters) : CoroutineWorker(ctx, params) { - - companion object { - const val TAG = "DiarizationWorker" - private val isRunning = AtomicBoolean(false) - // WorkManager 给每个 worker 硬上限 10min，超时直接 kill。预算压到 5min， - // 留足余量让最后一块 diarize 收尾 + 全局 recluster + 写 cursor，绝不撞红线。 - const val RUN_BUDGET_MS = 5 * 60 * 1000L - const val MIN_SESSION_SEGMENTS = 2 // 单段会话无需 diarize - // 单块 ≤3min speech：一次 diarize ~3min，远小于 10min 红线，一轮能跑完多块、 - // 多次写 cursor 推进；块内说话人 index 不跨块——身份交全局 reclusterAll。 - const val CHUNK_SAMPLES = 3 * 60 * 16000 - private const val KEY_DONE = "diarized_sessions" // prefs StringSet of windowStartTs - private const val KEY_CURSOR = "diar_cursor_" // + key → 已处理段数（断点续跑长会议） - } - - private val sr = 16000 - - override suspend fun doWork(): Result { - if (!isRunning.compareAndSet(false, true)) { - Log.i(TAG, "another diarization in progress, skip"); return Result.success() - } - return try { - run() - } catch (t: Throwable) { - Log.w(TAG, "diarization run failed", t); Result.success() - } finally { - isRunning.set(false) - } - } - - private suspend fun run(): Result { - val runStart = System.currentTimeMillis() - val now = System.currentTimeMillis() - val prefs = applicationContext.getSharedPreferences(SummaryWorker.PREFS, Context.MODE_PRIVATE) - val gapMs = prefs.getInt(SummaryWorker.KEY_GAP_MIN, SummaryWorker.DEFAULT_GAP_MIN) * 60_000L - val threshold = SpeakerClusterer.thresholdFromPrefs(applicationContext) - - val repo = Repository(applicationContext) - val segments = repo.segments.page(limit = 5000, offset = 0).sortedBy { it.startTs } - if (segments.size < MIN_SESSION_SEGMENTS) return Result.success() - - val sessions = SummaryWorker.groupSessions(segments, gapMs) - val done = prefs.getStringSet(KEY_DONE, emptySet())!!.toMutableSet() - - // 模型路径 - val downloader = ModelDownloader(applicationContext) - val speakerFile = downloader.ensureSpeakerModel { } - if (!speakerFile.present) { Log.w(TAG, "speaker model missing"); return Result.success() } - val pyannote = downloader.ensurePyannoteModel { Log.i(TAG, "pyannote: $it") } - if (!pyannote.present) { Log.w(TAG, "pyannote seg model missing/download failed"); return Result.success() } - - var engine: DiarizationEngine? = null - var embedder: SpeakerEmbedder? = null - var processed = 0 - var attempted = false - try { - // 最新会话优先（用户更关心最近的；旧会话后续轮次补）。 - for (session in sessions.reversed()) { - val start = session.first().startTs - val end = session.last().endTs - val key = start.toString() - - if (now - end <= gapMs) continue // 未结束 - if (session.size < MIN_SESSION_SEGMENTS) continue - if (done.contains(key)) continue // 已 diarize - if (System.currentTimeMillis() - runStart > RUN_BUDGET_MS) { - Log.i(TAG, "time budget hit, rest next round"); break - } - - // 懒加载模型（有活才加载） - if (engine == null) { - engine = DiarizationEngine.load(pyannote.path, speakerFile.path, numThreads = 4, threshold = threshold) - embedder = SpeakerEmbedder.load(speakerFile.path, numThreads = 2) - } - - attempted = true - val startIdx = prefs.getInt(KEY_CURSOR + key, 0) - val reached = diarizeSession(repo, session, engine!!, embedder!!, runStart, startIdx) - if (reached >= session.size) { - done.add(key); processed++ - prefs.edit().remove(KEY_CURSOR + key).apply() - } else { - // 预算中途用完：记录断点，下轮从此处续跑（不重头），然后退出本轮。 - prefs.edit().putInt(KEY_CURSOR + key, reached).apply() - Log.i(TAG, "session $key partial: cursor=$reached/${session.size}, resume next round"); break - } - } - } finally { - engine?.release() - embedder?.release() - } - - // 全局重聚类：用 diarization 产出的"干净"per-turn embedding 统一聚类 - //（reclusterAll 本就全局 + 保留人工命名锚点）。只要这轮碰过会话就跑。 - if (attempted) { - 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() - } - - /** - * 分块 diarize 整个会话（长会议不再只取前 N min），从 [startIdx] 段断点续跑。 - * 返回**已处理到的段下标**（== session.size 表示整场跑完；< 表示预算中途用完， - * 外层据此存断点、下轮从这里续跑——块内说话人 index 不跨块，身份交全局 reclusterAll）。 - */ - private suspend fun diarizeSession( - repo: Repository, - session: List, - engine: DiarizationEngine, - embedder: SpeakerEmbedder, - runStart: Long, - startIdx: Int, - ): Int = withContext(Dispatchers.Default) { - var idx = startIdx.coerceIn(0, session.size) - var nChunks = 0 - var totalTurns = 0 - var embedded = 0 - while (idx < session.size) { - if (System.currentTimeMillis() - runStart > RUN_BUDGET_MS) { - Log.i(TAG, "session ${session.first().startTs}: budget hit at seg $idx/${session.size}") - return@withContext idx // 断点，外层续跑 - } - // 攒一块（≤CHUNK_SAMPLES），解码即拼，记录每段在块内的 [from,to] - val segs = ArrayList() - val froms = ArrayList() - val tos = ArrayList() - val pcms = ArrayList() - var len = 0 - while (idx < session.size && len < CHUNK_SAMPLES) { - val seg = session[idx]; idx++ - val pcm = AudioDecoder.decodeToFloat(seg.audioPath) - if (pcm.isEmpty()) continue - segs.add(seg); froms.add(len); pcms.add(pcm); len += pcm.size; tos.add(len) - } - if (segs.isEmpty()) continue - val samples = FloatArray(len) - var off = 0 - for (p in pcms) { p.copyInto(samples, off); off += p.size } - pcms.clear() - - val turns = engine.diarize(samples) - if (turns.isEmpty()) continue - nChunks++; totalTurns += turns.size - - val turnEmb = HashMap() // turn 下标 → embedding（多段共用同一 turn 缓存） - for (k in segs.indices) { - val s0 = froms[k].toFloat() / sr - val s1 = tos[k].toFloat() / sr - var bestIdx = -1 - var bestOv = 0f - turns.forEachIndexed { i, t -> - val ov = minOf(s1, t.endSec) - maxOf(s0, t.startSec) - 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(segs[k].id, emb.toLeBytes()) } - embedded++ - } - } - } - Log.i(TAG, "session ${session.first().startTs}: $startIdx→${session.size} segs, $nChunks chunks, $totalTurns turns, $embedded clean-embedded") - idx // == session.size，整场跑完 - } -} diff --git a/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/service/BlockBuffer.kt b/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/service/BlockBuffer.kt new file mode 100644 index 0000000..5105aab --- /dev/null +++ b/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/service/BlockBuffer.kt @@ -0,0 +1,145 @@ +package com.lemon.lifelog.service + +import android.util.Log +import java.io.File +import java.nio.ByteBuffer +import java.nio.ByteOrder +import org.json.JSONArray +import org.json.JSONObject + +/** + * 当前录音「块」的累积器（diarization-first 管线的录音端）。 + * + * VAD 每产出一段语音就 [append] 进块的 `audio.pcm`（raw PCM16 LE，16k mono），并把该段的 + * `{wallTs, sampleOffset, sampleLen}` 记进 manifest（用于之后把 turn 的样本偏移还原成真实时间）。 + * + * 关块条件（[shouldClose]，由 RecorderService 每帧/每段检查）： + * - 距上次语音静音 ≥ [SILENCE_MS]（5min） 或 + * - 块时长 ≥ [MAX_MS]（10min，长会议保活，避免全程不出文字） + * + * 块目录：`filesDir/blocks//{audio.pcm, manifest.json}`。 + * [close] 落 manifest(closed=true) 后交给 BlockProcessor；崩溃时块目录仍在，重启 [listPending] 重扫。 + * 只存 speech（静音不落盘）→ 内存/磁盘按语音时长计，不按墙钟。 + */ +class BlockBuffer(private val root: File, private val sampleRate: Int = 16000) { + + data class Run(val ts: Long, val off: Int, val len: Int) + + private var dir: File? = null + private var out: java.io.BufferedOutputStream? = null + private val runs = ArrayList() + private var sampleCount = 0 + private var blockStartTs = 0L + private var lastSpeechEndTs = 0L + + val isOpen get() = dir != null + + init { + root.mkdirs() + } + + /** 追加一段语音（startTs = 该段真实起始墙钟）。块未开则自动开。 */ + fun append(samples: FloatArray, startTs: Long) { + if (samples.isEmpty()) return + if (!isOpen) open(startTs) + val bytes = ByteArray(samples.size * 2) + val bb = ByteBuffer.wrap(bytes).order(ByteOrder.LITTLE_ENDIAN) + for (s in samples) bb.putShort((s.coerceIn(-1f, 1f) * 32767f).toInt().toShort()) + out!!.write(bytes) + out!!.flush() + runs.add(Run(startTs, sampleCount, samples.size)) + sampleCount += samples.size + lastSpeechEndTs = startTs + samples.size.toLong() * 1000 / sampleRate + writeManifest(dir!!, closed = false) + } + + /** 在 [ts] 这一刻是否该关块（静音超时或块超长）。块未开恒 false。 */ + fun shouldClose(ts: Long): Boolean { + if (!isOpen) return false + return (ts - lastSpeechEndTs >= SILENCE_MS) || (ts - blockStartTs >= MAX_MS) + } + + /** 关块：落 manifest(closed=true)，返回块目录；空块直接删并返回 null。 */ + fun close(): File? { + if (!isOpen) return null + runCatching { out?.flush(); out?.close() } + out = null + val d = dir!! + dir = null + if (sampleCount == 0) { + d.deleteRecursively() + return null + } + writeManifest(d, closed = true) + Log.i(TAG, "block close ${d.name}: ${runs.size} runs / $sampleCount samples") + return d + } + + private fun open(ts: Long) { + blockStartTs = ts + lastSpeechEndTs = ts + runs.clear() + sampleCount = 0 + val d = File(root, ts.toString()).apply { mkdirs() } + dir = d + out = File(d, "audio.pcm").outputStream().buffered() + Log.i(TAG, "block open @$ts -> ${d.name}") + } + + private fun writeManifest(d: File, closed: Boolean) { + val arr = JSONArray() + for (r in runs) arr.put(JSONObject().put("ts", r.ts).put("off", r.off).put("len", r.len)) + val obj = JSONObject() + .put("start", blockStartTs) + .put("sr", sampleRate) + .put("closed", closed) + .put("runs", arr) + runCatching { File(d, "manifest.json").writeText(obj.toString()) } + } + + companion object { + private const val TAG = "BlockBuffer" + const val SILENCE_MS = 5 * 60 * 1000L // 静音 ≥5min 切块 + const val MAX_MS = 10 * 60 * 1000L // 块 ≥10min 强切（长会议保活） + + data class Loaded(val startTs: Long, val sampleRate: Int, val samples: FloatArray, val runs: List) + + /** 启动时列出所有未处理块目录（崩溃恢复：处理后目录会被删除）。 */ + fun listPending(root: File): List = + root.listFiles() + ?.filter { it.isDirectory && File(it, "audio.pcm").exists() } + ?.sortedBy { it.name.toLongOrNull() ?: 0L } + ?: emptyList() + + /** 读块：manifest + audio.pcm → 整块 float + runs。 */ + fun load(dir: File): Loaded? { + val af = File(dir, "audio.pcm") + if (!af.exists()) return null + val mf = File(dir, "manifest.json") + var start = dir.name.toLongOrNull() ?: 0L + var sr = 16000 + val runs = ArrayList() + if (mf.exists()) { + runCatching { + val o = JSONObject(mf.readText()) + start = o.optLong("start", start) + sr = o.optInt("sr", 16000) + o.optJSONArray("runs")?.let { a -> + for (i in 0 until a.length()) { + val r = a.getJSONObject(i) + runs.add(Run(r.getLong("ts"), r.getInt("off"), r.getInt("len"))) + } + } + } + } + val bytes = af.readBytes() + val n = bytes.size / 2 + val floats = FloatArray(n) + val bb = ByteBuffer.wrap(bytes).order(ByteOrder.LITTLE_ENDIAN) + for (i in 0 until n) floats[i] = bb.short / 32768f + // 崩溃时 manifest 可能没记全 runs → 合成一个整块 run（时间退化为连续）。 + if (runs.isEmpty()) runs.add(Run(start, 0, n)) + return Loaded(start, sr, floats, runs) + } + } +} diff --git a/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/service/BlockProcessor.kt b/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/service/BlockProcessor.kt new file mode 100644 index 0000000..5bf7fe1 --- /dev/null +++ b/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/service/BlockProcessor.kt @@ -0,0 +1,105 @@ +package com.lemon.lifelog.service + +import android.util.Log +import com.lemon.lifelog.asr.DiarizationEngine +import com.lemon.lifelog.asr.ParaformerEngine +import com.lemon.lifelog.asr.SpeakerClusterer +import com.lemon.lifelog.asr.SpeakerEmbedder +import com.lemon.lifelog.asr.toLeBytes +import com.lemon.lifelog.audio.SegmentWriter +import com.lemon.lifelog.data.Repository +import com.lemon.lifelog.data.Segment +import com.lemon.lifelog.data.Transcript +import java.io.File +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.withContext + +/** + * 把一个关闭的录音块处理成时间轴 turns（diarization-first 管线的处理端）： + * + * pyannote 分轮 → 每个 turn 切片 → paraformer 转写 + CAMPPlus 嵌入 + 全局 anchor 匹配 + * → 写 turn WAV + `segments` 行 + `transcripts` 行。 + * + * 一个 **turn = 时间轴一行**（一人一段连续发言）。embedding 是单说话人的干净切片嵌入， + * 从源头避免了实时逐句聚类的噪声污染。处理完删块目录（= 已处理标记）。 + * + * [diar] 为 null（pyannote 模型缺失）时降级：整块当一个 turn 转写。 + */ +class BlockProcessor( + private val repo: Repository, + private val segWriter: SegmentWriter, + private val diar: DiarizationEngine?, + private val asr: ParaformerEngine, + private val embedder: SpeakerEmbedder?, + private val clusterer: SpeakerClusterer, +) { + suspend fun process(dir: File) { + val loaded = withContext(Dispatchers.IO) { BlockBuffer.load(dir) } + if (loaded == null || loaded.samples.isEmpty()) { + withContext(Dispatchers.IO) { dir.deleteRecursively() } + return + } + val sr = loaded.sampleRate + val samples = loaded.samples + val runs = loaded.runs + + // 样本下标 → 真实墙钟（speech-only 拼接，用 run 偏移还原）。runs 按 off 升序。 + fun wallAt(sample: Int): Long { + var r = runs.first() + for (run in runs) if (run.off <= sample) r = run else break + val within = (sample - r.off).coerceIn(0, r.len) + return r.ts + within.toLong() * 1000 / sr + } + + var turns = diar?.diarize(samples) ?: emptyList() + if (turns.isEmpty()) { + // 降级 / 极短块：整块一个 turn。 + turns = listOf(DiarizationEngine.Turn(0f, samples.size.toFloat() / sr, 0)) + } + + var written = 0 + for (t in turns) { + val a = (t.startSec * sr).toInt().coerceIn(0, samples.size) + val b = (t.endSec * sr).toInt().coerceIn(a, samples.size) + if (b - a < sr / 5) continue // <200ms 跳过 + val slice = samples.copyOfRange(a, b) + val startTs = wallAt(a) + val durMs = (b - a).toLong() * 1000 / sr + + val text = try { + asr.transcribeFloat(slice, sr).text + } catch (e: Throwable) { + Log.w(TAG, "asr failed", e); "" + } + val emb = embedder?.let { + try { it.embed(slice, sr) } catch (e: Throwable) { Log.w(TAG, "embed failed", e); null } + } + val assign = emb?.let { clusterer.assign(it, durMs) } + ?: SpeakerClusterer.Assignment(null, null, null) + + val audioPath = withContext(Dispatchers.IO) { segWriter.write(slice, startTs, sr) } + withContext(Dispatchers.IO) { + val id = repo.segments.insert( + Segment( + startTs = startTs, + endTs = startTs + durMs, + audioPath = audioPath, + durationMs = durMs, + energyDb = 0f, + embedding = emb?.toLeBytes(), + speakerId = assign.speakerId, + speakerSim = assign.sim, + ), + ) + if (text.isNotBlank()) repo.transcripts.insert(Transcript(segmentId = id, text = text)) + } + written++ + } + Log.i(TAG, "block ${dir.name}: ${turns.size} turns -> $written rows") + withContext(Dispatchers.IO) { dir.deleteRecursively() } + } + + companion object { + private const val TAG = "BlockProcessor" + } +} diff --git a/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/service/RecorderService.kt b/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/service/RecorderService.kt index 8d3030b..e702878 100644 --- a/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/service/RecorderService.kt +++ b/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/service/RecorderService.kt @@ -14,23 +14,21 @@ import com.k2fsa.sherpa.onnx.SileroVadModelConfig import com.k2fsa.sherpa.onnx.Vad import com.k2fsa.sherpa.onnx.VadModelConfig import com.lemon.lifelog.MainActivity +import com.lemon.lifelog.asr.DiarizationEngine import com.lemon.lifelog.asr.ModelDownloader import com.lemon.lifelog.asr.ParaformerEngine import com.lemon.lifelog.asr.SpeakerClusterer import com.lemon.lifelog.asr.SpeakerEmbedder -import com.lemon.lifelog.asr.toLeBytes import com.lemon.lifelog.audio.AudioCapture import com.lemon.lifelog.audio.SegmentWriter import com.lemon.lifelog.data.Repository -import com.lemon.lifelog.data.Segment -import com.lemon.lifelog.data.SpeakerProfile -import com.lemon.lifelog.data.Transcript import java.io.File import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.cancel +import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.flow.collect import kotlinx.coroutines.launch @@ -70,10 +68,17 @@ class RecorderService : Service() { private var vad: Vad? = null private var engine: ParaformerEngine? = null private var speakerEmbedder: SpeakerEmbedder? = null + private var diarEngine: DiarizationEngine? = null /** 在线说话人聚类器（方案 B）：自动发现说话人并持久化为 speaker_profiles。 */ private var clusterer: SpeakerClusterer? = null + /** diarization-first 管线：VAD 段攒进当前块，块关闭后整块 diarize+转写。 */ + private lateinit var blockBuffer: BlockBuffer + + /** 关闭的块排队顺序处理（与采集解耦，不阻塞录音）。 */ + private val pendingBlocks = Channel(Channel.UNLIMITED) + override fun onCreate() { super.onCreate() startForegroundCompat() @@ -83,6 +88,7 @@ class RecorderService : Service() { RecorderState.reset() repo = Repository(this) segWriter = SegmentWriter(File(filesDir, "segments")) + blockBuffer = BlockBuffer(File(filesDir, "blocks")) capture = AudioCapture() pipelineJob = scope.launch { runPipeline() } @@ -113,7 +119,7 @@ class RecorderService : Service() { acquireWakeLock() capture.start(scope) RecorderState.setPaused(false) - RecorderState.setStatus("listening") + RecorderState.setStatus("录音中") notify("录音中") Log.i(TAG, "resumed") } @@ -126,9 +132,13 @@ class RecorderService : Service() { override fun onDestroy() { pipelineJob?.cancel() capture.stop() + // 关掉当前块（落 manifest 留在磁盘，下次启动 listPending 续处理）。 + runCatching { blockBuffer.close() } + pendingBlocks.close() vad?.release() engine?.release() speakerEmbedder?.release() + diarEngine?.release() scope.cancel() wakeLock?.takeIf { it.isHeld }?.release() wakeLock = null @@ -210,94 +220,73 @@ class RecorderService : Service() { repo.speakers.observeAll().collect { clusterer?.load() } } + // 2e. Diarization 引擎（pyannote 分割 + CAMPPlus），块级整段分轮用。缺失则降级单 turn。 + try { + RecorderState.setStatus("loading diarization…") + val pyannote = downloader.ensurePyannoteModel { Log.i(TAG, "pyannote dl: $it") } + val speaker = downloader.speakerPath() + if (pyannote.present && speaker.present) { + diarEngine = DiarizationEngine.load( + pyannote.path, speaker.path, numThreads = 4, + threshold = SpeakerClusterer.thresholdFromPrefs(this), + ) + } else { + Log.w(TAG, "pyannote/speaker model missing; block will fall back to single-turn") + } + } catch (t: Throwable) { + Log.w(TAG, "diarization engine unavailable; single-turn fallback", t) + } + + // 2f. 块处理器 + 顺序处理协程（与采集解耦）。 + val processor = BlockProcessor(repo, segWriter, diarEngine, engine!!, speakerEmbedder, clusterer!!) + scope.launch { + for (dir in pendingBlocks) { + RecorderState.setStatus("处理段落…") + try { + processor.process(dir) + } catch (t: Throwable) { + Log.w(TAG, "block ${dir.name} process failed", t) + } + RecorderState.setStatus(if (paused) "已暂停" else "录音中") + } + } + // 崩溃恢复：把上次没处理完的块重新入队。 + BlockBuffer.listPending(File(filesDir, "blocks")).forEach { pendingBlocks.trySend(it) } + // 3. Start capturing (unless paused arrived during model load). if (!paused) { capture.start(scope) - RecorderState.setStatus("listening") + RecorderState.setStatus("录音中") } - Log.i(TAG, "pipeline running, clusterer loaded") + Log.i(TAG, "pipeline running (block-based diarization-first)") capture.frames.collect { samples -> val v = vad ?: return@collect v.acceptWaveform(samples) + val now = System.currentTimeMillis() while (!v.empty()) { val seg = v.front() v.pop() if (seg.samples.isEmpty()) continue - handleSegment(seg.samples) + appendSpeech(seg.samples) } + // 静音超时 / 块超长 → 关块入队（即使本帧没有新语音也要检查）。 + if (blockBuffer.shouldClose(now)) closeBlock() } } - private suspend fun handleSegment(samples: FloatArray) { + /** VAD 段 → 攒进当前块（不再实时转写/聚类）。跨静音/超长时先关旧块。 */ + private fun appendSpeech(samples: FloatArray) { val sr = AudioCapture.SAMPLE_RATE val audioMs = samples.size.toLong() * 1000 / sr - val now = System.currentTimeMillis() - val startTs = now - audioMs - - // a. persist audio (off CPU thread, Opus encode). - val audioPath = kotlinx.coroutines.withContext(Dispatchers.IO) { - segWriter.write(samples, startTs, sr) - } - - // b. paraformer. - val wallStart = System.nanoTime() - val text = try { - engine!!.transcribeFloat(samples, sr).text - } catch (t: Throwable) { - Log.w(TAG, "transcribe failed", t) - "" - } - val wallMs = (System.nanoTime() - wallStart) / 1_000_000 - - // c. speaker embedding + online clustering (auto-discovers 说话人N). - val embedding: FloatArray? = speakerEmbedder?.let { embedder -> - try { - embedder.embed(samples, sr) - } catch (t: Throwable) { - Log.w(TAG, "embed failed", t) - null - } - } - val assignment = embedding?.let { emb -> - clusterer?.assign(emb, audioMs) - } ?: SpeakerClusterer.Assignment(null, null, null) - val matchedSpeakerId = assignment.speakerId - val matchedSim = assignment.sim - val matchedName = assignment.name - - // d. Room insert. - val segmentId = kotlinx.coroutines.withContext(Dispatchers.IO) { - val id = repo.segments.insert( - Segment( - startTs = startTs, - endTs = now, - audioPath = audioPath, - durationMs = audioMs, - energyDb = 0f, - embedding = embedding?.toLeBytes(), - speakerId = matchedSpeakerId, - speakerSim = matchedSim, - ), - ) - if (text.isNotEmpty()) { - repo.transcripts.insert( - Transcript( - segmentId = id, - text = text, - ), - ) - } - id - } + val startTs = System.currentTimeMillis() - audioMs + if (blockBuffer.isOpen && blockBuffer.shouldClose(startTs)) closeBlock() + blockBuffer.append(samples, startTs) + } - Log.i( - TAG, - "segment#$segmentId audio=${audioMs}ms wall=${wallMs}ms " + - "speaker=${matchedName ?: "—"} sim=${matchedSim?.let { "%.2f".format(it) } ?: "—"} " + - "text=$text", - ) - RecorderState.onSegment(text, audioMs, wallMs) + private fun closeBlock() { + val dir = blockBuffer.close() ?: return + pendingBlocks.trySend(dir) } private fun startForegroundCompat() { -- libgit2 0.21.4