starpivot/apps/client/lib/data/utils/pcm_stream_player.dart

import 'dart:async';
import 'dart:math' as math;
import 'dart:typed_data';

import 'package:audio_session/audio_session.dart';
import 'package:just_audio/just_audio.dart';

import '../../core/utils/logger.dart';

/// 流式 PCM 播放器：一边收字节一边出声，用于百炼下行的 TTS 音频。
///
/// 实现方式是给 just_audio 喂一个「长度未知」的 [StreamAudioSource]：
/// just_audio 内部会起一个本地代理 HTTP 服务，`contentLength == null` 时
/// 它按 chunked 传输回给播放器，ExoPlayer / AVPlayer 都当直播流处理，
/// 于是首包到达就能出声，不用等整段合成完。
///
/// WAV 头里的 RIFF/data 长度写 [_unknownSize]（0xFFFFFFFF）——
/// 这是流式 WAV 的通行写法，表示"长度未知，读到流结束为止"。
/// 写真实长度是不可能的：合成还没结束，长度就还不知道。
///
/// 一轮回复对应一个 source：[begin] → 若干次 [feed] → [end]。
/// [end] 只是关掉输入流，播放器会把已经收到的字节播完再报 [onCompleted]。
/// 打断走 [stopNow]，立即断流并停止播放。
class PcmStreamPlayer {
  PcmStreamPlayer({
    this.sampleRate = 24000,
    this.channels = 1,
  });

  static const String _tag = 'PcmStreamPlayer';

  /// 流式 WAV 的"长度未知"魔数
  static const int _unknownSize = 0xFFFFFFFF;

  final int sampleRate;
  final int channels;

  final AudioPlayer _player = AudioPlayer();
  _LivePcmSource? _source;
  StreamSubscription<PlayerState>? _stateSub;
  StreamSubscription<PlaybackEvent>? _errSub;
  bool _sessionConfigured = false;

  /// 当前这轮是否已经建好 source
  bool _active = false;

  /// [begin] 正在进行中。用来挡住并发：下行音频包比 RespondingStarted 先到时，
  /// [feed] 会自己去 begin，而后续几包还会连着进来——没有这个标志就会同时起好几轮。
  bool _starting = false;

  /// begin 期间到达的音频，等 source 建好再补进去，避免开头被吞
  final _pending = BytesBuilder(copy: false);

  bool get isPlaying => _player.playing;

  final _levelCtl = StreamController<double>.broadcast();

  /// 下行音频的实时音量（0~1），驱动 UI 的说话动效
  Stream<double> get level => _levelCtl.stream;

  final _completedCtl = StreamController<void>.broadcast();

  /// 一轮回复的音频全部播完（或本轮压根没有音频）。
  /// 调用方据此给服务端回 LocalRespondingEnded。
  Stream<void> get onCompleted => _completedCtl.stream;

