import 'dart:async'; import 'dart:convert'; import 'package:flutter/cupertino.dart'; import 'package:flutter_bloc/flutter_bloc.dart'; import 'package:get_it/get_it.dart'; import 'package:maibu_satabot_v2/features/devices/domain/entities/running_status_entity.dart'; import 'package:maibu_satabot_v2/features/devices/domain/entities/gps_entity.dart'; import 'package:maibu_satabot_v2/core/network/protocol_decoder.dart'; import 'package:maibu_satabot_v2/core/network/mqtt/domain/repositories/task_message_repository.dart'; import 'package:maibu_satabot_v2/core/network/mqtt/domain/entities/task_arrive_entity.dart'; import '../../../../core/logging/i_logger_service.dart'; import '../../../../core/network/net_message_dispatcher.dart'; import '../../../../core/network/tcp/tcp_client.dart'; import 'device_status_event.dart'; import 'device_status_state.dart'; import 'devices_cubit.dart'; // 定义全局的TaskMessageRepository获取方式 TaskMessageRepository get _taskMessageRepo => GetIt.I(); class DeviceStatusBloc extends Bloc { final NetMessageDispatcher _dispatcher; final TcpClient tcpClient; // 🔥 新增:直接访问 TcpClient final ILoggerService _logger = GetIt.I(); // 🔥 保存订阅引用,用于管理生命周期 StreamSubscription? _tcpSubscription; StreamSubscription? _mqttArriveSubscription; // 🔥 节流相关:500ms节流控制0x02数据推送频率 Timer? _throttleTimer; static const _throttleDuration = Duration(milliseconds: 500); RunningStatusEntity? _cachedStatus; GPSEntity? _cachedGps; // 🔥 当前监听的设备ID String? _currentDeviceId; DeviceStatusBloc(this._dispatcher, {TcpClient? client}) : tcpClient = client ?? GetIt.I(), super(DeviceStatusInitial()) { // 🔥 核心改动:直接在构造函数中建立TCP监听,类似RemoteControlCubit _initDirectTcpListener(); // 🔥 初始化MQTT到达点监听 _initMqttArriveListener(); // 保留事件处理(用于手动重置等场景) on(_handleReset); on(_handleDeviceStatusLoaded); on(_handlePushMessageReceived); on(_handleDeviceTaskArrivePointEvent); } // 🔥 已废弃:重新初始化TCP监听器会导致数据流中断 // DeviceStatusBloc是单例,TCP监听器在构造函数中建立后应始终保持活跃 // 任何页面只需通过 BlocBuilder 或 stream 订阅状态即可 /* void reinitializeListener() { debugPrint('🔄 [DeviceStatusBloc] 重新初始化TCP监听器 - isClosed=$isClosed'); _logger.log('🔄 [DeviceStatusBloc] 重新初始化TCP监听器 - isClosed=$isClosed'); if (isClosed) { debugPrint('❌ [DeviceStatusBloc] BLoC已关闭,无法重新初始化!'); _logger.log('❌ [DeviceStatusBloc] BLoC已关闭,无法重新初始化!'); return; } // ❌ 取消旧的订阅会导致数据流中断! debugPrint('📝 [DeviceStatusBloc] 取消旧订阅'); _tcpSubscription?.cancel(); // ❌ 重置状态会清空UI显示的数据! debugPrint('📝 [DeviceStatusBloc] emit Initial状态'); emit(DeviceStatusInitial()); // 重新建立监听 debugPrint('📝 [DeviceStatusBloc] 调用 _initDirectTcpListener'); _initDirectTcpListener(); } */ // 🔥 新增:直接监听TCP 0x02指令,实时解析并emit状态 void _initDirectTcpListener() { debugPrint('🔗 [DeviceStatusBloc] 初始化直接TCP监听器 - 使用 tcpClient.packetStream'); _logger.log('🔗 [DeviceStatusBloc] 初始化直接TCP监听器'); // 🔥 关键修复:直接监听 tcpClient.packetStream,不经过 dispatcher 的 filtered stream // 这样即使没有其他监听者,TCP流也不会暂停 _tcpSubscription = tcpClient.packetStream .where((p) => p.command == 0x02) .map((p) { try { final result = utf8.decode(p.payload, allowMalformed: true); // debugPrint('✅ [DeviceStatusBloc] 收到0x02数据: $result'); return result; } catch (e) { // debugPrint('❌ [DeviceStatusBloc] 解码失败: $e'); return ''; } }) .listen( (message) { //debugPrint('📩 [DeviceStatusBloc] 直接收到0x02数据,长度:${message.length}'); if (message.isEmpty) { //debugPrint('⚠️ [DeviceStatusBloc] 消息为空,跳过'); return; } try { // 🔥 直接解析,不经过事件转换 final fields = message.trim().split(','); if (fields.length < 18) { //debugPrint('⚠️ [DeviceStatusBloc] 字段不足:${fields.length},期望≥18'); // 🔥 错误不节流,立即emit以便UI显示错误 if (!isClosed) { emit(DeviceStatusError('字段不足,期望≥18,实际:${fields.length}')); } return; } final status = RunningStatusEntity.fromFields(fields); final gps = GPSEntity(status.latitude, status.longitude); // 🔥 缓存最新数据用于节流发射 _cachedStatus = RunningStatusEntity( voltage: status.voltage, leftTargetSpeed: status.leftTargetSpeed, rightTargetSpeed: status.rightTargetSpeed, leftMeasureSpeed: status.leftMeasureSpeed, rightMeasureSpeed: status.rightMeasureSpeed, leftCurrent: status.leftCurrent, rightCurrent: status.rightCurrent, leftMotorTemp: status.leftMotorTemp, rightMotorTemp: status.rightMotorTemp, chipTemp: status.chipTemp, yaw: status.yaw, pitch: status.pitch, roll: status.roll, satelliteCnt: status.satelliteCnt, qual: status.qual, headingStatus: status.headingStatus, latitude: status.latitude, longitude: status.longitude, timestamp: status.timestamp, knifeCuttingSpeed: status.knifeCuttingSpeed, controlMode: status.controlMode, battery: status.battery, workingArea: status.workingArea, obstacleFlag: status.obstacleFlag, ); _cachedGps = GPSEntity(status.latitude, status.longitude); // 🔥 节流:取消之前的timer,重新计时500ms _throttleTimer?.cancel(); _throttleTimer = Timer(_throttleDuration, () { _emitCachedStatus(); }); } catch (e, stack) { //debugPrint('❌ [DeviceStatusBloc] 直接解析异常:$e\n$stack'); // _logger.log('❌ [DeviceStatusBloc] 直接解析异常:$e'); // 🔥 解析错误不节流,立即emit以便UI显示错误 if (!isClosed) { emit(DeviceStatusError('解析失败:$e')); } } }, onDone: () => debugPrint('⚠️ [DeviceStatusBloc] TCP流已结束(onDone)'), onError: (e) => debugPrint('❌ [DeviceStatusBloc] TCP流错误:$e'), ); debugPrint('✅ [DeviceStatusBloc] 直接TCP监听器已建立完成'); _logger.log('✅ [DeviceStatusBloc] 直接TCP监听器已建立完成'); } // 🔥 初始化MQTT到达点监听 void _initMqttArriveListener() { debugPrint('🔗 [DeviceStatusBloc] 初始化MQTT到达点监听器'); _logger.log('🔗 [DeviceStatusBloc] 初始化MQTT到达点监听器'); _mqttArriveSubscription = _taskMessageRepo.taskArriveStream.listen( (TaskArriveEntity arrive) { debugPrint( '📍 [DeviceStatusBloc] 收到MQTT到达点消息: type=${arrive.type}, deviceId=${arrive.deviceId}', ); _logger.log( '📍 [DeviceStatusBloc] 收到MQTT到达点消息: type=${arrive.type}, deviceId=${arrive.deviceId}', ); // 过滤出路径规划到达点消息 if (arrive.type == 'device_task_arrive_point') { // 检查设备ID是否匹配当前监听的设备 if (_currentDeviceId != null && arrive.deviceId != _currentDeviceId) { debugPrint( '⚠️ [DeviceStatusBloc] 设备ID不匹配,跳过 - 当前:$_currentDeviceId, 收到:${arrive.deviceId}', ); return; } // 提取坐标(支持空entity场景) double? lat = arrive.entity?.lat; double? lng = arrive.entity?.lng; // 发送事件到Bloc add( DeviceTaskArrivePointEvent( deviceId: arrive.deviceId, taskId: arrive.taskId, lat: lat, lng: lng, ), ); } }, onError: (e) { debugPrint('❌ [DeviceStatusBloc] MQTT到达点监听错误: $e'); _logger.log('❌ [DeviceStatusBloc] MQTT到达点监听错误: $e'); }, ); debugPrint('✅ [DeviceStatusBloc] MQTT到达点监听器已建立完成'); _logger.log('✅ [DeviceStatusBloc] MQTT到达点监听器已建立完成'); } // 🔥 设置当前监听的设备ID(用于过滤MQTT消息) void setListeningDeviceId(String deviceId) { debugPrint('📱 [DeviceStatusBloc] 设置监听设备ID: $deviceId'); _logger.log('📱 [DeviceStatusBloc] 设置监听设备ID: $deviceId'); _currentDeviceId = deviceId; } // 🔥 节流发射:500ms到期后发射缓存的最新数据 void _emitCachedStatus() { if (_cachedStatus != null && _cachedGps != null && !isClosed) { // debugPrint('📤 [DeviceStatusBloc] 🔥节流发射 - 电压:${_cachedStatus!.voltage}, 电量:${_cachedStatus!.battery}'); emit(DeviceStatusUpdated(_cachedStatus!, _cachedGps!)); } } // 🔥 重置时仅清空状态 Future _handleReset( DeviceStatusReset event, Emitter emit, ) async { debugPrint('🔄 [DeviceStatusBloc] 收到重置事件:清空状态 - 当前状态:${state.runtimeType}'); _logger.log( '🔄 [DeviceStatusBloc] 收到重置事件:清空状态 - 当前状态:${state.runtimeType}', ); // 只 emit 初始状态,让 UI 清除旧设备的数据 emit(DeviceStatusInitial()); debugPrint('⚠️ [DeviceStatusBloc] 已emit DeviceStatusInitial'); } Future _handleDeviceStatusLoaded( DeviceStatusLoaded event, Emitter emit, ) async { try { // debugPrint('🔍 开始解析数据:${event.jsonString}'); _logger.log('🔍 开始解析数据:${event.jsonString}'); final fields = event.jsonString.trim().split(','); if (fields.length < 18) { debugPrint('⚠️ 字段不足:${fields.length}'); emit(DeviceStatusError('字段不足,期望 ≥18,实际:${fields.length}')); return; } final status = RunningStatusEntity.fromFields(fields); final gps = GPSEntity(status.latitude, status.longitude); //debugPrint('✅ 解析成功,更新状态:Lat=${gps.latitude}, Lng=${gps.longitude}'); _logger.log('✅ 解析成功,更新状态:Lat=${gps.latitude}, Lng=${gps.longitude}'); emit(DeviceStatusUpdated(status, gps)); } catch (e, stack) { //debugPrint('❌ 解析异常:$e\n$stack'); _logger.log('❌ 解析异常:$e\n$stack'); emit(DeviceStatusError('解析失败:$e')); } } Future _handlePushMessageReceived( PushMessageReceived event, Emitter emit, ) async { try { final eventStr = event.jsonData['event'] ?? ''; final deviceId = event.jsonData['deviceId'] ?? '未知'; // debugPrint('收到推送事件:$eventStr, 设备:$deviceId'); _logger.log('收到推送事件:$eventStr, 设备:$deviceId'); // 这里可以根据需要 emit 新状态 } catch (e) { emit(DeviceStatusError('解析推送消息失败:$e')); } } // 🔥 处理MQTT到达点事件 Future _handleDeviceTaskArrivePointEvent( DeviceTaskArrivePointEvent event, Emitter emit, ) async { debugPrint( '📍 [DeviceStatusBloc] 处理到达点事件: deviceId=${event.deviceId}, taskId=${event.taskId}, lat=${event.lat}, lng=${event.lng}', ); _logger.log( '📍 [DeviceStatusBloc] 处理到达点事件: deviceId=${event.deviceId}, taskId=${event.taskId}, lat=${event.lat}, lng=${event.lng}', ); if (event.lat != null && event.lng != null && event.lat != 0.0) { // ✅ 收到有效到达点 - 更新位置状态 final gps = GPSEntity(event.lat!, event.lng!); // 发送到DevicesCubit更新到达位置(触发地图轨迹更新) final devicesCubit = GetIt.I(); devicesCubit.setArrivedLocation(event.lat!, event.lng!); // 同时更新DeviceStatusBloc的状态 emit(DeviceStatusUpdated(_cachedStatus ?? RunningStatusEntity(), gps)); } else { // ❌ entity为空或坐标无效 - 表示任务停止 debugPrint('🛑 [DeviceStatusBloc] 收到任务停止信号'); _logger.log('🛑 [DeviceStatusBloc] 收到任务停止信号'); final devicesCubit = GetIt.I(); devicesCubit.finishWork(); Future.delayed(Duration(seconds: 1), () { devicesCubit.resetWorkStatus(); }); } } @override Future close() { debugPrint('🚫 [DeviceStatusBloc] 页面退出,清理资源'); _logger.log('🚫 [DeviceStatusBloc] 页面退出,清理资源'); // 🔥 清理MQTT订阅 _mqttArriveSubscription?.cancel(); _mqttArriveSubscription = null; // 🔥 清理节流timer和缓存 _throttleTimer?.cancel(); _throttleTimer = null; _cachedStatus = null; _cachedGps = null; // 🔥 重置设备ID _currentDeviceId = null; return Future.value(); } }