MES / DataAcquisition · Files

using HHECS.BllModel;
using HHECS.DAQWebClient.Hubs;
using HslCommunication.Profinet.Siemens;
using Microsoft.AspNetCore.SignalR;
using MQTTnet.Formatter;
using MQTTnet;
using HHECS.DAQWebClient.Communications;
using HHECS.DAQWebClient.Model;
using HHECS.DAQShared.Dto;
using HHECS.DAQShared.Common.Enums;
using System.Text.Json;
using System.Text;
using HHECS.DAQWebClient.Models;

namespace HHECS.DAQWebClient.Services
{
    public class SystemBackgroundService : BackgroundService
    {
        private readonly IFreeSql _freeSql;
        private readonly IHubContext<DAQHub, IDAQHub> _hub;
        private readonly CommonService _commonService;

        private List<ICommunication> communications = new List<ICommunication>();
        private IMqttClient? mqttClient;

        private CancellationTokenSource _cts = new CancellationTokenSource();

        private readonly List<Task> tasks = new List<Task>();

        public SystemBackgroundService(IFreeSql freeSql, IHubContext<DAQHub, IDAQHub> hub, CommonService commonService)
        {
            _freeSql = freeSql;
            _hub = hub;
            _commonService = commonService;
        }

        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
        {
            while (!stoppingToken.IsCancellationRequested)
            {
                try
                {
                    await Task.Delay(1000, stoppingToken);
                    await _hub.Clients.All.QueueNotice(_commonService.EquipmentDataQueues.Count);
                    await _hub.Clients.All.MqttConnectedNotice(mqttClient?.IsConnected == true);
                    //停止
                    if (!_commonService.IsStart)
                    {
                        if (!_cts.IsCancellationRequested)
                        {
                            _cts.Cancel();
                        }
                        //停止状态，数据未释放，此处进行释放操作
                        if (tasks.Count != 0)
                        {
                            _commonService.PrintLog("正在释放资源...");
                            if (mqttClient != null)
                            {
                                await mqttClient.TryDisconnectAsync();
                                mqttClient = null;
                            }

                            Parallel.ForEach(communications, item => item.ConnectClose());
                            communications.Clear();
                            _commonService.Equipments.Clear();
                            tasks.Clear();
                            _commonService.PrintLog("资源已释放");
                        }
                        continue;
                    }

                    //启动状态，数据已释放，需要初始化
                    if (tasks.Count == 0)
                    {
                        if (_cts.IsCancellationRequested)
                        {
                            _cts = new CancellationTokenSource();
                        }

                        if (mqttClient == null)
                        {
                            InitMqttClient();
                        }
                        var equipmentResult = _commonService.InitialEquipmentData();
                        if (!equipmentResult.Success)
                        {
                            _commonService.PrintLog(equipmentResult.Msg, LogLevel.Warning);
                            continue;
                        }

                        var equipmentIPAddressList = _commonService.Equipments.Select(x => x.IP).ToList();

                        var communicationConfigs = _freeSql.Queryable<CommunicationConfig>().Where(x => !x.Disable && equipmentIPAddressList.Contains(x.IpAddress)).ToList();

                        if (communicationConfigs.Count == 0)
                        {
                            _commonService.PrintLog($"通讯配置数据为空，请配置数据后操作！", LogLevel.Warning);
                            continue;
                        }
                        communications = InitialCommunication(communicationConfigs);
                        //采集逻辑
                        foreach (var item in communications)
                        {
                            var task = Task.Run(async () =>
                            {
                                while (!_cts.IsCancellationRequested)
                                {
                                    try
                                    {
                                        await Task.Delay(1000, _cts.Token);
                                        var equipmentTemps = _commonService.Equipments.Where(x => x.IP == item.IpAddress).ToList();
                                        var props = equipmentTemps.SelectMany(x => x.EquipmentProps).Where(x => x.PropType != EquipmentPropType.Self).ToList();

                                        if (props.Count == 0)
                                        {
                                            continue;
                                        }

                                        var temps = props.Select(x => new DataItem
                                        {
                                            Id = x.Id,
                                            Code = x.Code,
                                            DataAddress = x.Address,
                                            DataType = x.DataType,
                                            Value = string.Empty
                                        }).ToList();
                                        //读取数据
                                        var result = item.Read(temps);
                                        if (!result.Success)
                                        {
                                            _commonService.PrintLog($"读取设备[{item.IpAddress}]数据失败：{result.Msg}", LogLevel.Error);
                                            continue;
                                        }

                                        //赋值
                                        foreach (var item in props)
                                        {
                                            item.Value = temps.Where(x => x.Id == item.Id).Select(x => x.Value).FirstOrDefault() ?? string.Empty;
                                            item.Updated = DateTime.Now;
                                        }

                                        foreach (var item in equipmentTemps)
                                        {
                                            var tags = item.EquipmentProps.Where(x =>
                                            {
                                                //过滤掉不需要上报的报警地址
                                                if (x.PropType == EquipmentPropType.PLCMonitorAddress && !string.IsNullOrWhiteSpace(x.MonitorCompareValue))
                                                {
                                                    //报警地址，只返回有报警的记录
                                                    return x.Value != x.MonitorCompareValue;
                                                }
                                                return true;
                                            }).Select(x => new TagItemDto
                                            {
                                                Tag = x.Code,
                                                Value = x.Value
                                            }).ToList();

                                            var record = new EquipmentDataQueue
                                            {
                                                EquipmentCode = item.Code,
                                                EquipmentName = item.Name,
                                                EquipmentTypeCode = item.EquipmentType.Code,
                                                IsCommit = false,
                                                Reported = JsonSerializer.Serialize(tags),
                                                Version = 1,
                                                SourceTimestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(),
                                                Created = DateTime.Now,
                                            };
                                            _commonService.EquipmentDataQueues.Enqueue(record);
                                        }
                                    }
                                    catch (Exception ex)
                                    {
                                        _commonService.PrintLog($"读取IP：{item.IpAddress}设备出现异常：{ex.Message}", LogLevel.Debug);
                                    }
                                }
                            }, _cts.Token);
                            tasks.Add(task);
                        }
                    }
                }
                catch (Exception ex)
                {
                    _commonService.PrintLog($"{nameof(SystemBackgroundService)}异常：{ex.Message}", LogLevel.Debug);
                }
            }
        }