  /// 配置音频会话（只做一次；配置本身是静态的）并**每轮激活一次**。
  ///
  /// ⚠️ 激活必须每轮都做，不能只在首次配置时做（iOS 真机踩到）：
  /// 手机这条路径是「按住说话」，`MicPcmStreamer` 用 record 的
  /// `echoCancel/noiseSuppress`，在 iOS 上会让插件把 AVAudioSession 切到
  /// voice-processing（.voiceChat）并自己持有；松手时 `_mic.stop()` 只停引擎，
  /// **不会把会话还原**。随后这一轮回复音频到达，AVPlayer 在那个会话上
  /// 起不来，报 `-11850 Operation Stopped`，而且每来一包就重试一次，
  /// 日志里刷成几十条。
  ///
  /// 显式 setActive(true) 是把会话按本方的配置重新激活一次，
  /// 相当于「录音刚放开，播放把它拿回来」。
  Future<void> _ensureSession() async {
    try {
      final session = await AudioSession.instance;
      if (_sessionConfigured) {
        await session.setActive(true);
        return;
      }
      _sessionConfigured = true;
      // ⚠️ usage 千万别用 voiceCommunication。
      // 华为(EMUI/HarmonyOS)的音频策略对通话流卡得很严：不在
      // MODE_IN_COMMUNICATION 下建的 VOICE_COMMUNICATION 音轨会被直接静音，
      // logcat 里表现为 `AudioTrack: set playback slient` —— 音轨建得出来、
      // 数据也在写，就是没声音，非常难查。ALN-AL80 上实测复现。
      //
      // 用 media 就没这问题。手机这条路径上录音和播放本来也不并发：
      // 按住说话时 startTalk() 会先把播放停掉，所以不需要占通话流来共存。
      // （耳机那条路径的下行走 SPP/G.722，压根不经过 AudioTrack。）
      await session.configure(const AudioSessionConfiguration(
        // iOS 仍要 playAndRecord：按住说话要录音，纯 playback 会把录音会话踢掉。
        // 但 mode 从 voiceChat 换成 default —— voiceChat 会上通话处理链，
        // 在部分机型上把输出拽到听筒。
        avAudioSessionCategory: AVAudioSessionCategory.playAndRecord,
        avAudioSessionCategoryOptions:
            AVAudioSessionCategoryOptions.defaultToSpeaker,
        avAudioSessionMode: AVAudioSessionMode.defaultMode,
        androidAudioAttributes: AndroidAudioAttributes(
          contentType: AndroidAudioContentType.speech,
          usage: AndroidAudioUsage.media,
        ),
        // 助手说话时把背景音乐暂停掉，而不是压低——压低了在嘈杂环境里还是听不清
        androidAudioFocusGainType: AndroidAudioFocusGainType.gainTransient,
      ));
      await session.setActive(true);
    } catch (e) {
      Logger.w(_tag, '音频会话配置失败（继续尝试播放）: $e');
    }
  }

  /// 开一轮新的播放。并发调用只有第一次生效。
  Future<void> begin() async {
    if (_starting) return;
    _starting = true;
    try {
      if (_active) await _teardown();
      await _ensureSession();

      final src = _LivePcmSource(_wavHeader());
      _source = src;
      _active = true;

      _stateSub = _player.playerStateStream.listen((s) {
        if (s.processingState == ProcessingState.completed) _finish();
      });
      // 播放器报错（解码失败、代理断开）也要收尾，
      // 否则 LocalRespondingEnded 永远不发，服务端会一直卡在 Responding
      _errSub = _player.playbackEventStream.listen(
        (_) {},
        onError: (Object e) {
          Logger.e(_tag, '播放错误: $e');
          _finish();
        },
      );

      await _player.setAudioSource(src, preload: false);
      await _player.play();

      if (_pending.length > 0) src.add(_pending.takeBytes());
    } catch (e) {
      Logger.e(_tag, '启动播放失败: $e');
      _active = false;
      _pending.clear();
      // 起播失败多半是会话被别人（录音的 voice-processing）占着。
      // 复位标志，下一轮重新走完整的 configure + setActive，
      // 而不是只补一次 setActive —— 只激活救不回被改过 category/mode 的会话。
      _sessionConfigured = false;
    } finally {
      _starting = false;
    }
  }

  /// 喂一包下行 PCM（16bit / [sampleRate] / [channels]，小端）
  void feed(Uint8List pcm) {
    if (pcm.isEmpty) return;
    _emitLevel(pcm);

    final src = _source;
    if (_active && src != null) {
      src.add(pcm);
      return;
    }
    // RespondingStarted 不一定先于首包音频到：先缓冲，顺手把这一轮拉起来
    _pending.add(pcm);
    if (!_starting) unawaited(begin());
  }

  /// 本轮下行结束：关掉输入流，剩余缓冲播完后会触发 [onCompleted]。
  /// 这一轮一个音频字节都没有时直接报完成——不然调用方会一直等一个不会来的回调。
  Future<void> end() async {
    final src = _source;
    if (src == null || !src.hasData) {
      await _teardown();
      _emitCompleted();
      return;
    }
    await src.close();
  }

  /// 打断：立刻断流停声，不报完成
  Future<void> stopNow() async {
    await _teardown();
    if (!_levelCtl.isClosed) _levelCtl.add(0);
  }

