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/repositories/mower_realtime_repository.dart'; import 'package:maibu_satabot_v2/core/network/mqtt/domain/entities/task_arrive_entity.dart'; import 'package:maibu_satabot_v2/core/network/mqtt/domain/entities/task_status_entity.dart'; import '../../../../core/logging/i_logger_service.dart'; import '../../../../core/logging/log_time.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(); // 🔥 MQTT机器状态数据源(替代TCP 0x02) MowerRealtimeRepository get _mowerRealtimeRepo => GetIt.I(); class DeviceStatusBloc extends Bloc { final NetMessageDispatcher _dispatcher; final TcpClient tcpClient; // 🔥 新增:直接访问 TcpClient final ILoggerService _logger = GetIt.I(); // 🔥 保存订阅引用,用于管理生命周期 StreamSubscription? _tcpSubscription; StreamSubscription? _mqttArriveSubscription; StreamSubscription? _mqttStatusSubscription; // 🔥 MQTT机器状态订阅(车辆实时+定位,替代TCP 0x02) StreamSubscription? _mqttVehicleSubscription; StreamSubscription? _mqttLocationSubscription; // 🔥 节流相关:500ms节流控制0x02数据推送频率 Timer? _throttleTimer; static const _throttleDuration = Duration(milliseconds: 500); RunningStatusEntity? _cachedStatus; GPSEntity? _cachedGps; // 🔥 调试计数器:跟踪0x02收包序号,排查断断续续问题 int _packetSeq = 0; // 🔥 当前监听的设备ID String? _currentDeviceId; // 🔥 MQTT机器状态收帧计数(用于链路日志) int _mqttVehicleFrameCount = 0; int _mqttLocationFrameCount = 0; DeviceStatusBloc(this._dispatcher, {TcpClient? client}) : tcpClient = client ?? GetIt.I(), super(DeviceStatusInitial()) { // 🔥 核心改动:直接在构造函数中建立TCP监听,类似RemoteControlCubit // 🔥 已停用:机器状态数据源已改为MQTT(见 startMqttRealtimeListening, // 由用户切换设备时触发)。TCP监听逻辑保留,回滚时取消下行注释即可。 // _initDirectTcpListener(); // 🔥 初始化MQTT到达点监听 _initMqttArriveListener(); // 🔥 初始化MQTT任务状态监听(接收完成推送) _initMqttStatusListener(); // 保留事件处理(用于手动重置等场景) 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) { final is02 = p.command == 0x02; if (is02) { _packetSeq++; debugPrint('📥 [0x02] #$_packetSeq 收到原始TCP包 | payload长度=${p.payload.length} | ${DateTime.now().toString().substring(11, 19)}'); } return is02; }) .map((p) { try { final result = utf8.decode(p.payload, allowMalformed: true); return result; } catch (e) { debugPrint('❌ [0x02] #$_packetSeq 解码失败: $e, payload前20字节=${p.payload.take(20).toList()}'); return ''; } }) .listen( (message) { if (message.isEmpty) { debugPrint('⚠️ [0x02] #$_packetSeq 解码后为空,跳过'); return; } try { // 🔥 直接解析,不经过事件转换 final fields = message.trim().split(','); if (fields.length < 18) { debugPrint('⚠️ [0x02] #$_packetSeq 字段不足:${fields.length},期望≥18, 原始数据前100字符=${message.substring(0, message.length > 100 ? 100 : message.length)}'); // 🔥 错误不节流,立即emit以便UI显示错误 if (!isClosed) { emit(DeviceStatusError('字段不足,期望≥18,实际:${fields.length}')); } return; } final status = RunningStatusEntity.fromFields(fields); final gps = GPSEntity(status.latitude, status.longitude); debugPrint('✅ [0x02] #$_packetSeq 解析成功 | 字段数=${fields.length} | 电压=${status.voltage}V 电量=${status.battery}% 控制模式=${status.controlMode} | 节流等待${_throttleDuration.inMilliseconds}ms'); // 🔥 缓存最新数据用于节流发射 _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); // 🔥 节流:如果定时器已在运行,只更新缓存不重置;否则启动新的500ms节流周期 if (_throttleTimer == null || !_throttleTimer!.isActive) { _throttleTimer = Timer(_throttleDuration, () { _emitCachedStatus(); }); } } catch (e, stack) { debugPrint('❌ [0x02] #$_packetSeq 解析异常:$e'); // _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机器状态监听(替代TCP 0x02数据源) /// /// [deviceSn] 与TCP时代 switchDevice / connectBySwitch 的设备号获取方式一致 /// (即 DeviceEntity.deviceName),由用户点击设备列表切换设备时调用。 /// /// 第一阶段:订阅并打印原始报文;确认格式后在第二阶段解析为 /// RunningStatusEntity + GPSEntity,复用现有500ms节流逻辑。 void startMqttRealtimeListening(String deviceSn) { if (deviceSn.isEmpty) { debugPrint('${LogTime.now()} ⚠️ [DeviceStatusBloc] MQTT订阅跳过:设备SN为空'); return; } debugPrint('${LogTime.now()} 🚀 [DeviceStatusBloc] 启动MQTT机器状态监听: $deviceSn'); // 按SN订阅(数据源内部自动退旧订新、同SN去重) _mowerRealtimeRepo.startListening(deviceSn: deviceSn).then((result) { result.fold( (failure) => debugPrint( '${LogTime.now()} ❌ [DeviceStatusBloc] MQTT机器状态订阅失败: ${failure.message}', ), (_) => debugPrint( '${LogTime.now()} ✅ [DeviceStatusBloc] MQTT机器状态订阅完成: $deviceSn', ), ); }); // 切换设备时重置收帧计数,便于观察新设备的首帧与频率 _mqttVehicleFrameCount = 0; _mqttLocationFrameCount = 0; // 流监听只建立一次,第一阶段仅打印解析结果验证链路 _mqttVehicleSubscription ??= _mowerRealtimeRepo.vehicleStream.listen(( message, ) { _mqttVehicleFrameCount++; debugPrint( '${LogTime.now()} 📊 [MQTT-机器状态] 车辆实时第$_mqttVehicleFrameCount帧 type=${message.type} 数据点=${message.data.length}', ); // TODO 第二阶段:将数据点映射到 RunningStatusEntity 字段,复用500ms节流emit }); _mqttLocationSubscription ??= _mowerRealtimeRepo.locationStream.listen(( location, ) { _mqttLocationFrameCount++; debugPrint( '${LogTime.now()} 📍 [MQTT-location] 定位第$_mqttLocationFrameCount帧 lat=${location.latitude}, lng=${location.longitude}, heading=${location.heading}', ); // 🔥 核心:用 MQTT location 数据更新缓存并节流 emit if (location.isValid) { // 更新 _cachedGps _cachedGps = GPSEntity(location.latitude!, location.longitude!); // 更新 _cachedStatus 中的航向角和坐标(复用已有缓存或新建空壳) final base = _cachedStatus ?? RunningStatusEntity(); _cachedStatus = base.copyWith( latitude: location.latitude!, longitude: location.longitude!, yaw: location.heading ?? base.yaw, ); // 500ms节流:与 TCP 0x02 时代逻辑一致 if (_throttleTimer == null || !_throttleTimer!.isActive) { _throttleTimer = Timer(_throttleDuration, () { _emitCachedStatus(); }); } } else { debugPrint( '${LogTime.now()} ⚠️ [MQTT-location] 坐标无效,跳过 - lat=${location.latitude}, lng=${location.longitude}', ); } }); debugPrint('${LogTime.now()} ✅ [DeviceStatusBloc] MQTT机器状态监听已启动: $deviceSn'); _logger.log('✅ [DeviceStatusBloc] MQTT机器状态监听已启动: $deviceSn'); } // 🔥 初始化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到达点监听器已建立完成'); } // 🔥 初始化MQTT任务状态监听(接收 task/+/status 完成推送) void _initMqttStatusListener() { debugPrint('🔗 [DeviceStatusBloc] 初始化MQTT任务状态监听器'); _logger.log('🔗 [DeviceStatusBloc] 初始化MQTT任务状态监听器'); _mqttStatusSubscription = _taskMessageRepo.taskStatusStream.listen( (TaskStatusEntity status) { debugPrint( '📋 [DeviceStatusBloc] 收到MQTT任务状态: type=${status.type}, deviceId=${status.deviceId}, status=${status.status}', ); _logger.log( '📋 [DeviceStatusBloc] 收到MQTT任务状态: type=${status.type}, deviceId=${status.deviceId}, status=${status.status}', ); // 检查设备ID是否匹配当前监听的设备 if (_currentDeviceId != null && status.deviceId != _currentDeviceId) { debugPrint( '⚠️ [DeviceStatusBloc] 任务状态设备ID不匹配,跳过 - 当前:$_currentDeviceId, 收到:${status.deviceId}', ); return; } // 🔥 状态为 FINISH 表示任务完成 if (status.status == 'FINISH') { debugPrint('🏁 [DeviceStatusBloc] 收到任务完成推送,触发完成流程'); _logger.log('🏁 [DeviceStatusBloc] 收到任务完成推送,触发完成流程'); final devicesCubit = GetIt.I(); devicesCubit.finishWork(); // 🔥 无论缓存是否有效,都必须 emit 触发 BlocBuilder 重建 if (!isClosed) { emit(DeviceStatusUpdated( _cachedStatus ?? RunningStatusEntity(), _cachedGps ?? GPSEntity(0.0, 0.0), )); debugPrint('📤 [DeviceStatusBloc] 已 emit 完成信号,触发 UI 更新'); } Future.delayed(Duration(seconds: 1), () { devicesCubit.resetWorkStatus(); }); } }, 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('📤 [节流发射] 电压=${_cachedStatus!.voltage}V 电量=${_cachedStatus!.battery}% 经纬度=(${_cachedGps!.latitude}, ${_cachedGps!.longitude}) yaw=${_cachedStatus!.yaw} | ${DateTime.now().toString().substring(11, 19)}'); 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(); // 🔥 关键:无论缓存是否有效,都必须 emit 触发 BlocBuilder 重建 if (!isClosed) { emit(DeviceStatusUpdated( _cachedStatus ?? RunningStatusEntity(), _cachedGps ?? GPSEntity(0.0, 0.0), )); debugPrint('📤 [DeviceStatusBloc] 已 emit 停止信号,触发 UI 更新'); } Future.delayed(Duration(seconds: 1), () { devicesCubit.resetWorkStatus(); }); } } @override Future close() { debugPrint('🚫 [DeviceStatusBloc] 页面退出,清理资源'); _logger.log('🚫 [DeviceStatusBloc] 页面退出,清理资源'); // 🔥 清理MQTT订阅 _mqttArriveSubscription?.cancel(); _mqttArriveSubscription = null; _mqttStatusSubscription?.cancel(); _mqttStatusSubscription = null; // 🔥 清理MQTT机器状态订阅 _mqttVehicleSubscription?.cancel(); _mqttVehicleSubscription = null; _mqttLocationSubscription?.cancel(); _mqttLocationSubscription = null; // 🔥 清理节流timer和缓存 _throttleTimer?.cancel(); _throttleTimer = null; _cachedStatus = null; _cachedGps = null; // 🔥 重置设备ID _currentDeviceId = null; return Future.value(); } }