        private List<ICommunication> InitialCommunication(IEnumerable<CommunicationConfig> communicationConfigs)
        {
            var result = new List<ICommunication>();
            try
            {
                foreach (var item in communicationConfigs)
                {
                    ICommunication communication;
                    switch (item.CommunicationType)
                    {
                        case CommunicationTypeConst.None:
                            break;
                        case CommunicationTypeConst.KukaVarProxy:
                            communication = new KukaAvarProxyCommunication(item.Id, item.IpAddress, item.Port);
                            result.Add(communication);
                            break;

                        case CommunicationTypeConst.Siemens_S1200:
                            communication = new SiemensS7Communication(item.Id, SiemensPLCS.S1200, item.IpAddress);
                            result.Add(communication);
                            break;
                        case CommunicationTypeConst.Siemens_S1500:
                            communication = new SiemensS7Communication(item.Id, SiemensPLCS.S1500, item.IpAddress);
                            result.Add(communication);
                            break;
                        case CommunicationTypeConst.Siemens_S200Smart:
                            communication = new SiemensS7Communication(item.Id, SiemensPLCS.S200Smart, item.IpAddress);
                            result.Add(communication);
                            break;
                        case CommunicationTypeConst.Siemens_S200Smart:
                            communication = new MelsecCommunication(item.Id, item.IpAddress);
                            result.Add(communication);
                            break;
                        case CommunicationTypeConst.TcpClient:
                            communication = new TcpClientCommunication(item.Id, item.IpAddress, item.Port);
                            result.Add(communication);
                            break;
                        case CommunicationTypeConst.ModbusTcp:
                            communication = new ModbusTcpCommunication(item.Id, item.IpAddress, item.Port);
                            result.Add(communication);
                            break;
                        default:
                            break;
                    }
                }
            }
            catch (Exception ex)
            {
                _commonService.PrintLog($"设备通讯初始化异常：{ex.Message}", LogLevel.Debug);
            }
            return result;
        }

