1180 lines
45 KiB
C#
1180 lines
45 KiB
C#
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;
|
||
}
|
||
}
|
||
}
|