Files
1257B_DB/IOT.Services/Service1.cs
T

1180 lines
45 KiB
C#
Raw Normal View History

2026-08-06 09:23:30 +08:00
using JinYuan.DataConvertLib;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using PLCCommunication.MQTT;
using System;
using System.Collections.Generic;
using System.Configuration;
using System.Data;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.ServiceProcess;
using System.Threading;
using System.Threading.Tasks;
using System.Xml.Linq;
namespace WindowsService1
{
public partial class Service1 : ServiceBase
{
private readonly List<Task> _tasks = new List<Task>();
private readonly CancellationTokenSource _cts = new CancellationTokenSource();
private PLCWrapper _plcWrapper;
private MqttClientWrapper mqttClient = null;
private System.Threading.Timer _periodicTimer;
private Dictionary<AcquisitionType, CollectionGroupConfig> _collectionGroups;
private Dictionary<AcquisitionType, Dictionary<string, string>> _plcAddresses;
private AcquisitionType[] AcqTypes = null;
private const int ReconnectionInterval = 5000; // 5 seconds
private const int MaxReconnectAttempts = 3;
public string PlcIpAddress = "127.0.0.1";
public int PlcPort = 9600;
public string MQTTServerAddress = "127.0.0.1";
public int MQTTServerPort = 1883;
public string ClientId = "601UPW01";
//PPM主题
public string Topic = "EVE/60J/EQP/NEW_PPM/601UPW01";
//PPM其他数据
public string Topic1 = "eve/60J/EQP/601UPW01";
//直角坐标机械臂
public string LRobotTopic = "EVE/60J/LRobot/601PWM01_LUR01";
//关节机械臂
public string JRobotTopic = "EVE/60J/JRobot/601PWM01_STR01";
public MqttCredential credentials = null;
public string BU_id = "EVE2BU";
public string District_id = "QJ";
public string Factory_id = "60J";
public string Production_line_id = "MW-601";
public string Device_name = "601UPW01";
public int ThreadInterval = 50;
public static bool SwitchLog = false;
public Service1()
{
InitializeComponent();
ServiceName = "TestMqttService";
CanStop = true; // 服务可以被停止
CanPauseAndContinue = false;
AutoLog = true;
}
protected override void OnStart(string[] args)
{
Logger.WriteInfo($"OnStart:服务启动-{DateTime.Now}");
InitializeConfig();
_plcWrapper = new PLCWrapper(PlcIpAddress, PlcPort);
InitializeTask();
InitializePLC();
InitializeMQTT();
Task.Run(() => RunChangeDetectionTaskAsync(_cts.Token));// 启动变化检测任务
Logger.WriteInfo($"OnStart:服务启动完毕-{DateTime.Now}");
}
protected override async void OnStop()
{
_cts.Cancel();
_periodicTimer?.Dispose();
// 停止 PLC 读取
if (_plcWrapper != null)
{
_plcWrapper.Disconnect();
Logger.WriteInfo("PLC 通讯停止.");
}
if (mqttClient != null)
{
await mqttClient.Disconnect();
Logger.WriteInfo("MQTT 连接已关闭.");
}
Logger.WriteInfo("OnStop");
}
/// <summary>
/// 系统配置文件读取
/// </summary>
private void InitializeConfig()
{
ReadConfigFromXml();
PlcIpAddress = ConfigurationManager.AppSettings["PlcIpAddress"] ?? "127.0.0.1";
PlcPort = int.Parse(ConfigurationManager.AppSettings["PlcPort"] ?? "9600");
MQTTServerAddress = ConfigurationManager.AppSettings["MQTTServerAddress"] ?? "127.0.0.1";
MQTTServerPort = int.Parse(ConfigurationManager.AppSettings["MQTTServerPort"] ?? "1883");
ClientId = ConfigurationManager.AppSettings["ClientId"] ?? "601UPW01";
Topic = ConfigurationManager.AppSettings["Topic"] ?? "EVE/60J/EQP/NEW_PPM/601UPW01";
Topic1 = ConfigurationManager.AppSettings["Topic1"] ?? "eve/60J/EQP/601UPW01";
LRobotTopic = ConfigurationManager.AppSettings["LRobotTopic"] ?? "EVE/60J/LRobot/601PWM01_LUR01";
JRobotTopic = ConfigurationManager.AppSettings["JRobotTopic"] ?? "EVE/60J/JRobot/601PWM01_STR01";
BU_id = ConfigurationManager.AppSettings["BU_id"] ?? "601UPW01";
District_id = ConfigurationManager.AppSettings["District_id"] ?? "QJ";
Factory_id = ConfigurationManager.AppSettings["Factory_id"] ?? "60J";
Production_line_id = ConfigurationManager.AppSettings["Production_line_id"] ?? "MW-601";
Device_name = ConfigurationManager.AppSettings["Device_name"] ?? "601UPW01";
ThreadInterval = int.Parse(ConfigurationManager.AppSettings["ThreadInterval"] ?? "50");
SwitchLog = bool.Parse(ConfigurationManager.AppSettings["SwitchLog"] ?? "True");
MqttDataService.BU_id = BU_id;
MqttDataService.District_id = District_id;
MqttDataService.Factory_id = Factory_id;
MqttDataService.Production_line_id = Production_line_id;
MqttDataService.Device_name = Device_name;
MqttDataService.Work_Center_id = "601FGE02";
Logger.WriteInfo("读取系统配置文件成功!");
}
private void ReadConfigFromXml()
{
try
{
var configPath = Path.Combine(AppDomain.CurrentDomain.BaseDirectory, "Config.xml");
var doc = XDocument.Load(configPath);
string deviceName = "";
_plcAddresses = new Dictionary<AcquisitionType, Dictionary<string, string>>();
_collectionGroups = new Dictionary<AcquisitionType, CollectionGroupConfig>();
foreach (var deviceElement in doc.Descendants("Device"))
{
var isEnabled = bool.Parse(deviceElement.Attribute("isEnable")?.Value ?? "false");
if (isEnabled)
{
deviceName = deviceElement.Attribute("name")?.Value;
foreach (var groupElement in deviceElement.Elements("CollectionGroup"))
{
var groupName = groupElement.Attribute("name").Value;
Logger.WriteInfo($"添加寄存器组类型: {groupName}");
var acquisitionType = (AcquisitionType)Enum.Parse(typeof(AcquisitionType), groupName);
var groupConfig = new CollectionGroupConfig
{
Name = groupName,
StartAddress = groupElement.Attribute("StartAddress").Value,
Length = int.Parse(groupElement.Attribute("length").Value),
DefaultDataType = groupElement.Attribute("DataType")?.Value ?? "Short",
Types = new Dictionary<string, TypeConfig>()
};
var groupAddresses = new Dictionary<string, string>();
foreach (var typeElement in groupElement.Elements("Type"))
{
var typeName = typeElement.Attribute("name").Value;
var typeAddress = typeElement.Attribute("address")?.Value ?? typeElement.Value;
var dataType = typeElement.Attribute("DataType")?.Value ?? groupConfig.DefaultDataType;
// 创建TypeConfig对象
var typeConfig = new TypeConfig
{
Name = typeName,
Address = typeAddress,
DataType = dataType
};
groupConfig.Types[typeName] = typeConfig;
groupAddresses[typeName] = typeAddress;
Logger.WriteInfo($"添加类型: {typeName}, 地址: {typeAddress}, 数据类型: {dataType}");
}
_collectionGroups[acquisitionType] = groupConfig;
_plcAddresses[acquisitionType] = groupAddresses;
}
break; // 只处理第一个启用的Device
}
}
Logger.WriteInfo($"读取 【{deviceName}】地址文件成功!!!");
}
catch (Exception ex)
{
Logger.WriteError($"读取配置文件错误: {ex.Message}");
}
}
private static PLCDataType DetermineDataType(string groupName, string dataTypeAttribute)
{
if (!string.IsNullOrEmpty(dataTypeAttribute))
{
return (PLCDataType)Enum.Parse(typeof(PLCDataType), dataTypeAttribute);
}
if (!string.IsNullOrEmpty(groupName))
{
AcquisitionType type = (AcquisitionType)Enum.Parse(typeof(AcquisitionType), groupName);
// 如果 DataType 属性为空,根据组名推断类型
switch (type)
{
case AcquisitionType.PPM:
return PLCDataType.Short;
case AcquisitionType.Axis:
case AcquisitionType.Cylinder:
return PLCDataType.Float;
default:
throw new ArgumentException($"未知的组名: {groupName}");
}
}
else
{
throw new ArgumentException($"未知的组名: {groupName}");
}
}
#region 初始化Task线程,PLC,MQTT
private void InitializeTask()
{
if (_collectionGroups != null && _collectionGroups.Count > 0)
{
List<AcquisitionType> typesList = new List<AcquisitionType>();
foreach (var kvp in _collectionGroups)
{
switch (kvp.Key)
{
case AcquisitionType.Axis:
case AcquisitionType.Cylinder:
StartPeriodicTask(); // 启动定时读取任务
break;
case AcquisitionType.PPM:
typesList.Add(kvp.Key);
break;
default:
break;
}
}
// 将 List 转换为数组
AcqTypes = typesList.ToArray();
// 如果需要,你可以在这里使用 types 数组
if (AcqTypes.Length > 0)
{
Logger.WriteInfo("添加类型: " + string.Join(", ", AcqTypes));
// 这里可以添加其他使用 types 数组的逻辑
}
}
}
private async void InitializePLC()
{
try
{
bool connected = await _plcWrapper.ConnectAsync();
if (connected)
{
Logger.WriteInfo("PLC初始化并连接成功.");
}
else
{
Logger.WriteInfo("无法连接 PLC.");
throw new Exception("PLC 连接失败");
}
}
catch (Exception ex)
{
Logger.WriteError($"初始化 PLC 错误: {ex.Message}");
}
}
private async void InitializeMQTT()
{
try
{
mqttClient = new MqttClientWrapper(MQTTServerAddress, MQTTServerPort, ClientId, credentials); //初始化
await mqttClient.InitializeAndConnectAsync();//连接MQTT
SubscribeTopic(Topic);//订阅主题
SubscribeTopic(Topic1);//订阅主题1
SubscribeTopic(LRobotTopic);//订阅直角坐标机械臂
SubscribeTopic(JRobotTopic);//订阅关机机械臂
}
catch (Exception ex)
{
Logger.WriteError($"连接 MQTT 代理时出错: {ex.Message}");
}
}
/// <summary>
/// 订阅主题
/// </summary>
/// <param name="topicToSubscribe"></param>
private void SubscribeTopic(string topicToSubscribe)
{
mqttClient.OnMessageReceived += (timestamp, topic, message) =>
{
//Console.WriteLine($"{timestamp} Topic[{topic}]: {message}");
//Logger.WriteMqtt($"{timestamp} Topic[{topic}]");
// 在这里更新你的 UI
};
mqttClient.SubscribeMessage(topicToSubscribe, MqttClientWrapper.MessageFormat.Json, true);
}
/// <summary>
/// 类发布消息
/// </summary>
/// <param name="publishTopic"></param>
/// <param name="sendMessage"></param>
/// <returns></returns>
private async Task SendMqttAsync(string source, string publishTopic, string type, JObject messageObj)
{
try
{
if (publishTopic != null)
{
await mqttClient.SendMessageAsync(source, publishTopic, type, messageObj);
}
}
catch (Exception ex)
{
Console.WriteLine($"发布信息: {ex.Message}");
}
}
/// <summary>
/// 字符串发布消息
/// </summary>
/// <param name="source"></param>
/// <param name="publishTopic"></param>
/// <param name="type"></param>
/// <param name="messageObj"></param>
/// <returns></returns>
private async Task SendMqttAsync(string source, string publishTopic, string type, string messageObj)
{
try
{
if (publishTopic != null)
{
await mqttClient.SendMessageAsync(source, publishTopic, type, messageObj);
}
}
catch (Exception ex)
{
Console.WriteLine($"发布信息: {ex.Message}");
}
}
#endregion
#region 创建定时线程
/// <summary>
/// 轴扭矩定时器采集线程
/// </summary>
private void StartPeriodicTask()
{
Logger.WriteInfo("启动定期数据收集任务");
_periodicTimer = new System.Threading.Timer(async _ =>
{
try
{
Logger.WriteInfo($"定期触发任务: {DateTime.Now}");
await CollectAndPublishData("AxisTask", new[] { AcquisitionType.Axis });
await CollectAndPublishData("AxisTask", new[] { AcquisitionType.Cylinder });
await CollectAndPublishData("LRobotTask", new[] { AcquisitionType.LRobot });
Logger.WriteInfo($"定期任务完成: {DateTime.Now}");
}
catch (Exception ex)
{
Logger.WriteError($"周期性任务执行错误: {ex.Message}");
}
}, null, TimeSpan.Zero, TimeSpan.FromSeconds(1)); // 每秒执行一次
}
/// <summary>
/// 有变化数据采集线程
/// </summary>
/// <param name="workerId"></param>
/// <param name="ct"></param>
/// <returns></returns>
private async Task RunChangeDetectionTaskAsync(CancellationToken ct)
{
while (!ct.IsCancellationRequested)
{
if (_plcWrapper._plcIsConnected)
{
try
{
await CollectAndPublishData("ChangeTask", AcqTypes);
await Task.Delay(500, ct);
}
catch (OperationCanceledException)
{
// 正常取消,无需额外处理
}
catch (Exception ex)
{
Logger.WriteError($"变更检测任务错误: {ex.Message}");
await Task.Delay(1000, ct);
}
}
}
}
#endregion
/// <summary>
/// 读取数据
/// </summary>
/// <param name="source"></param>
/// <param name="addresses"></param>
/// <returns></returns>
private async Task CollectAndPublishData(string source, AcquisitionType[] types)
{
const int maxRetries = 3;
for (int retry = 0; retry < maxRetries; retry++)
{
try
{
var requests = new List<PLCDataReadRequest>();
foreach (var type in types)
{
var config = _collectionGroups[type];
requests.Add(new PLCDataReadRequest
{
StartAddress = config.StartAddress,
Length = (ushort)config.Length,
StorageName = config.Name
});
}
foreach (var request in requests)
{
var result = await _plcWrapper._dataReader.ReadDataAsync(request);
if (result.IsSuccess)
{
await ProcessPLCData(source, request.StorageName, result.Data);
}
else
{
Logger.WriteInfo($"错误读取 {result.StorageName}: {result.ErrorMessage}");
throw new Exception($"读取 PLC 数据失败: {result.ErrorMessage}");
}
}
break; // 如果成功,退出重试循环
}
catch (Exception ex)
{
Logger.WriteError($"Error {source}: {ex.Message}");
if (retry < maxRetries - 1)
{
Logger.WriteInfo($"尝试重新连接到 PLC,尝试 {retry + 1}");
bool reconnected = await AttemptReconnection(_cts.Token);
if (!reconnected)
{
Logger.WriteError("无法重新连接 PLC,放弃进一步尝试");
break;
}
}
else
{
Logger.WriteError($"最大重试次数 ({maxRetries}) 已到达,放弃操作");
}
}
}
}
/// <summary>
/// 重连PLC
/// </summary>
/// <param name="ct"></param>
/// <returns></returns>
private async Task<bool> AttemptReconnection(CancellationToken ct)
{
const int MaxReconnectAttempts = 3;
const int ReconnectionInterval = 5000; // 5 seconds
for (int attempt = 1; attempt <= MaxReconnectAttempts; attempt++)
{
if (ct.IsCancellationRequested)
{
return false;
}
Logger.WriteInfo($"尝试重新连接PLC,第 {attempt} 次尝试");
bool reconnected = await _plcWrapper.ReconnectAsync();
if (reconnected)
{
Logger.WriteInfo("PLC重新连接成功");
return true;
}
if (attempt < MaxReconnectAttempts)
{
await Task.Delay(ReconnectionInterval, ct);
}
}
Logger.WriteError($"PLC重新连接失败,已尝试 {MaxReconnectAttempts} 次");
return false;
}
/// <summary>
/// 读取PLC数据
/// </summary>
/// <param name="source"></param>
/// <param name="acquisitionTypeName"></param>
/// <param name="data"></param>
/// <returns></returns>
private async Task ProcessPLCData(string source, string acquisitionTypeName, byte[] data)
{
try
{
AcquisitionType type = (AcquisitionType)Enum.Parse(typeof(AcquisitionType), acquisitionTypeName);
var typeConfigs = GetTypeConfigs(type).ToList();
bool hasChanges = false;
List<StationData> processedValues = null;
switch (type)
{
case AcquisitionType.PPM:
var ppmResult = ProcessShortData(type, data, typeConfigs);
hasChanges = ppmResult.hasChanges;
processedValues = ppmResult.values;
break;
case AcquisitionType.Axis:
case AcquisitionType.Cylinder:
case AcquisitionType.LRobot:
var floatResult = ProcessFloatData(type, data, typeConfigs);
hasChanges = floatResult.hasChanges;
processedValues = floatResult.values;
break;
default:
Logger.WriteError($"未知的采集类型: {type}");
return;
}
if (hasChanges && processedValues != null)
{
// 上传到MQTT服务器
await UpMQTTServerDataNew(source, type, processedValues);
}
else
{
Logger.WriteInfo($"{type} 数据未改变,未上传");
}
}
catch (Exception ex)
{
Logger.WriteError($"处理 PLC 数据时出错(源: {source}, 类型: {acquisitionTypeName}): {ex.Message}");
}
}
// 相应地,ProcessFloatData方法也需要修改返回类型
private List<StationData> _lastFloatStationData = new List<StationData>();
private (bool hasChanges, List<StationData> values) ProcessFloatData(AcquisitionType type, byte[] rawData, List<TypeConfig> typeConfigs)
{
bool hasChanges = false;
var currentValues = new DataValue[typeConfigs.Count];
// 记录实际需要处理的数据点数量
Logger.WriteInfo($"开始处理数据: typeConfigs.Count={typeConfigs.Count}, rawData.Length={rawData.Length}");
var sw = new Stopwatch();
sw.Restart();
// 初始化数组为完整的大小
currentValues = new DataValue[typeConfigs.Count];
var stationDataList = new List<StationData>();
try
{
// 第一步:处理所有Float类型数据(主数据)
var floatConfigs = typeConfigs.Where(t => t.DataType.Equals("Float", StringComparison.OrdinalIgnoreCase)).ToList();
foreach (var config in floatConfigs)
{
// 从地址中提取偏移量
if (int.TryParse(config.Address.Substring(1), out int address))
{
int startAddress = int.Parse(typeConfigs[0].Address.Substring(1));
int offset = (address - startAddress) * 2;
if (offset >= 0 && offset + 4 <= rawData.Length)
{
byte[] valueBytes = new byte[4];
for (int j = 0; j < 4; j++)
{
valueBytes[j] = (byte)(Convert.ToInt32(rawData[offset + j]) & 0xFF);
}
object convertedValue = ConvertFromBytes(valueBytes, "Float");
int index = typeConfigs.IndexOf(config);
currentValues[index] = new DataValue
{
Name = config.Name,
Value = convertedValue,
DataType = config.DataType
};
}
}
}
// 第二步:处理所有Short类型数据
var shortConfigs = typeConfigs.Where(t => t.DataType.Equals("Short", StringComparison.OrdinalIgnoreCase)).ToList();
foreach (var config in shortConfigs)
{
if (int.TryParse(config.Address.Substring(1), out int address))
{
int startAddress = int.Parse(typeConfigs[0].Address.Substring(1));
int offset = (address - startAddress) * 2;
if (offset >= 0 && offset + 2 <= rawData.Length)
{
byte[] valueBytes = new byte[2];
for (int j = 0; j < 2; j++)
{
valueBytes[j] = (byte)(Convert.ToInt32(rawData[offset + j]) & 0xFF);
}
object convertedValue = ConvertFromBytes(valueBytes, "Short");
int index = typeConfigs.IndexOf(config);
currentValues[index] = new DataValue
{
Name = config.Name,
Value = convertedValue,
DataType = config.DataType
};
}
}
}
// 第三步:组织StationData列表
foreach (var floatConfig in floatConfigs)
{
string baseName = floatConfig.Name; // 例如 "S_MagLev"
var mainValue = currentValues[typeConfigs.IndexOf(floatConfig)]?.Value;
// 先获取MainValue,如果为0则跳过这个工位,暂时不过滤
//if (mainValue == null || Convert.ToDouble(mainValue) == 0)
//{
// Logger.WriteInfo($"跳过工位 {baseName}: MainValue为0或null");
// continue;
//}
// 只检查MainValue是否与上一次的值相同,暂时不检查
//var lastData = _lastStationData.FirstOrDefault(x => x.Name == baseName);
//if (lastData != null)
//{
// bool valueEqual = Convert.ToDouble(lastData.MainValue) == Convert.ToDouble(mainValue);
// if (valueEqual)
// {
// Logger.WriteInfo($"跳过工位 {baseName}: 值未变化 当前值:{mainValue}");
// continue;
// }
// Logger.WriteInfo($"工位 {baseName} 数值发生变化: {lastData.MainValue} -> {mainValue}");
//}
// 创建StationData对象
var stationData = new StationData
{
Name = baseName,
MainValue = mainValue,
Timestamp = DateTime.Now,
};
stationDataList.Add(stationData);
Logger.WriteInfo($"添加工位数据: {stationData.Name}, 值: {stationData.MainValue}, 时间戳: {stationData.Timestamp:HH:mm:ss.fff}");
}
// 检查是否有新数据需要更新
if (stationDataList.Any())
{
hasChanges = true; // 如果有数据通过过滤,就表示有变化
// 创建深拷贝以避免引用问题
_lastFloatStationData = stationDataList.Select(s => new StationData
{
Name = s.Name,
MainValue = s.MainValue,
Timestamp = s.Timestamp
}).ToList();
_lastFloatStationData = stationDataList;//把当前的新数据赋值到旧的集合中
//// 记录变化信息
//foreach (var newData in stationDataList)
//{
// Logger.WriteInfo($"工位 {newData.Name} 更新: 时间戳: {newData.Timestamp:HH:mm:ss.fff}, 值: {newData.MainValue}");
//}
}
else
{
Logger.WriteInfo("没有发现时间戳变化的数据");
hasChanges = false;
}
sw.Stop();
Logger.WriteInfo($"处理结束耗时:{sw.ElapsedMilliseconds}ms");
return (hasChanges, stationDataList);
}
catch (Exception ex)
{
Logger.WriteError($"处理数据时出错: {ex.Message}");
return (false, stationDataList);
}
}
/// <summary>
/// 检测PPM数据是否有变化
/// </summary>
/// <param name="data"></param>
/// <returns></returns>
private List<StationData> _lastStationData = new List<StationData>();
private readonly StationDataComparer2 _comparer = new StationDataComparer2();
private (bool hasChanges, List<StationData> values) ProcessShortData(AcquisitionType type, byte[] rawData, List<TypeConfig> typeConfigs)
{
var sw = new Stopwatch();
sw.Start();
Logger.WriteInfo($"开始新一轮数据处理");
try
{
var currentStationData = new Dictionary<string, StationData>();
bool hasNewData = false;
// 1. 获取所有主值配置(Float类型)
var mainValueConfigs = typeConfigs
.Where(t => string.Equals(t.DataType, "Float", StringComparison.OrdinalIgnoreCase))
.OrderBy(t => int.Parse(t.Address.Substring(1)))
.ToList();
Logger.WriteInfo($"找到 {mainValueConfigs.Count} 个Float类型配置");
foreach (var config in mainValueConfigs)
{
Logger.WriteInfo($"Float配置: {config.Name}, 地址: {config.Address}");
}
// 2. 获取所有时间配置(Short类型)
var timeConfigs = typeConfigs
.Where(t => string.Equals(t.DataType, "Short", StringComparison.OrdinalIgnoreCase))
.ToList();
// 3. 首先读取所有工位的当前值和时间戳
foreach (var mainConfig in mainValueConfigs)
{
string stationName = mainConfig.Name;
// 读取主值
double? mainValue = ReadFloatValue(mainConfig, rawData, typeConfigs[0].Address);
if (mainValue == null || mainValue <= 0)
{
Logger.WriteInfo($"工位 {stationName} 主值无效: {mainValue}");
continue;
}
// 读取时间组件
DateTime timestamp = ReadTimestamp(stationName, timeConfigs, rawData, typeConfigs[0].Address);
// 添加到当前数据集合
currentStationData[stationName] = new StationData
{
Name = stationName,
MainValue = mainValue,
Timestamp = timestamp
};
Logger.WriteInfo($"读取工位数据: {stationName}, 值: {mainValue}, 时间: {timestamp:HH:mm:ss.fff}");
}
// 4. 比较新旧数据,确定是否有变化
if (currentStationData.Any())
{
if (_lastStationData == null || !_lastStationData.Any())
{
hasNewData = true;
Logger.WriteInfo("首次数据采集,记录所有数据");
}
else
{
// 检查每个工位的数据是否有变化
foreach (var current in currentStationData.Values)
{
var lastData = _lastStationData.FirstOrDefault(x => x.Name == current.Name);
if (lastData == null)
{
hasNewData = true;
Logger.WriteInfo($"发现新工位数据: {current.Name}");
break;
}
// 检查值和时间戳是否都有变化
bool valueChanged = Math.Abs(Convert.ToDouble(lastData.MainValue) - Convert.ToDouble(current.MainValue)) > 0.001;
bool timeChanged = Math.Abs((current.Timestamp - lastData.Timestamp).TotalMilliseconds) >= 500;
if (valueChanged || timeChanged)
{
hasNewData = true;
Logger.WriteInfo($"工位 {current.Name} 数据发生变化 - 值变化: {valueChanged}, 时间变化: {timeChanged}");
break;
}
}
}
}
// 5. 如果有新数据,更新最后处理的数据
var resultList = currentStationData.Values.ToList();
if (hasNewData)
{
_lastStationData = new List<StationData>(resultList);
Logger.WriteInfo($"更新状态数据,数量: {resultList.Count}");
}
sw.Stop();
Logger.WriteInfo($"处理完成,耗时: {sw.ElapsedMilliseconds}ms, 数据点数量: {resultList.Count}, 有变化: {hasNewData}");
return (hasNewData, resultList);
}
catch (Exception ex)
{
Logger.WriteError($"数据处理错误: {ex.Message}\n{ex.StackTrace}");
return (false, new List<StationData>());
}
}
private double? ReadFloatValue(TypeConfig config, byte[] rawData, string baseAddress)
{
try
{
if (!int.TryParse(config.Address.Substring(1), out int address) ||
!int.TryParse(baseAddress.Substring(1), out int startAddress))
{
return null;
}
int offset = (address - startAddress) * 2;
if (offset < 0 || offset + 4 > rawData.Length)
{
return null;
}
byte[] valueBytes = new byte[4];
Array.Copy(rawData, offset, valueBytes, 0, 4);
var value = ConvertFromBytes(valueBytes, "Float");
return value != null ? Convert.ToDouble(value) : null;
}
catch
{
return null;
}
}
private DateTime ReadTimestamp(string stationName, List<TypeConfig> timeConfigs, byte[] rawData, string baseAddress)
{
// 获取时间组件配置
var hourConfig = timeConfigs.FirstOrDefault(t => t.Name == $"{stationName}_Hour");
var minConfig = timeConfigs.FirstOrDefault(t => t.Name == $"{stationName}_Min");
var secConfig = timeConfigs.FirstOrDefault(t => t.Name == $"{stationName}_Sec");
var msecConfig = timeConfigs.FirstOrDefault(t => t.Name == $"{stationName}_MSec");
// 读取时间值
int hour = ReadShortValue(hourConfig, rawData, baseAddress);
int minute = ReadShortValue(minConfig, rawData, baseAddress);
int second = ReadShortValue(secConfig, rawData, baseAddress);
int millisecond = ReadShortValue(msecConfig, rawData, baseAddress);
// 验证时间值
hour = Math.Min(Math.Max(hour, 0), 23);
minute = Math.Min(Math.Max(minute, 0), 59);
second = Math.Min(Math.Max(second, 0), 59);
millisecond = Math.Min(Math.Max(millisecond, 0), 999);
// 获取当前时间
var now = DateTime.Now;
// 构建读取的时间
var readTime = new DateTime(
now.Year,
now.Month,
now.Day,
hour,
minute,
second,
millisecond
);
// 如果读取的时间比当前时间大,说明应该是前一天的数据
if (readTime > now)
{
readTime = readTime.AddDays(-1);
Logger.WriteInfo($"检测到跨天数据,调整日期为前一天: {readTime:yyyy-MM-dd HH:mm:ss.fff}");
}
return readTime;
}
private int ReadShortValue(TypeConfig config, byte[] rawData, string baseAddress)
{
try
{
if (config == null || !int.TryParse(config.Address.Substring(1), out int address) ||
!int.TryParse(baseAddress.Substring(1), out int startAddress))
{
return 0;
}
int offset = (address - startAddress) * 2;
if (offset < 0 || offset + 2 > rawData.Length)
{
return 0;
}
byte[] valueBytes = new byte[2];
Array.Copy(rawData, offset, valueBytes, 0, 2);
var value = ConvertFromBytes(valueBytes, "Short");
return Convert.ToInt32(value);
}
catch
{
return 0;
}
}
private object ConvertFromBytes(byte[] bytes, string dataType)
{
try
{
if (bytes == null || bytes.Length == 0)
return null;
switch (dataType.ToLower())
{
case "bool":
return BitConverter.ToInt16(bytes, 0) != 0;
case "short":
return ShortLib.GetShortFromByteArray(bytes, 0, DataFormat.CDAB);
case "int":
return IntLib.GetIntFromByteArray(bytes, 0, DataFormat.CDAB);
case "float":
return FloatLib.GetFloatFromByteArray(bytes, 0, DataFormat.CDAB);
default:
return ShortLib.GetShortFromByteArray(bytes, 0, DataFormat.CDAB);
}
}
catch (Exception ex)
{
Logger.WriteError($"转换数据类型失败: {BitConverter.ToString(bytes)} to {dataType}, {ex.Message}");
return null;
}
}
/// <summary>
/// 上传MQTT
/// </summary>
/// <param name="source"></param>
/// <param name="type"></param>
/// <param name="data"></param>
/// <returns></returns>
public async Task UpMQTTServerDataNew(string source, AcquisitionType type, List<StationData> processedValues)
{
try
{
if (processedValues == null || processedValues.Count == 0)
{
return;
}
//var addresses = _plcAddresses[type];
//var config = _collectionGroups[type];
var message = MqttDataService.CreateMqttMessage(source, type, processedValues, _plcAddresses, _collectionGroups);
if (message != null)
{
var messageJson = JsonConvert.SerializeObject(message, new JsonSerializerSettings
{
NullValueHandling = NullValueHandling.Ignore,
Formatting = Formatting.None
});
if (type.ToString() == "PPM")
{
await SendMqttAsync(source, Topic, type.ToString(), messageJson);
}
else if (type.ToString() == "Axis" || type.ToString() == "Cylinder")
{
await SendMqttAsync(source, Topic1, type.ToString(), messageJson);
}
}
if (type.ToString() == "LRobot")
{
var message1 = MqttDataService.LRobotMessage(source, type, processedValues, _plcAddresses, _collectionGroups);
if (message1 != null)
{
var messageJson1 = JsonConvert.SerializeObject(message1, new JsonSerializerSettings
{
NullValueHandling = NullValueHandling.Ignore,
Formatting = Formatting.None
});
if (type.ToString() == "LRobot")
{
await SendMqttAsync(source, LRobotTopic, type.ToString(), messageJson1);
}
}
}
}
catch (Exception ex)
{
Logger.WriteError($"方法:上传MQTTServer数据,上传失败: {ex.Message}", ex);
}
}
private List<short> ConvertToShortList(object[] data)
{
return data.Select(d => Convert.ToInt16(d)).ToList();
}
private List<float> ConvertToFloatList(object[] data)
{
return data.Select(d => Convert.ToSingle(d)).ToList();
}
// 获取特定类型的数据类型
public string GetDataType(AcquisitionType acquisitionType, string typeName)
{
if (_collectionGroups.TryGetValue(acquisitionType, out var groupConfig))
{
if (groupConfig.Types.TryGetValue(typeName, out var typeConfig))
{
return typeConfig.DataType;
}
}
return "Short"; // 默认返回Short类型
}
// 获取数据类型的字节大小
public int GetTypeSize(string dataType)
{
switch (dataType.ToLower())
{
case "bool":
return 2; // 在PLC中通常用2字节表示
case "short":
return 2; // 2字节
case "int":
return 4; // 4字节
case "float":
return 4; // 4字节
default:
return 2; // 默认2字节
}
}
// 获取组内所有类型配置
public IEnumerable<TypeConfig> GetTypeConfigs(AcquisitionType acquisitionType)
{
if (_collectionGroups.TryGetValue(acquisitionType, out var groupConfig))
{
return groupConfig.Types.Values;
}
return Enumerable.Empty<TypeConfig>();
}
}
/// <summary>
/// 时间转化
/// </summary>
public static class DateTimeExtensions
{
public static long ToUnixTimeMilliseconds(this DateTime dateTime)
{
return (long)(dateTime.ToUniversalTime() - new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Utc)).TotalMilliseconds;
}
public static long GetUnixTimestamp()
{
return (DateTime.Now.ToUniversalTime().Ticks - 621355968000000000) / 10000000;
}
}
// 辅助类
public class DataValue
{
public string Name { get; set; }
public object Value { get; set; }
public string DataType { get; set; }
public override bool Equals(object obj)
{
if (obj is DataValue other)
{
if (Value == null && other.Value == null) return true;
if (Value == null || other.Value == null) return false;
switch (DataType.ToLower())
{
case "float":
if (Value is float f1 && other.Value is float f2)
return Math.Abs(f1 - f2) < 0.000001f;
break;
case "bool":
if (Value is bool b1 && other.Value is bool b2)
return b1 == b2;
break;
}
return Value.Equals(other.Value);
}
return false;
}
public override int GetHashCode()
{
return Value?.GetHashCode() ?? 0;
}
}
}