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 音频。
///
/// 实现方式是给 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;

  /// iOS 专用的原生 PCM 播放通道，见 [_iosNative] 的说明
  static const MethodChannel _iosChannel =
      MethodChannel('azure_speech/pcm_player');

  /// iOS 上不能用 just_audio 播这条流（原因见 [_iosNative]）
  bool get _iosNative => !kIsWeb && Platform.isIOS;

  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 => _iosNative ? _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();

  /// 正在进行中的 [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 (_iosNative) {
        _iosChannel.setMethodCallHandler(_onIosCall);
        final ok = await _iosChannel
            .invokeMethod<bool>('start', {'sampleRate': sampleRate});
        if (ok != true) {
          Logger.e(_tag, 'iOS 原生播放器启动失败');
          return;
        }
        _active = true;
        if (_pending.length > 0) {
          await _iosChannel.invokeMethod('feed', {'data': _pending.takeBytes()});
        }
        return;
      }

      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 (_iosNative) {
      if (_active) {
        _iosChannel.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 (_iosNative) {
      // 起播还在路上就先等它：不等的话短回复会被判成「没有音频」
      if (_starting) {
        try {
          await _startFuture;
        } catch (_) {}
      }
      if (!_active) {
        // 这一轮一个字节都没来过：原生侧没东西可播，直接报完成
        _emitCompleted();
        return;
      }
      await _iosChannel.invokeMethod('end');
      return;
    }
    final src = _source;
    if (src == null || !src.hasData) {
      await _teardown();
      _emitCompleted();
      return;
    }
    await src.close();
  }

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

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

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

  Future<void> _teardown() async {
    _active = false;
    _pending.clear();
    if (_iosNative) {
      try {
        await _iosChannel.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();
    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',
    );
  }
}