using CommunityToolkit.Mvvm.ComponentModel; using Serilog; using YZWater.Core.Models; namespace YZWater.Core.Services; /// /// PLC 轮询服务 - 负责实时数据采集、自动重连、数据分发 /// public partial class PlcPollingService : ObservableObject, IDisposable, IServiceHealthCheck, IRestartableService { private static PlcPollingService? _instance; private static readonly object _instanceLock = new(); public static PlcPollingService Instance { get { lock (_instanceLock) { return _instance ??= new PlcPollingService(); } } } /// /// 重置单例(用于登出后重新登录) /// public static void ResetInstance() { lock (_instanceLock) { var old = _instance; _instance = null; // 先置 null,确保并发访问 Instance 时创建新实例 old?.Dispose(); } } private readonly PlcDataModel _data = new(); private CancellationTokenSource? _cts; private Task? _pollingTask; private bool _disposed; // PLC 读取并发限制(防止同时发起过多请求导致 PLC 过载) private static readonly SemaphoreSlim _plcReadSemaphore = new(4, 4); // ─── 配置 ─── public int PollingIntervalMs { get; set; } = 1000; public float DeadbandPercent { get; set; } = 0.5f; public int MaxReconnectAttempts { get; set; } = 10; public int ReconnectBaseDelayMs { get; set; } = 1000; // ─── 数据变更事件 ─── public event Action? DataUpdated; public event Action? ConnectionStateChanged; public event Action? ErrorOccurred; /// /// 当前数据模型 /// public PlcDataModel Data => _data; private PlcPollingService() { } /// /// 启动轮询 /// public void Start() { if (_pollingTask != null && !_pollingTask.IsCompleted) { Log.Warning("轮询已在运行中"); return; } _cts = new CancellationTokenSource(); _pollingTask = Task.Run(() => PollingLoopAsync(_cts.Token)); Log.Information("PLC 轮询服务已启动"); } /// /// 停止轮询 /// public void Stop() { Log.Information("PlcPollingService 正在停止... _cts={HasCts}, _pollingTask={HasTask}", _cts != null, _pollingTask != null); _cts?.Cancel(); if (_pollingTask != null) { var completed = _pollingTask.Wait(TimeSpan.FromSeconds(5)); if (!completed) { Log.Warning("PlcPollingService 轮询任务在5秒内未完成,可能仍在运行"); } else { Log.Information("PlcPollingService 轮询任务已完成"); } } _cts?.Dispose(); _cts = null; _pollingTask = null; Log.Information("PlcPollingService 已停止"); } /// /// 轮询主循环 /// private async Task PollingLoopAsync(CancellationToken ct) { var reconnectAttempts = 0; var isInDegradedMode = false; // 降级模式标志 const int degradedModeRetryIntervalMs = 30000; // 降级模式下 30 秒重试一次 while (!ct.IsCancellationRequested) { try { if (MockPlcService.IsEnabled) { // Mock 模式:直接生成模拟数据 MockPlcService.UpdateData(_data); } else { // 真实模式:确保连接 if (!PlcService.IsConnected) { var connected = await TryConnectWithRetryAsync(reconnectAttempts, ct); if (!connected) { reconnectAttempts++; // 进入或保持在降级模式 if (reconnectAttempts >= MaxReconnectAttempts) { if (!isInDegradedMode) { // 首次进入降级模式 isInDegradedMode = true; var errorMsg = $"PLC 连接失败,已重试 {MaxReconnectAttempts} 次,进入降级模式(每 30 秒重试)"; Log.Error(errorMsg); ErrorOccurred?.Invoke(errorMsg); // 发送 PLC 断开通知 _ = NotificationService.Instance.SendPlcConnectionNotificationAsync(false); } else { // 已在降级模式,定期记录日志 if (reconnectAttempts % 10 == 0) // 每 10 次(5 分钟)记录一次 { Log.Warning($"PLC 连接仍在尝试中,累计重试 {reconnectAttempts} 次"); } } // 降级模式:使用更长的重试间隔 await Task.Delay(degradedModeRetryIntervalMs, ct); continue; } // 正常模式:指数退避 var delayMs = ReconnectBaseDelayMs * Math.Min(reconnectAttempts, 5); await Task.Delay(delayMs, ct); continue; } // 连接成功 if (isInDegradedMode) { // 从降级模式恢复 Log.Information($"PLC 连接恢复,退出降级模式(累计重试 {reconnectAttempts} 次)"); ErrorOccurred?.Invoke($"PLC 连接已恢复"); isInDegradedMode = false; // 发送 PLC 恢复通知 _ = NotificationService.Instance.SendPlcConnectionNotificationAsync(true); } reconnectAttempts = 0; } // 读取数据 await ReadAllDataAsync(ct); } // 数据更新事件 DataUpdated?.Invoke(_data); // 等待下一次轮询(降级模式下使用更长间隔) var nextDelay = isInDegradedMode ? degradedModeRetryIntervalMs : PollingIntervalMs; await Task.Delay(nextDelay, ct); } catch (OperationCanceledException) { break; } catch (Exception ex) { Log.Error(ex, "轮询异常"); ErrorOccurred?.Invoke($"轮询异常: {ex.Message}"); // 连接断开,尝试重连 if (!PlcService.IsConnected) { _data.IsPlcConnected = false; ConnectionStateChanged?.Invoke(false); } await Task.Delay(PollingIntervalMs * 2, ct); } } } /// /// 尝试连接(带重试) /// private async Task TryConnectWithRetryAsync(int attempt, CancellationToken ct) { for (int i = 0; i <= attempt && i < 3; i++) { if (ct.IsCancellationRequested) return false; var connected = await PlcService.ConnectAsync(); if (connected) { _data.IsPlcConnected = true; ConnectionStateChanged?.Invoke(true); return true; } if (i < attempt) await Task.Delay(ReconnectBaseDelayMs * (i + 1), ct); } return false; } /// /// 批量读取所有 PLC 数据 /// 根据是否加载了项目档案选择读取模式 /// private async Task ReadAllDataAsync(CancellationToken ct) { if (ProjectService.Current != null) await ReadAllDataFromProfile(ct); else await ReadAllDataDefault(ct); } /// /// 默认模式:硬编码地址批量读取(向后兼容) /// private async Task ReadAllDataDefault(CancellationToken ct) { if (ct.IsCancellationRequested) return; // 并行发起所有批量读取(使用 PlcConfig 中的地址常量) var tankTask = PlcService.ReadBytesAsync(PlcConfig.Tank1Level, 16); // 4 floats: 水箱液位 var flowTask = PlcService.ReadBytesAsync(PlcConfig.InflowRate, 8); // 2 floats: 流量 var freqTask = PlcService.ReadBytesAsync(PlcConfig.Pump1Freq, 20); // 5 floats: 泵频率 var qTask = PlcService.ReadBytesAsync("Q0", 1); // 1 byte: 泵/风机运行状态 var iTask = PlcService.ReadBytesAsync("I0", 1); // 1 byte: 泵/风机故障状态 var mTask = PlcService.ReadBytesAsync("M0", 1); // 1 byte: 系统模式 await Task.WhenAll(tankTask, flowTask, freqTask, qTask, iTask, mTask); if (ct.IsCancellationRequested) return; var tankData = tankTask.Result; var flowData = flowTask.Result; var freqData = freqTask.Result; var qData = qTask.Result; var iData = iTask.Result; var mData = mTask.Result; // 任一批量读取失败则跳过本周期(全部 6 个都检查) if (tankData == null || flowData == null || freqData == null || qData == null || iData == null || mData == null) { _data.IsDataValid = false; return; } // 解析水箱液位 (VD100-VD112, 4 floats) var tank1 = PlcService.ParseFloat(tankData, 0); var tank2 = PlcService.ParseFloat(tankData, 4); var tank3 = PlcService.ParseFloat(tankData, 8); var tank4 = PlcService.ParseFloat(tankData, 12); // 解析流量 (VD200-VD204, 2 floats) var inflow = PlcService.ParseFloat(flowData, 0); var outflow = PlcService.ParseFloat(flowData, 4); // 解析泵频率 (VD300-VD316, 5 floats) var f1 = PlcService.ParseFloat(freqData, 0); var f2 = PlcService.ParseFloat(freqData, 4); var f3 = PlcService.ParseFloat(freqData, 8); var f4 = PlcService.ParseFloat(freqData, 12); var f5 = PlcService.ParseFloat(freqData, 16); // 解析 Q0 字节: bit 0-4 = 泵运行, bit 5-6 = 风机运行 var p1 = PlcService.ParseBit(qData, 0, 0); var p2 = PlcService.ParseBit(qData, 0, 1); var p3 = PlcService.ParseBit(qData, 0, 2); var p4 = PlcService.ParseBit(qData, 0, 3); var p5 = PlcService.ParseBit(qData, 0, 4); var fan1 = PlcService.ParseBit(qData, 0, 5); var fan2 = PlcService.ParseBit(qData, 0, 6); // 解析 I0 字节: bit 0-4 = 泵故障, bit 5-6 = 风机故障 var pf1 = PlcService.ParseBit(iData, 0, 0); var pf2 = PlcService.ParseBit(iData, 0, 1); var pf3 = PlcService.ParseBit(iData, 0, 2); var pf4 = PlcService.ParseBit(iData, 0, 3); var pf5 = PlcService.ParseBit(iData, 0, 4); var fanf1 = PlcService.ParseBit(iData, 0, 5); var fanf2 = PlcService.ParseBit(iData, 0, 6); // 解析 M0 字节: bit 0 = 系统模式 var autoMode = PlcService.ParseBit(mData, 0, 0); // 应用死区过滤后更新数据 ApplyWithDeadband(_data.Tank1Level, tank1, v => _data.Tank1Level = v); ApplyWithDeadband(_data.Tank2Level, tank2, v => _data.Tank2Level = v); ApplyWithDeadband(_data.Tank3Level, tank3, v => _data.Tank3Level = v); ApplyWithDeadband(_data.Tank4Level, tank4, v => _data.Tank4Level = v); ApplyWithDeadband(_data.InflowRate, inflow, v => _data.InflowRate = v); ApplyWithDeadband(_data.OutflowRate, outflow, v => _data.OutflowRate = v); _data.Pump1Running = p1; _data.Pump2Running = p2; _data.Pump3Running = p3; _data.Pump4Running = p4; _data.Pump5Running = p5; _data.Pump1Fault = pf1; _data.Pump2Fault = pf2; _data.Pump3Fault = pf3; _data.Pump4Fault = pf4; _data.Pump5Fault = pf5; _data.Pump1Freq = f1; _data.Pump2Freq = f2; _data.Pump3Freq = f3; _data.Pump4Freq = f4; _data.Pump5Freq = f5; _data.Fan1Running = fan1; _data.Fan2Running = fan2; _data.Fan1Fault = fanf1; _data.Fan2Fault = fanf2; _data.IsAutoMode = autoMode; _data.LastReadTime = DateTime.Now; _data.IsDataValid = true; _data.IsPlcConnected = PlcService.IsConnected; _data.FlowDelta = _data.InflowRate - _data.OutflowRate; // 触发计算属性更新 OnPropertyChanged(nameof(PlcDataModel.FlowDelta)); } /// /// 项目档案模式:根据 ProjectProfile 动态读取 PLC 数据 /// private async Task ReadAllDataFromProfile(CancellationToken ct) { if (ct.IsCancellationRequested) return; var profile = ProjectService.Current; if (profile == null) return; // 收集所有需要读取的模拟量地址 (REAL/float) var analogReads = new List<(string key, string addr)>(); foreach (var pump in profile.Pumps) { if (!string.IsNullOrEmpty(pump.CurrentAddress)) analogReads.Add(($"{pump.Name}.Current", pump.CurrentAddress)); if (!string.IsNullOrEmpty(pump.DailyHoursAddress)) analogReads.Add(($"{pump.Name}.DailyHours", pump.DailyHoursAddress)); if (!string.IsNullOrEmpty(pump.DailyMinutesAddress)) analogReads.Add(($"{pump.Name}.DailyMinutes", pump.DailyMinutesAddress)); } foreach (var tank in profile.Tanks) { if (!string.IsNullOrEmpty(tank.LevelAddress)) analogReads.Add(($"{tank.Name}.Level", tank.LevelAddress)); if (!string.IsNullOrEmpty(tank.StartPump1LevelAddress)) analogReads.Add(($"{tank.Name}.StartPump1Level", tank.StartPump1LevelAddress)); if (!string.IsNullOrEmpty(tank.StartPump2LevelAddress)) analogReads.Add(($"{tank.Name}.StartPump2Level", tank.StartPump2LevelAddress)); if (!string.IsNullOrEmpty(tank.StartPump3LevelAddress)) analogReads.Add(($"{tank.Name}.StartPump3Level", tank.StartPump3LevelAddress)); if (!string.IsNullOrEmpty(tank.StopLevelAddress)) analogReads.Add(($"{tank.Name}.StopLevel", tank.StopLevelAddress)); if (!string.IsNullOrEmpty(tank.ProtectionLevelAddress)) analogReads.Add(($"{tank.Name}.ProtectionLevel", tank.ProtectionLevelAddress)); } foreach (var gauge in profile.Gauges) { if (!string.IsNullOrEmpty(gauge.ValueAddress)) analogReads.Add(($"{gauge.Name}.Value", gauge.ValueAddress)); if (!string.IsNullOrEmpty(gauge.HighSetAddress)) analogReads.Add(($"{gauge.Name}.HighSet", gauge.HighSetAddress)); if (!string.IsNullOrEmpty(gauge.LowSetAddress)) analogReads.Add(($"{gauge.Name}.LowSet", gauge.LowSetAddress)); } // 收集传感器读取(带数据类型) var sensorReads = new List<(string key, string addr, string dataType)>(); foreach (var sensor in profile.Sensors) { if (string.IsNullOrEmpty(sensor.Address)) continue; sensorReads.Add(($"{sensor.Name}.Value", sensor.Address, (sensor.DataType ?? "REAL").ToUpper())); } // 按数据类型分类处理 foreach (var sensor in sensorReads) { switch (sensor.dataType) { case "REAL": // 4字节浮点数 analogReads.Add((sensor.key, sensor.addr)); break; case "INT": case "WORD": case "DWORD": case "DINT": // 整数类型:读取字节后手动转换 analogReads.Add(($"{sensor.key}#{sensor.dataType}", sensor.addr)); break; default: // 默认按 REAL 处理 analogReads.Add((sensor.key, sensor.addr)); break; } } // 收集所有需要读取的数字量地址 (bool/bit) var digitalReads = new List<(string key, string addr)>(); foreach (var pump in profile.Pumps) { if (!string.IsNullOrEmpty(pump.RunAddress)) digitalReads.Add(($"{pump.Name}.Run", pump.RunAddress)); if (!string.IsNullOrEmpty(pump.FaultAddress)) digitalReads.Add(($"{pump.Name}.Fault", pump.FaultAddress)); if (!string.IsNullOrEmpty(pump.AutoStateAddress)) digitalReads.Add(($"{pump.Name}.AutoMode", pump.AutoStateAddress)); if (!string.IsNullOrEmpty(pump.ModeSelectAddress)) digitalReads.Add(($"{pump.Name}.ModeSelect", pump.ModeSelectAddress)); } // 通用数字量输入(如变频器故障) foreach (var di in profile.DigitalInputs) { if (!string.IsNullOrEmpty(di.Address)) digitalReads.Add(($"DI.{di.Name}", di.Address)); } // 并行发起所有读取(使用信号量限制并发数,避免 PLC 过载) // 性能优化:S7AddressParser 已实现地址解析和分组算法 // 对于连续地址(如 VD500, VD504, VD508),使用批量读取优化 // 立即物化 Task 数组,避免 IEnumerable 双重枚举导致 PLC 读取量翻倍 // 尝试批量读取优化 var batchResults = await TryBatchReadAsync(analogReads, ct); var analogTasks = analogReads.Select(async r => { await _plcReadSemaphore.WaitAsync(ct); try { float value; var key = r.key; // 优先使用批量读取结果 if (batchResults != null && batchResults.TryGetValue(key, out var batchValue)) { return (key, Value: batchValue); } // 检查是否为整数类型(key 格式:name#DATATYPE) if (key.Contains('#')) { var parts = key.Split('#'); key = parts[0]; var dataType = parts[1]; // 读取字节后手动转换 var bytes = await PlcService.ReadBytesAsync(r.addr, dataType is "INT" or "WORD" ? (ushort)2 : (ushort)4); if (bytes == null || bytes.Length == 0) { return (key, Value: 0f); } value = dataType switch { "INT" => (float)BitConverter.ToInt16(bytes, 0), "WORD" => (float)BitConverter.ToUInt16(bytes, 0), "DINT" => (float)BitConverter.ToInt32(bytes, 0), "DWORD" => (float)BitConverter.ToUInt32(bytes, 0), _ => PlcService.ParseFloat(bytes, 0) }; } else { // REAL 类型直接读浮点数 value = await PlcService.ReadFloatAsync(r.addr); } return (key, Value: value); } finally { _plcReadSemaphore.Release(); } }).ToArray(); var digitalTasks = digitalReads.Select(async r => { await _plcReadSemaphore.WaitAsync(ct); try { var value = await PlcService.ReadBoolAsync(r.addr); return (r.key, Value: value); } finally { _plcReadSemaphore.Release(); } }).ToArray(); var allTasks = analogTasks.Cast().Concat(digitalTasks.Cast()).ToArray(); if (allTasks.Length > 0) await Task.WhenAll(allTasks); if (ct.IsCancellationRequested) return; // 更新模拟量寄存器 foreach (var task in analogTasks) { var (key, value) = await task; if (_data.AnalogRegisters.TryGetValue(key, out var currentValue)) { ApplyWithDeadband(currentValue, value, v => _data.AnalogRegisters[key] = v); } } // 更新数字量寄存器 foreach (var task in digitalTasks) { var (key, value) = await task; _data.DigitalRegisters[key] = value; } _data.LastReadTime = DateTime.Now; _data.IsDataValid = true; _data.IsPlcConnected = PlcService.IsConnected; } /// /// 尝试批量读取连续地址(优化性能) /// private async Task?> TryBatchReadAsync( List<(string key, string addr)> analogReads, CancellationToken ct) { // 只对 REAL 类型的地址进行批量读取(排除带 #DATATYPE 后缀的整数类型) var realReads = analogReads .Where(r => !r.key.Contains('#')) .Select(r => (r.key, r.addr)) .ToList(); if (realReads.Count < 3) return null; // 地址太少,不值得批量 try { // 分组连续地址 var groups = S7AddressParser.GroupConsecutiveAddresses(realReads); // 只对包含多个地址的分组进行批量读取 var batchGroups = groups.Where(g => g.Items.Count >= 2).ToList(); if (batchGroups.Count == 0) return null; var results = new Dictionary(); foreach (var group in batchGroups) { if (ct.IsCancellationRequested) return results; // 构建批量读取地址 var prefix = S7AddressParser.GetAddressPrefix(group.Area, group.TotalBytes switch { >= 4 => 4, >= 2 => 2, _ => 1 }); var batchAddr = $"{prefix}{group.StartByte}"; await _plcReadSemaphore.WaitAsync(ct); try { var bytes = await PlcService.ReadBytesAsync(batchAddr, (ushort)group.TotalBytes); if (bytes == null) continue; // 解析每个子项 foreach (var (key, offset) in group.Items) { if (offset + 4 <= bytes.Length) { results[key] = PlcService.ParseFloat(bytes, offset); } } } finally { _plcReadSemaphore.Release(); } } if (results.Count > 0) { Log.Debug("批量读取优化: {Count} 个地址合并为 {Groups} 次读取", results.Count, batchGroups.Count); } return results.Count > 0 ? results : null; } catch (Exception ex) { Log.Warning(ex, "批量读取失败,回退到单独读取"); return null; } } /// /// 死区过滤:只有变化超过阈值才更新 /// private void ApplyWithDeadband(float currentValue, float newValue, Action setter) { if (Math.Abs(currentValue) < 0.001f && Math.Abs(newValue) < 0.001f) return; // 都是零值,跳过 var threshold = Math.Abs(currentValue) * DeadbandPercent / 100.0f; if (Math.Abs(newValue - currentValue) > threshold || currentValue == 0) { setter(newValue); } } /// /// 立即读取一次(手动刷新) /// public async Task RefreshNowAsync() { if (PlcService.IsConnected) { await ReadAllDataAsync(CancellationToken.None); DataUpdated?.Invoke(_data); } } /// /// 触发数据更新事件(供客户端模式使用) /// public void RaiseDataUpdated(PlcDataModel data) { DataUpdated?.Invoke(data); } public void Dispose() { if (_disposed) return; _disposed = true; Log.Information("PlcPollingService 正在释放..."); Stop(); Log.Information("PlcPollingService 已释放"); } /// /// 健康检查(看门狗调用) /// public Task<(bool IsHealthy, string? Reason)> CheckHealthAsync() { // 检查轮询任务是否还在运行 if (_pollingTask == null || _pollingTask.IsCompleted) { return Task.FromResult((false, "轮询任务未运行")); } // 检查数据更新时间(超过 60 秒无数据视为不健康) if (_data.LastReadTime > DateTime.MinValue) { var elapsed = (DateTime.Now - _data.LastReadTime).TotalSeconds; if (elapsed > 60) { return Task.FromResult((false, $"数据更新超时: {elapsed:F0} 秒")); } } // 检查取消令牌是否已释放 if (_cts == null) { return Task.FromResult((false, "轮询服务已停止")); } return Task.FromResult((true, "运行正常")); } /// /// 重启服务(看门狗调用) /// public async Task RestartAsync() { Log.Warning("PLC 轮询服务正在重启..."); Stop(); await Task.Delay(1000); // 等待资源释放 Start(); Log.Information("PLC 轮询服务已重启"); } }