        private void InitMqttClient()
        {
            try
            {
                var clientId = _freeSql.Queryable<LocalConfig>().Where(x => x.Code == ConfigType.ClientId.ToString()).First(x => x.Value);
                var mqttWebSocketServerConfig = _commonService.IsProductionEnvironment ? ConfigType.MqttWebSocketServer : ConfigType.MqttWebSocketDevelopmentServer;

                var mqttWebSocketServerUrl = _freeSql.Queryable<LocalConfig>().Where(x => x.Code == mqttWebSocketServerConfig.ToString()).First(x => x.Value);
                if (string.IsNullOrWhiteSpace(mqttWebSocketServerUrl))
                {
                    _commonService.PrintLog($"MQTT服务地址未配置", LogLevel.Warning);
                    return;
                }
                if (!mqttWebSocketServerUrl.StartsWith("ws://", StringComparison.OrdinalIgnoreCase)
                    && !mqttWebSocketServerUrl.StartsWith("wss://", StringComparison.OrdinalIgnoreCase))
                {
                    _commonService.PrintLog($"MQTT服务地址格式不正确，目前仅支持WebSocketServer，地址必须是以“ws://”或者“wss://”开始的地址格式", LogLevel.Warning);
                    return;
                }

                var mqttFactory = new MqttClientFactory();
                mqttClient = mqttFactory.CreateMqttClient();

                //定义Topic
                var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder()
                .WithTopicFilter($"MQTTnet.RPC/+/+")
                    .Build();

                //消息接收
                mqttClient.ApplicationMessageReceivedAsync += async e =>
                {
                    try
                    {
                        if (e.ApplicationMessage.Topic.EndsWith("/response", StringComparison.OrdinalIgnoreCase))
                        {
                            return;
                        }

                        var topic = e.ApplicationMessage.Topic;
                        var tempArray = topic.Split('/');
                        if (tempArray.Length != 3)
                        {
                            return;
                        }
                        var methodName = tempArray[2];
                        if (!methodName.StartsWith(clientId, StringComparison.OrdinalIgnoreCase))
                        {
                            return;
                        }

                        var requestPayload = e.ApplicationMessage.Payload;
                        var requestPayloadString = Encoding.Default.GetString(requestPayload);
                        var responseTopic = e.ApplicationMessage.ResponseTopic;
                        if (string.IsNullOrWhiteSpace(responseTopic))
                        {
                            responseTopic = $"{topic}/response";
                        }
                        if (methodName.Equals($"{clientId}_Read", StringComparison.OrdinalIgnoreCase))
                        {
                            var requestData = JsonSerializer.Deserialize<List<EquipmentAddressData>>(requestPayloadString)!;
                            var result = MqttRpcReadHandler(requestData);
                            var responsePayload = JsonSerializer.Serialize(result);
                            await mqttClient.PublishStringAsync(responseTopic, responsePayload, cancellationToken: _cts.Token);
                            _commonService.PrintLog($"[{e.ClientId}]远程触发了Read操作，状态：{result.Success}，信息：{result.Msg}");
                        }
                        else if (methodName.Equals($"{clientId}_Write", StringComparison.OrdinalIgnoreCase))
                        {
                            var requestData = JsonSerializer.Deserialize<List<EquipmentAddressData>>(requestPayloadString)!;
                            var result = MqttRpcWriteHandler(requestData);
                            var responsePayload = JsonSerializer.Serialize(result);
                            await mqttClient.PublishStringAsync(responseTopic, responsePayload, cancellationToken: _cts.Token);
                            _commonService.PrintLog($"[{e.ClientId}]远程触发了Write操作，状态：{result.Success}，信息：{result.Msg}");
                        }
                    }
                    catch (Exception ex)
                    {
                        _commonService.PrintLog($"[MQTT]ApplicationMessageReceived异常：{ex.Message}", LogLevel.Debug);
                    }
                };

                mqttClient.DisconnectedAsync += async arg =>
                {
                    if (_cts.IsCancellationRequested)
                    {
                        //线程已取消
                        return;
                    }
                    await mqttClient.ReconnectAsync(_cts.Token);
                    if (mqttClient.IsConnected)
                    {
                        _commonService.PrintLog($"MQTT服务重连成功！");
                        //重连成功后，重新订阅主题
                        await mqttClient.SubscribeAsync(mqttSubscribeOptions, _cts.Token);
                    }
                };

                var mqttClientOptions = new MqttClientOptionsBuilder()
                    //.WithWebSocketServer(o => o.WithUri("wss://broker.emqx.io:8084/mqtt"))
                    .WithWebSocketServer(o => o.WithUri(mqttWebSocketServerUrl))
                    .WithProtocolVersion(MqttProtocolVersion.V500)
                    .WithTlsOptions(o =>
                    {
                        // The used public broker sometimes has invalid certificates. This sample accepts all
                        // certificates. This should not be used in live environments.
                        o.WithCertificateValidationHandler(_ => true);
                    }).Build();
                var connectResult = mqttClient.ConnectAsync(mqttClientOptions, _cts.Token).GetAwaiter().GetResult();
                if (connectResult.ResultCode == MqttClientConnectResultCode.Success)
                {
                    mqttClient.SubscribeAsync(mqttSubscribeOptions, _cts.Token).Wait(_cts.Token);
                    _commonService.PrintLog($"连接MQTT服务成功");
                }
                else
                {
                    _commonService.PrintLog($"连接MQTT服务失败：{connectResult.ResultCode}", LogLevel.Warning);
                }
            }
            catch (Exception ex)
            {
                _commonService.PrintLog($"[MQTT]异常：{ex.Message}", LogLevel.Debug);
            }
        }