  Future<void> dispose() async {
    await _teardown();
    await _levelCtl.close();
    await _completedCtl.close();
    await _player.dispose();
  }

  Future<void> _teardown() async {
    _active = false;
    _pending.clear();
    await _stateSub?.cancel();
    _stateSub = null;
    await _errSub?.cancel();
    _errSub = null;
    final src = _source;
    _source = null;
    await src?.close();
    try {
      await _player.stop();
    } catch (_) {}
  }

  /// 播完/出错的统一收尾
  void _finish() {
    _active = false;
    _stateSub?.cancel();
    _stateSub = null;
    _errSub?.cancel();
    _errSub = null;
    _source = null;
    _emitCompleted();
    if (!_levelCtl.isClosed) _levelCtl.add(0);
  }

  void _emitCompleted() {
    if (!_completedCtl.isClosed) _completedCtl.add(null);
  }

  void _emitLevel(Uint8List pcm) {
    if (_levelCtl.isClosed) return;
    _levelCtl.add(rmsLevel(pcm));
  }

  /// 44 字节标准 WAV 头，长度字段填"未知"
  Uint8List _wavHeader() {
    const int bitsPerSample = 16;
    final int byteRate = sampleRate * channels * bitsPerSample ~/ 8;
    final int blockAlign = channels * bitsPerSample ~/ 8;

    final b = BytesBuilder();
    void ascii(String s) => b.add(s.codeUnits);
    void u32(int v) =>
        b.add(Uint8List(4)..buffer.asByteData().setUint32(0, v, Endian.little));
    void u16(int v) =>
        b.add(Uint8List(2)..buffer.asByteData().setUint16(0, v, Endian.little));

    ascii('RIFF');
    u32(_unknownSize); // 长度未知
    ascii('WAVE');
    ascii('fmt ');
    u32(16); // PCM fmt chunk 长度
    u16(1); // PCM
    u16(channels);
    u32(sampleRate);
    u32(byteRate);
    u16(blockAlign);
    u16(bitsPerSample);
    ascii('data');
    u32(_unknownSize); // 长度未知
    return b.toBytes();
  }

  /// 16bit 小端 PCM 的归一化 RMS（0~1），UI 音浪用
  static double rmsLevel(Uint8List pcm) {
    if (pcm.length < 2) return 0;
    final data = pcm.buffer.asByteData(pcm.offsetInBytes, pcm.length);
    final int samples = pcm.length ~/ 2;
    double sum = 0;
    for (int i = 0; i < samples; i++) {
      final s = data.getInt16(i * 2, Endian.little) / 32768.0;
      sum += s * s;
    }
    final rms = math.sqrt(sum / samples);
    // 语音 RMS 常年在 0.02~0.3，直接用几乎看不出动静，开方拉一下动态范围
    return math.min(1.0, math.sqrt(rms) * 1.6);
  }
}

/// 长度未知的 WAV 流：头部固定，后面是不断追加的 PCM。
///
/// [request] 只会被代理服务调用一次（`rangeRequestsSupported: false`，
/// 播放器不会再发 Range 请求），所以内部用单订阅 StreamController 即可。
class _LivePcmSource extends StreamAudioSource {
  _LivePcmSource(this._header);

  final Uint8List _header;
  final _ctl = StreamController<List<int>>();
  bool _headerSent = false;

  /// 是否已经收到过真实音频数据
  bool get hasData => _headerSent;

  void add(Uint8List pcm) {
    if (_ctl.isClosed) return;
    if (!_headerSent) {
      _headerSent = true;
      _ctl.add(_header);
    }
    _ctl.add(pcm);
  }

  Future<void> close() async {
    if (_ctl.isClosed) return;
    await _ctl.close();
  }

  @override
  Future<StreamAudioResponse> request([int? start, int? end]) async {
    return StreamAudioResponse(
      // 长度未知 → 关掉 Range 支持，代理会走 chunked 200 而不是 206
      rangeRequestsSupported: false,
      sourceLength: null,
      contentLength: null,
      offset: null,
      stream: _ctl.stream,
      contentType: 'audio/wav',
    );
  }
}