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

import 'dart:async';
import 'dart:io' show Platform;
import 'dart:math' as math;
import 'dart:typed_data';

import 'package:flutter/foundation.dart' show kIsWeb;
import 'package:flutter/services.dart';
import 'package:just_audio/just_audio.dart';

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

/// 流式 PCM 播放器：一边收字节一边出声，用于百炼下行的 TTS 音频。
///
/// **默认走原生播放器**（`azure_speech/pcm_player`，iOS `AVAudioEngine` /
/// Android `AudioTrack`，两端方法名逐字一致），PCM 直接进音频硬件。
/// 只有原生不可用（老包没有这条通道）时才退回下面那套 just_audio 的路径。
/// 两条路的取舍见 [_useNative]。
///
/// ## 退路：just_audio 的「长度未知」流
/// 给 just_audio 喂一个 [StreamAudioSource]，它内部起一个**本地代理 HTTP 服务**，
/// `contentLength == null` 时按 chunked 传输回给播放器，ExoPlayer 当直播流处理，
/// 于是首包到达就能出声。WAV 头里的 RIFF/data 长度写 [_unknownSize]（0xFFFFFFFF）——
/// 流式 WAV 的通行写法，表示"长度未知，读到流结束为止"；写真实长度是不可能的，
/// 合成还没结束长度就还不知道。
///
/// 一轮回复对应一轮播放：[begin] → 若干次 [feed] → [end]。
/// [end] 只是告诉播放器没有更多数据了，它会把已经收到的字节**播完**再报
/// [onCompleted]（原生侧按已渲染帧数判定，不是队列排空就算完，否则尾音会被切）。
/// 打断走 [stopNow]，立即断流并停止播放。
class PcmStreamPlayer {
  PcmStreamPlayer({
    this.sampleRate = 24000,
    this.channels = 1,
    this.playbackOnly = false,
  });

  static const String _tag = 'PcmStreamPlayer';

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

  final int sampleRate;
  final int channels;

  /// 这条链路手机**不录音**（麦克风在设备上）。
  ///
  /// true 时音频会话走 [EmaiAudioProfile.playbackOnly]：输出跟随系统路由，
  /// 蓝牙 A2DP 连着就从设备喇叭出来。默认 false 是手机麦克风那条链路，
  /// 必须 playAndRecord，不能改。
  final bool playbackOnly;

  /// 原生 PCM 播放通道（两端同名同方法，见 [_useNative] 的说明）
  static const MethodChannel _nativeChannel =
      MethodChannel('azure_speech/pcm_player');

  /// 这个包里有没有原生播放器。老包没有这条通道 → `MissingPluginException`，
  /// 置 false 之后本进程一律退回 just_audio，功能不受影响。
  static bool _nativeUnavailable = false;

  /// 走原生播放器而不是 just_audio。
  ///
  /// - **iOS 上是被迫的**：AVPlayer 放不了 just_audio 代理服务给的那种
  ///   「长度未知的 chunked WAV」，只发一次 `Range: bytes=0-1` 的嗅探请求就判定
  ///   资源不可播、回 `-11850 Operation Stopped`，一次都没真正起播。
  /// - **Android 上是为了少绕一圈**：ExoPlayer 不发那个嗅探请求，所以老路能用，
  ///   但一条实时语音流要先经过 just_audio 内部起的**本地代理 HTTP 服务**
  ///   （chunked 编码 → localhost TCP → ExoPlayer）。白白多一次拷贝、一次 HTTP
  ///   分帧和 ExoPlayer 自己的起播缓冲。直接喂 AudioTrack 更短也更快出声。
  ///
  /// ⚠️ 只支持**单声道**：两端的原生播放器都是按 mono 建的链路
  /// （iOS 的 AVAudioFormat、Android 的 CHANNEL_OUT_MONO）。真出现双声道用法时
  /// 退回 just_audio，而不是把交织 PCM 当单声道播成两倍语速。
  /// 当前两个调用点（EMAI 24k、车载香薰 16k）都是单声道。
  bool get _useNative =>
      !kIsWeb &&
      !_nativeUnavailable &&
      channels == 1 &&
      (Platform.isIOS || Platform.isAndroid);

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

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

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

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

  bool get isPlaying => _useNative ? _active : _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() => EmaiAudioSession.ensure(
        profile: playbackOnly
            ? EmaiAudioProfile.playbackOnly
            : EmaiAudioProfile.phoneMic,
      );

  /// 正在进行中的 [begin]。[end] 要等它落地才能判断这一轮有没有音频——
  /// 回复很短时（音频首包和 RespondingEnded 几乎同时到）`_active` 还是 false，
  /// 不等就会把整轮回复当成「没有音频」直接报完成，声音全丢。
  Future<void>? _startFuture;