        private BllResult<List<EquipmentAddressData>> MqttRpcReadHandler(List<EquipmentAddressData> requestData)
        {
            try
            {
                foreach (var item in requestData)
                {
                    var equipment = _commonService.Equipments.Where(x => x.Code == item.EquipmentCode).FirstOrDefault();
                    if (equipment == null)
                    {
                        return BllResultFactory.Error<List<EquipmentAddressData>>($"未查询到设备[{item.EquipmentCode}]数据");
                    }
                    var communication = communications.Where(x => x.IpAddress == equipment.IP).FirstOrDefault();
                    if (communication == null)
                    {
                        return BllResultFactory.Error<List<EquipmentAddressData>>($"获取设备[{item.EquipmentCode}]连接对象失败");
                    }
                    var temps = item.Nodes.Select(x => new DataItem
                    {
                        Id = x.Id,
                        Code = item.EquipmentCode,
                        DataAddress = x.DataAddress,
                        DataType = Enum.Parse<EquipmentDataType>(x.DataType)
                    }).ToList();
                    var result = communication.Read(temps);
                    if (!result.Success)
                    {
                        return BllResultFactory.Error<List<EquipmentAddressData>>($"读取设备[{item.EquipmentCode}]数据失败：{result.Msg}");
                    }
                    //赋值
                    foreach (var temp in temps)
                    {
                        var prop = item.Nodes.Where(x => x.DataAddress == temp.DataAddress).FirstOrDefault();
                        if (prop == null)
                        {
                            continue;
                        }
                        prop.Value = temp.Value;
                    }
                }
                return BllResultFactory.Success(requestData);
            }
            catch (Exception ex)
            {
                return BllResultFactory.Error<List<EquipmentAddressData>>(ex.Message);
            }
        }

        private BllResult MqttRpcWriteHandler(List<EquipmentAddressData> requestData)
        {
            try
            {
                foreach (var item in requestData)
                {
                    var equipment = _commonService.Equipments.Where(x => x.Code == item.EquipmentCode).FirstOrDefault();
                    if (equipment == null)
                    {
                        return BllResultFactory.Error<List<EquipmentAddressData>>($"未查询到设备[{item.EquipmentCode}]数据");
                    }
                    var communication = communications.Where(x => x.IpAddress == equipment.IP).FirstOrDefault();
                    if (communication == null)
                    {
                        return BllResultFactory.Error<List<EquipmentAddressData>>($"获取设备[{item.EquipmentCode}]连接对象失败");
                    }
                    var temps = item.Nodes.Select(x => new DataItem
                    {
                        Id = x.Id,
                        Code = item.EquipmentCode,
                        DataAddress = x.DataAddress,
                        DataType = Enum.Parse<EquipmentDataType>(x.DataType)
                    }).ToList();
                    var result = communication.Write(temps);
                    if (!result.Success)
                    {
                        return BllResultFactory.Error<List<EquipmentAddressData>>($"读取设备[{item.EquipmentCode}]数据失败：{result.Msg}");
                    }
                }
                return BllResultFactory.Success();
            }
            catch (Exception ex)
            {
                return BllResultFactory.Error(ex.Message);
            }
        }
    }
}