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 轮询服务已重启");
}
}