  /// 开一轮新的播放。并发调用共用同一个 future。
  Future<void> begin() {
    if (_starting) return _startFuture ?? Future<void>.value();
    _starting = true;
    return _startFuture = _begin();
  }

  Future<void> _begin() async {
    try {
      if (_active) await _teardown();
      await _ensureSession();

      if (_useNative) {
        final started = await _startNative();
        if (started) return;
        // 起不来的原因有两种：老包没有这条通道（已置 _nativeUnavailable，
        // 下面直接走 just_audio），或者原生自己失败了（那就这一轮没声音）。
        if (!_nativeUnavailable) return;
        Logger.w(_tag, '原生播放器不可用，本进程退回 just_audio');
      }

      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();
    } finally {
      _starting = false;
    }
  }

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

    if (_useNative) {
      if (_active) {
        _nativeChannel.invokeMethod('feed', {'data': pcm});
      } else {
        _pending.add(pcm);
        if (!_starting) unawaited(begin());
      }
      return;
    }

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

  /// 本轮下行结束：关掉输入流，剩余缓冲播完后会触发 [onCompleted]。
  /// 这一轮一个音频字节都没有时直接报完成——不然调用方会一直等一个不会来的回调。
  Future<void> end() async {
    if (_useNative) {
      // 起播还在路上就先等它：不等的话短回复会被判成「没有音频」
      if (_starting) {
        try {
          await _startFuture;
        } catch (_) {}
      }
      if (!_active) {
        // 这一轮一个字节都没来过：原生侧没东西可播，直接报完成
        _emitCompleted();
        return;
      }
      await _nativeChannel.invokeMethod('end');
      return;
    }
    final src = _source;
    if (src == null || !src.hasData) {
      await _teardown();
      _emitCompleted();
      return;
    }
    await src.close();
  }

  /// 起原生播放器。返回 false 且 [_nativeUnavailable] 为 true 时表示这个包
  /// 压根没有原生播放器，调用方应改走 just_audio。
  Future<bool> _startNative() async {
    try {
      _nativeChannel.setMethodCallHandler(_onNativeCall);
      final ok =
          await _nativeChannel.invokeMethod<bool>('start', {
            'sampleRate': sampleRate,
            // 只有车载香薰的设备会话用 playbackOnly（声音走 A2DP）。Android 原生据此在
            // 回复之间垫静音，别让 A2DP 流被挂起吞掉下一句的开头；其它调用方不受影响。
            'keepAlive': playbackOnly,
          });
      if (ok != true) {
        Logger.e(_tag, '原生播放器启动失败');
        return false;
      }
      _active = true;
      if (_pending.length > 0) {
        await _nativeChannel.invokeMethod('feed', {'data': _pending.takeBytes()});
      }
      return true;
    } on MissingPluginException {
      _nativeUnavailable = true;
      return false;
    } catch (e) {
      Logger.e(_tag, '原生播放器启动异常: $e');
      return false;
    }
  }

  /// 原生播放器回调：本轮排队的 PCM 全部播完
  Future<dynamic> _onNativeCall(MethodCall call) async {
    if (call.method == 'onCompleted') _finish();
    return null;
  }

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

  Future<void> dispose() async {
    await _teardown();
    await _levelCtl.close();
    await _completedCtl.close();
    if (_useNative) {
      try {
        await _nativeChannel.invokeMethod('dispose');
      } catch (_) {}
      return;
    }
    await _player.dispose();
  }

  Future<void> _teardown() async {
    _active = false;
    _pending.clear();
    if (_useNative) {
      try {
        await _nativeChannel.invokeMethod('stop');
      } catch (_) {}
      return;
    }
    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();
    _resetLevelWindow();
    if (!_levelCtl.isClosed) _levelCtl.add(0);
  }

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

  /// 音浪发布间隔。下行包到得比屏幕刷新还密，包包都发等于让
  /// 界面在两帧画面之间白白重建好几次。
  static const Duration _levelEmitInterval = Duration(milliseconds: 100);

  /// 本窗口内的峰值 / 上次发布时刻
  double _levelPeak = 0;
  DateTime? _levelEmitAt;

  /// 取窗口峰值而不是"最后一包"：说话的尖峰往往只占一两包，
  /// 按最后一包取样会把音浪削平。
  void _emitLevel(Uint8List pcm) {
    if (_levelCtl.isClosed) return;
    final v = rmsLevel(pcm);
    if (v > _levelPeak) _levelPeak = v;
    final now = DateTime.now();
    final last = _levelEmitAt;
    if (last != null && now.difference(last) < _levelEmitInterval) return;
    _levelEmitAt = now;
    _levelCtl.add(_levelPeak);
    _levelPeak = 0;
  }

  /// 一轮结束/打断时把窗口清掉，下一轮从零开始计峰值
  void _resetLevelWindow() {
    _levelPeak = 0;
    _levelEmitAt = null;
  }

  /// 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',
    );
  }
}