如何在Dart中通过gRPC向Python服务器正确发送摄像头视频流？

问题描述 #

我正在开发一款Flutter应用，需要通过gRPC将摄像头采集的视频流发送至Python服务器，服务器会以字节形式接收该流。我希望在Python端对输入进行解码，以便用于计算机视觉（CV）模型，请问如何在Dart中正确实现该视频流发送？

当前实现代码：

Future<void> streamVideo(CameraImage image) async {
    try {
      var bytes1 = image.planes[0].bytes;
      var stream = Stream.fromIterable([
        VideoFrame()
          ..data = bytes1
          ..width = image.width
          ..height = image.height
          ..format = 1
      ]);
      var response = await stub.streamVideo(stream);
      print('Streaming status: ${response.success}, Message: ${response.message}');
    } catch (e) {
      print('Caught error: $e');
    }
  }

  // Method to cleanly shut down the client
  void shutdown() async {
    await channel.shutdown();
  }
}

当前代码的问题 #

1. 单次推送而非持续流：每次调用streamVideo只创建包含单帧的流，无法实现连续的视频流推送，摄像头的每一帧都需要被持续加入到同一个gRPC流中。
2. 格式信息不明确：仅传递planes[0].bytes（通常是YUV格式的Y分量），缺少完整的色彩通道数据，Python端无法正确解码；format=1没有明确约定格式类型，容易导致解码错误。
3. 资源管理不当：每次创建新流会增加gRPC连接开销，且没有处理流的生命周期。

修正后的实现方案 #

1. 初始化流控制器管理视频帧 #

使用StreamController持续收集摄像头帧并推送给gRPC服务：

import 'dart:typed_data';
import 'package:grpc/grpc.dart';
// 导入你的gRPC生成代码，比如import 'generated/video_service.pbgrpc.dart';

class VideoStreamClient {
  final ClientChannel channel;
  final VideoServiceClient stub;
  final StreamController<VideoFrame> _videoFrameController = StreamController<VideoFrame>();

  VideoStreamClient(this.channel) : stub = VideoServiceClient(channel) {
    // 初始化时启动gRPC流式调用
    _startStreaming();
  }

  Future<void> _startStreaming() async {
    try {
      final response = await stub.streamVideo(_videoFrameController.stream);
      print('Streaming status: ${response.success}, Message: ${response.message}');
    } catch (e) {
      print('Stream error: $e');
    }
  }

2. 处理摄像头帧并推送 #

修改帧发送逻辑，完整处理图像格式并持续推送：

void sendVideoFrame(CameraImage image) {
    try {
      Uint8List frameBytes;
      int formatCode;

      // 根据设备输出的图像格式处理字节
      if (image.format.group == ImageFormatGroup.yuv420) {
        // YUV420格式：合并Y、U、V三个平面的字节
        final yBytes = image.planes[0].bytes;
        final uBytes = image.planes[1].bytes;
        final vBytes = image.planes[2].bytes;
        frameBytes = Uint8List.fromList([...yBytes, ...uBytes, ...vBytes]);
        formatCode = 2; // 和Python端约定：2代表YUV420
      } else if (image.format.group == ImageFormatGroup.rgb888) {
        // RGB888格式：直接取第一个平面的字节
        frameBytes = image.planes[0].bytes;
        formatCode = 1; // 和Python端约定：1代表RGB888
      } else {
        // 其他格式可转成JPEG减少带宽（需先添加image依赖：flutter pub add image）
        // import 'package:image/image.dart' as img;
        // final img.Image? decodedImg = img.decodeImageFromList(image.planes[0].bytes);
        // frameBytes = img.encodeJpg(decodedImg!, quality: 80);
        // formatCode = 3; // 3代表JPEG
        print('Unsupported format, skipping frame');
        return;
      }

      final videoFrame = VideoFrame()
        ..data = frameBytes
        ..width = image.width
        ..height = image.height
        ..format = formatCode;

      if (!_videoFrameController.isClosed) {
        _videoFrameController.add(videoFrame);
      }
    } catch (e) {
      print('Failed to send frame: $e');
    }
  }

3. 完善资源关闭逻辑 #

确保关闭流控制器和gRPC通道，避免内存泄漏：

Future<void> shutdown() async {
    await _videoFrameController.close();
    await channel.shutdown();
  }
}

关键注意事项 #

• 格式约定：必须和Python端明确format字段的枚举含义，比如YUV420、RGB888、JPEG对应的编码值，否则解码会出现色彩失真或无法解析的问题。
• 性能优化：如果网络带宽有限，建议将帧编码为JPEG后再发送，降低数据量。
• 生命周期管理：在Flutter页面销毁时，务必调用shutdown()方法关闭流控制器和gRPC通道，避免资源泄漏。

内容的提问来源于stack exchange，提问作者Lafi Odeh