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 399ecde..44a3d7b 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,6 +3,7 @@ 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 @@ -17,7 +18,8 @@ class LifelogApp : Application() { NotificationManager.IMPORTANCE_LOW, ) ) - // Hourly 纪要 worker (KEEP = 已排过就不动)。 + // 纪要 + diarization periodic worker(各自 15min,UPDATE)。 SummaryScheduler.schedulePeriodic(this) + DiarizationScheduler.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 80881af..e148cf9 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,6 +83,7 @@ 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 @@ -848,7 +849,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 = { SummaryScheduler.runNow(activity) }, enabled = !running) { + Button(onClick = { DiarizationScheduler.runNow(activity); 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/DiarizationEngine.kt b/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/asr/DiarizationEngine.kt new file mode 100644 index 0000000..861b98d --- /dev/null +++ b/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/asr/DiarizationEngine.kt @@ -0,0 +1,61 @@ +package com.lemon.lifelog.asr + +import android.util.Log +import com.k2fsa.sherpa.onnx.FastClusteringConfig +import com.k2fsa.sherpa.onnx.OfflineSpeakerDiarization +import com.k2fsa.sherpa.onnx.OfflineSpeakerDiarizationConfig +import com.k2fsa.sherpa.onnx.OfflineSpeakerSegmentationModelConfig +import com.k2fsa.sherpa.onnx.OfflineSpeakerSegmentationPyannoteModelConfig +import com.k2fsa.sherpa.onnx.SpeakerEmbeddingExtractorConfig +import java.io.File + +/** + * 离线说话人 diarization(sherpa-onnx):pyannote 分割(检测说话人变更点 + 重叠语音) + * + CAMPPlus embedding + FastClustering 自动聚类。跑在一整段会话音频上,输出 + * [start,end]→speaker turns,能把「几秒内多人」的区域拆成一人一段。 + * + * 与逐 VAD 段单 embedding 的实时方案不同——这是会话级批处理重跑,准确得多。 + */ +class DiarizationEngine private constructor(private val sd: OfflineSpeakerDiarization) { + + data class Turn(val startSec: Float, val endSec: Float, val speaker: Int) + + /** 16k mono PCM float([-1,1])→ 说话人 turns(按时间排序)。 */ + fun diarize(samples: FloatArray): List = + sd.process(samples).map { Turn(it.start, it.end, it.speaker) }.sortedBy { it.startSec } + + fun sampleRate(): Int = sd.sampleRate() + + fun release() = sd.release() + + companion object { + private const val TAG = "DiarizationEngine" + + /** + * @param threshold FastClustering 余弦阈值;越大分越细(同 SpeakerClusterer 取值习惯)。 + */ + fun load( + segModelPath: String, + embModelPath: String, + numThreads: Int = 4, + threshold: Float = 0.5f, + ): DiarizationEngine { + require(File(segModelPath).exists()) { "pyannote seg model 缺失: $segModelPath" } + require(File(embModelPath).exists()) { "CAMPPlus emb model 缺失: $embModelPath" } + val config = OfflineSpeakerDiarizationConfig( + segmentation = OfflineSpeakerSegmentationModelConfig( + pyannote = OfflineSpeakerSegmentationPyannoteModelConfig(model = segModelPath), + numThreads = numThreads, + ), + embedding = SpeakerEmbeddingExtractorConfig(model = embModelPath, numThreads = numThreads), + clustering = FastClusteringConfig(numClusters = -1, threshold = threshold), + minDurationOn = 0.2f, + minDurationOff = 0.5f, + ) + val t0 = System.nanoTime() + val sd = OfflineSpeakerDiarization(config = config) + Log.i(TAG, "loaded in ${(System.nanoTime() - t0) / 1_000_000}ms sr=${sd.sampleRate()} thr=$threshold") + return DiarizationEngine(sd) + } + } +} diff --git a/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/asr/ModelDownloader.kt b/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/asr/ModelDownloader.kt index 692d48c..66aa720 100644 --- a/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/asr/ModelDownloader.kt +++ b/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/asr/ModelDownloader.kt @@ -99,6 +99,31 @@ class ModelDownloader(private val ctx: Context) { speakerPath() } + fun pyannotePath(): SpeakerModelFile { + val model = File(ctx.filesDir, "models/pyannote-seg.onnx") + return SpeakerModelFile(model.absolutePath, model.exists() && model.length() > 4_000_000L) + } + + /** pyannote-segmentation-3.0(~6MB)—— diarization 的分割模型。 */ + suspend fun ensurePyannoteModel(progress: (String) -> Unit): SpeakerModelFile = + withContext(Dispatchers.IO) { + val state = pyannotePath() + if (state.present) { + progress("pyannote model present, skip"); return@withContext state + } + File(ctx.filesDir, "models").mkdirs() + val task = DownloadTask( + File(state.path), + listOf( + "https://hf-mirror.com/csukuangfj/sherpa-onnx-pyannote-segmentation-3-0/resolve/main/model.onnx", + "https://huggingface.co/csukuangfj/sherpa-onnx-pyannote-segmentation-3-0/resolve/main/model.onnx", + ), + expectedSize = 5_992_913L, + ) + downloadWithResume(http(), task, progress) + pyannotePath() + } + private fun http() = OkHttpClient.Builder() .connectTimeout(15, TimeUnit.SECONDS) .readTimeout(120, TimeUnit.SECONDS) 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 new file mode 100644 index 0000000..ab66c20 --- /dev/null +++ b/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/audio/AudioDecoder.kt @@ -0,0 +1,126 @@ +package com.lemon.lifelog.audio + +import android.media.MediaCodec +import android.media.MediaExtractor +import android.media.MediaFormat +import android.util.Log +import java.io.File +import java.nio.ByteOrder + +/** + * 把 SegmentWriter 写的 .opus(Opus-in-Ogg)解码回 16k mono PCM float([-1,1])。 + * + * 两个坑: + * - MediaCodec 的 Opus 解码常输出 48kHz(Opus 原生率),与 diarization 期望的 16k 不符 → 重采样。 + * - 编码是 mono,但保险起见处理 stereo → 下混。 + */ +object AudioDecoder { + private const val TAG = "AudioDecoder" + private const val TARGET_SR = 16000 + + fun decodeToFloat(path: String): FloatArray { + if (!File(path).exists()) return FloatArray(0) + val extractor = MediaExtractor() + return try { + extractor.setDataSource(path) + var track = -1 + var inFormat: MediaFormat? = null + for (i in 0 until extractor.trackCount) { + val f = extractor.getTrackFormat(i) + if (f.getString(MediaFormat.KEY_MIME)?.startsWith("audio/") == true) { + track = i; inFormat = f; break + } + } + if (track < 0 || inFormat == null) return FloatArray(0) + extractor.selectTrack(track) + + val codec = MediaCodec.createDecoderByType(inFormat.getString(MediaFormat.KEY_MIME)!!) + codec.configure(inFormat, null, null, 0) + codec.start() + + 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 info = MediaCodec.BufferInfo() + var inEos = false + var outEos = false + val timeout = 10_000L + + while (!outEos) { + if (!inEos) { + val ii = codec.dequeueInputBuffer(timeout) + if (ii >= 0) { + val buf = codec.getInputBuffer(ii)!! + val n = extractor.readSampleData(buf, 0) + if (n < 0) { + codec.queueInputBuffer(ii, 0, 0, 0, MediaCodec.BUFFER_FLAG_END_OF_STREAM) + inEos = true + } else { + codec.queueInputBuffer(ii, 0, n, extractor.sampleTime, 0) + extractor.advance() + } + } + } + val oi = codec.dequeueOutputBuffer(info, timeout) + when { + oi == MediaCodec.INFO_OUTPUT_FORMAT_CHANGED -> { + val of = codec.outputFormat + srcRate = of.getIntOr(MediaFormat.KEY_SAMPLE_RATE, srcRate) + channels = of.getIntOr(MediaFormat.KEY_CHANNEL_COUNT, channels) + } + 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()) + } + codec.releaseOutputBuffer(oi, false) + if (info.flags and MediaCodec.BUFFER_FLAG_END_OF_STREAM != 0) outEos = true + } + } + } + codec.stop(); codec.release() + + // 下混 stereo → mono + var mono = if (channels >= 2) { + FloatArray(shorts.size / channels) { i -> + var s = 0f + for (c in 0 until channels) s += shorts[i * channels + c] / 32768f + s / channels + } + } else { + FloatArray(shorts.size) { shorts[it] / 32768f } + } + if (srcRate != TARGET_SR) mono = resample(mono, srcRate, TARGET_SR) + mono + } catch (t: Throwable) { + Log.w(TAG, "decode failed $path: ${t.message}") + FloatArray(0) + } finally { + extractor.release() + } + } + + private fun MediaFormat.getIntOr(key: String, def: Int): Int = + if (containsKey(key)) getInteger(key) else def + + /** 线性重采样到 dstRate。 */ + private fun resample(input: FloatArray, srcRate: Int, dstRate: Int): FloatArray { + if (srcRate == dstRate || input.isEmpty()) return input + val ratio = dstRate.toDouble() / srcRate + val outLen = (input.size * ratio).toInt() + val out = FloatArray(outLen) + val last = input.size - 1 + for (i in out.indices) { + val srcPos = i / ratio + val i0 = srcPos.toInt() + val frac = (srcPos - i0).toFloat() + val a = input[i0.coerceIn(0, last)] + val b = input[(i0 + 1).coerceIn(0, last)] + out[i] = a + (b - a) * frac + } + return out + } +} 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 new file mode 100644 index 0000000..38ce850 --- /dev/null +++ b/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/diarize/DiarizationScheduler.kt @@ -0,0 +1,37 @@ +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 new file mode 100644 index 0000000..aa5df32 --- /dev/null +++ b/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/diarize/DiarizationWorker.kt @@ -0,0 +1,175 @@ +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.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) + 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,防内存/超时 + private const val KEY_DONE = "diarized_sessions" // prefs StringSet of windowStartTs + } + + 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 + try { + for (session in sessions) { + 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) + } + + val ok = diarizeSession(repo, session, engine!!, embedder!!, threshold) + if (ok) { done.add(key); processed++ } + } + } finally { + engine?.release() + embedder?.release() + } + + prefs.edit().putStringSet(KEY_DONE, done).apply() + Log.i(TAG, "done: ${sessions.size} sessions, $processed diarized") + return Result.success() + } + + /** 对单个会话做 diarization + 段重标。返回是否成功处理。 */ + private suspend fun diarizeSession( + repo: Repository, + session: List, + engine: DiarizationEngine, + embedder: SpeakerEmbedder, + threshold: Float, + ): Boolean = withContext(Dispatchers.Default) { + // 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) + 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 + } + if (spans.size < MIN_SESSION_SEGMENTS || audio.isEmpty()) return@withContext false + val samples = FloatArray(audio.size) { audio[it] } + + // 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 + for (span in spans) { + val s0 = span.from.toFloat() / sr + val s1 = span.to.toFloat() / sr + var bestTurn: DiarizationEngine.Turn? = null + var bestOv = 0f + for (t in turns) { + val ov = minOf(s1, t.endSec) - maxOf(s0, t.startSec) + if (ov > bestOv) { bestOv = ov; bestTurn = t } + } + 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 + } +}