MES / DataAcquisition · Files

using DataAcquisition.Common.Communications;
using DataAcquisition.Common.Enums;
using DataAcquisition.Common.Utils;
using DataAcquisition.DataAccess;
using DataAcquisition.Models;
using DataAcquisition.Services.DataAnalysis;
using DataAcquisition.ViewModels.IOT;
using HslCommunication.Profinet.Siemens;
using Microsoft.EntityFrameworkCore;
using System.Collections.Concurrent;
using System.Diagnostics;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Text.Json;

namespace DataAcquisition.Services
{
    public class SystemBackgroundService : BackgroundService
    {
        private readonly IDbContextFactory<DataContext> _dbContextFactory;
        private readonly DataCacheService _dataCacheService;
        private readonly IEnumerable<IDataAnalysis> _dataAnalyses;
        private readonly IotService _iotService;
        private readonly IConfiguration _configuration;
        private readonly SystemLog systemLog = SystemLog.Instance;
        private readonly object _lock = new object();
        private readonly bool _autoCommit;

        private readonly ConcurrentQueue<EquipmentDataQueue> _equipmentDataQueues = new ConcurrentQueue<EquipmentDataQueue>();
        private readonly TcpListener _tcpServer;

        private List<ICommunication> communications = new List<ICommunication>();

        public SystemBackgroundService(IDbContextFactory<DataContext> dbContextFactory, DataCacheService dataCacheService, IEnumerable<IDataAnalysis> dataAnalyses, IotService iotService, IConfiguration configuration)
        {
            _tcpServer = new TcpListener(IPAddress.Any, 8001);
            _tcpServer.Start();
            _dbContextFactory = dbContextFactory;
            _dataCacheService = dataCacheService;
            _dataAnalyses = dataAnalyses;
            _iotService = iotService;
            _configuration = configuration;
            _ = bool.TryParse(_configuration.GetSection("AutoCommit").Value, out var autoCommit);
            _autoCommit = autoCommit;
        }

        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
        {
            communications = Initial(_dataCacheService.Equipments.Select(x => x.CommunicationConfig).Distinct().ToList()!);
            _tcpServer.BeginAcceptTcpClient(DoAcceptTcpClient, _tcpServer);

            //采集数据
            foreach (var item in communications)
            {
                _ = Task.Run(async () =>
                {
                    var _actionA = DateTime.MinValue;
                    do
                    {
                        try
                        {
                            await Task.Delay(500);
                            var equipmentTemps = _dataCacheService.Equipments.Where(x => x.CommunicationConfig?.Id == item.CommunicationId).ToList();
                            var result = item.Read(equipmentTemps.SelectMany(x => x.EquipmentProperties));
                            if (!result.Success)
                            {
                                //超过1分钟未更新数据，则清空Value值
                                var temps = equipmentTemps.SelectMany(x => x.EquipmentProperties).Where(x => x.UpdateTime < DateTime.Now.AddSeconds(-30) && !string.IsNullOrWhiteSpace(x.Value)).ToList();
                                temps.ForEach(x => x.Value = string.Empty);
                                systemLog.LogError($"读取设备[{item.IpAddress}]数据失败，{result.Msg}");
                                continue;
                            }

                            if ((DateTime.Now - _actionA).TotalSeconds >= 1)
                            {
                                _actionA = DateTime.Now;
                                var records = equipmentTemps.Where(x => !string.IsNullOrWhiteSpace(x.EquipmentSN) && EquipmentCodes.Contains(x.Code)).Select(x => new EquipmentDataQueue
                                {
                                    EquipmentCode = x.EquipmentSN ?? string.Empty,
                                    EquipmentName = x.Name,
                                    EquipmentTypeCode = x.EquipmentType.ToString(),
                                    IsCommit = false,
                                    Reported = JsonSerializer.Serialize(x.EquipmentProperties.Select(p => new TagItem
                                    {
                                        Tag = p.Code,//这里的Tag使用属性Code，不同焊接设备地址数据格式不同，但需要采集的数据都是一样的
                                        Value = p.Value ?? string.Empty
                                    }).ToList()),
                                    SourceTimestamp = DateTimeOffset.Now.ToUnixTimeSeconds(),
                                    Version = 1,
                                    CreateTime = DateTime.Now,
                                }).ToList();

                                foreach (var item in records)
                                {
                                    _equipmentDataQueues.Enqueue(item);
                                }
                            }
                        }
                        catch (Exception ex)
                        {
                            systemLog.LogError($"读取IP：{item.IpAddress}设备出现异常：{ex.Message}");
                        }
                    } while (!stoppingToken.IsCancellationRequested);
                }, stoppingToken);
            };

            SendEquipmentDataHandle(stoppingToken);

            _ = Task.Run(async () =>
            {
                while (!stoppingToken.IsCancellationRequested)
                {
                    var result = _iotService.UpdateClientStatus();
                    if (!result.Success)
                    {
                        systemLog.LogError($"更新客户端状态失败：{result.Msg}");
                    }
                    await Task.Delay(TimeSpan.FromMinutes(1));
                }
            }, stoppingToken);

            ClearRecords();

            while (!stoppingToken.IsCancellationRequested)
            {
                try
                {
                    var dataAnalysesTasks = new List<Task>();
                    //数据解析
                    foreach (var item in _dataAnalyses)
                    {
                        dataAnalysesTasks.Add(Task.Run(() =>
                        {
                            item.Analysis(communications, _dataCacheService.Equipments);
                        }, stoppingToken));
                    }
                    Task.WaitAll(dataAnalysesTasks.ToArray(), stoppingToken);
                }
                catch (Exception ex)
                {
                    systemLog.LogError($"后台服务“{nameof(ExecuteAsync)}”方法出现异常：{ex.Message}");
                }
                await Task.Delay(500, stoppingToken);
            }
            CloesAllConnection(communications);
        }

        private void ClearRecords()
        {
            try
            {
                var endTime = DateTime.Today.AddDays(1 - DateTime.Today.Day).AddMonths(-2);
                var context = _dbContextFactory.CreateDbContext();
                var count = context.EquipmentPropertyRecords.Where(x => x.CreateTime < endTime).ExecuteDelete();
                if (count > 0)
                {
                    systemLog.LogInfo($"成功清除{count}条历史记录数据");
                }

                var productIds = context.WorkpieceProductions.Where(x => x.CreateTime < endTime).Select(x => x.Id).ToList();
                context.WeldProcessRecords.Where(x => productIds.Contains(x.ProductionId)).ExecuteDelete();
                var productCount = context.WorkpieceProductions.Where(x => productIds.Contains(x.Id)).ExecuteDelete();
                if (productCount > 0)
                {
                    systemLog.LogInfo($"成功清除{productCount}条工件记录数据");
                }
            }
            catch (Exception ex)
            {
                systemLog.LogError($"清理历史数据出现异常：{ex.Message}");
            }
        }

        private void DoAcceptTcpClient(IAsyncResult result)
        {
            var server = (TcpListener)result.AsyncState!;
            var tcpClient = server.EndAcceptTcpClient(result);
            var endPoint = tcpClient.Client.RemoteEndPoint?.ToString();
            systemLog.LogSuccess($"[客户端[{endPoint}]已成功建立连接");
            Receive(tcpClient);
            server.BeginAcceptTcpClient(DoAcceptTcpClient, server);
        }

        private void Receive(TcpClient tcpClient)
        {
            Task.Run(() =>
            {
                while (tcpClient.Connected)
                {
                    try
                    {
                        var buffer = new byte[8192];
                        var stream = tcpClient.GetStream();
                        stream.Read(buffer, 0, buffer.Length);
                        if (buffer.Length == 0) continue;

                        var bufferString = Encoding.Default.GetString(buffer);

                        const string kukaAckFlag1 = "<robot><ack>";
                        const string kukaAckFlag2 = "<ack><robot>";
                        if (bufferString.StartsWith(kukaAckFlag1) && bufferString.EndsWith(kukaAckFlag2))
                        {
                            ReceiveKukaData(bufferString);
                        }

                        //第一位等于02，且第二位等于01，则是设备推送给上位机的报文；否则，不做任何处理
                        if (buffer[0].ToString("X2") != "02" || buffer[1].ToString("X2") != "01" || buffer[126] != 0x0D || buffer[127] != 0x0A)
                        {
                            continue;
                        }
                        //_rgvService.Analysis(tcpClient, _equipmentExecutorService.Equipments, buffer);
                    }
                    catch (Exception ex)
                    {
                        systemLog.LogError(ex.Message);
                    }
                }
            });
        }

        private List<ICommunication> Initial(IList<CommunicationConfig> communicationConfigs)
        {
            var result = new List<ICommunication>();
            try
            {
                foreach (var item in communicationConfigs)
                {
                    ICommunication communication;
                    switch (item.CommunicationType)
                    {
                        case CommunicationTypeConst.None:
                        case CommunicationTypeConst.TcpServer:
                            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.ModbusTcp:
                            communication = new ModbusTcpCommunication(item.Id, item.IpAddress, item.Port);
                            result.Add(communication);
                            break;
                        case CommunicationTypeConst.TcpClient:
                            break;
                        default:
                            break;
                    }
                }
            }
            catch (Exception ex)
            {
                systemLog.LogError($"设备通讯初始化异常：{ex.Message}");
            }
            return result;
        }

        /// <summary>
        /// 接收库卡数据
        /// </summary>
        /// <param name="bufferString"></param>
        private void ReceiveKukaData(string bufferString)
        {
            var bufferData = bufferString.Split(',').ToArray();
            if (bufferString.Length == 26)
            {
                var kukaCode = bufferData[0];
                var kukaEquipment = _dataCacheService.Equipments.Where(x => x.Code == kukaCode).FirstOrDefault();
                if (kukaEquipment == null) return;
                //赋值
                kukaEquipment[RobotProps.BootFlag.ToString()].Value = bufferData[0];
                kukaEquipment[RobotProps.WorkFlag.ToString()].Value = bufferData[0];
                kukaEquipment[RobotProps.WeldFlag.ToString()].Value = bufferData[0];
                kukaEquipment[RobotProps.WeldCompleteFlag.ToString()].Value = bufferData[0];
                kukaEquipment[RobotProps.Weld_V.ToString()].Value = bufferData[0];
                kukaEquipment[RobotProps.Weld_I.ToString()].Value = bufferData[0];
                kukaEquipment[RobotProps.Weld_Speed.ToString()].Value = bufferData[0];
                kukaEquipment[RobotProps.Weld_Gas.ToString()].Value = bufferData[0];
                kukaEquipment[RobotProps.Weld_CleanGun.ToString()].Value = bufferData[0];
                kukaEquipment[RobotProps.Alarm.ToString()].Value = bufferData[0];
                kukaEquipment[RobotProps.Work_Time.ToString()].Value = bufferData[0];
                kukaEquipment[RobotProps.Work_Mode.ToString()].Value = bufferData[0];
                kukaEquipment[RobotProps.Type.ToString()].Value = bufferData[0];
                kukaEquipment[RobotProps.Pos_X.ToString()].Value = bufferData[0];
                kukaEquipment[RobotProps.Pos_Y.ToString()].Value = bufferData[0];
                kukaEquipment[RobotProps.Pos_Z.ToString()].Value = bufferData[0];
                kukaEquipment[RobotProps.Pos_A.ToString()].Value = bufferData[0];
                kukaEquipment[RobotProps.Pos_B.ToString()].Value = bufferData[0];
                kukaEquipment[RobotProps.Pos_C.ToString()].Value = bufferData[0];
                kukaEquipment[RobotProps.Pos_E1.ToString()].Value = bufferData[0];
                kukaEquipment[RobotProps.Pos_E2.ToString()].Value = bufferData[0];
                kukaEquipment[RobotProps.Pos_E3.ToString()].Value = bufferData[0];
                kukaEquipment[RobotProps.Pos_E4.ToString()].Value = bufferData[0];
                kukaEquipment[RobotProps.Program_No.ToString()].Value = bufferData[0];

                //时间赋值
                foreach (var item in kukaEquipment.EquipmentProperties)
                {
                    item.UpdateTime = DateTime.Now;
                }
            }
        }

        private static void CloesAllConnection(IEnumerable<ICommunication> communications)
        {
            foreach (var item in communications)
            {
                item.ConnectClose();
            }
        }

        public void SendEquipmentDataHandle(CancellationToken stoppingToken)
        {
            _ = Task.Run(async () =>
            {
                const int limit = 100;
                var lastExecute = DateTime.MinValue;
                while (!stoppingToken.IsCancellationRequested)
                {
                    try
                    {
                        //自动推送数据配置
                        await Task.Delay(1000);
                        var cacheData = new List<EquipmentDataQueue>();
                        for (int i = 0; i < limit; i++)
                        {
                            _equipmentDataQueues.TryDequeue(out var temp);
                            if (temp == null)
                            {
                                break;
                            }
                            cacheData.Add(temp);
                        }

                        using var context = _dbContextFactory.CreateDbContext();
                        if (cacheData.Count > 0)
                        {
                            //无需推送或数据库存在未提交的数据，存入数据库
                            if (!_autoCommit || context.EquipmentDataQueues.Where(x => !x.IsCommit).Any())
                            {
                                context.EquipmentDataQueues.AddRange(cacheData);
                                var count = context.SaveChanges();
                                systemLog.LogSuccess($"数据库新增{count}条设备数据记录");
                            }
                            //需要推送，且数据库无待推送数据
                            else
                            {
                                var stopwatch = Stopwatch.StartNew();
                                var result = _iotService.SendEquipmentData(cacheData);
                                stopwatch.Stop();
                                if (result.Success)
                                {
                                    systemLog.LogSuccess($"推送数据至IOTCloud成功，数量：{cacheData.Count}，耗时：{stopwatch.ElapsedMilliseconds}ms");
                                }
                                else
                                {
                                    //推送失败，保存至数据库
                                    context.EquipmentDataQueues.AddRange(cacheData);
                                    var count = context.SaveChanges();
                                    systemLog.LogSuccess($"数据库新增{count}条设备数据记录");
                                }
                            }
                        }

                        //默认5秒推送一次
                        var timeSpan = TimeSpan.FromSeconds(5);
                        if (context.EquipmentDataQueues.Where(x => !x.IsCommit).Count() >= limit)
                        {
                            //数据较多时，一秒推送一次
                            timeSpan = TimeSpan.FromSeconds(1);
                        }

                        //自动推送处于启用状态，则推送
                        if (_autoCommit && (DateTime.Now - lastExecute) >= timeSpan)
                        {
                            lastExecute = DateTime.Now;
                            var queues = context.EquipmentDataQueues.Where(x => !x.IsCommit).OrderBy(x => x.SourceTimestamp).Take(limit).ToList();
                            if (queues.Count == 0)
                            {
                                continue;
                            }
                            //推送数据
                            var stopwatch = Stopwatch.StartNew();
                            var result = _iotService.SendEquipmentData(queues);
                            stopwatch.Stop();
                            if (result.Success)
                            {
                                context.EquipmentDataQueues.RemoveRange(queues);
                                context.SaveChanges();
                                systemLog.LogSuccess($"推送数据至IOTCloud成功，数量：{queues.Count}，耗时：{stopwatch.ElapsedMilliseconds}ms");
                            }
                            else
                            {
                                systemLog.LogError($"数据推送失败：{result.Msg}");
                            }
                        }
                    }
                    catch (Exception ex)
                    {
                        systemLog.LogError($"数据推送线程异常.{ex.Message}");
                    }
                }
            }, stoppingToken);
        }

        /// <summary>
        /// 需要推送数据的设备编号
        /// </summary>
        private static List<string> EquipmentCodes => new List<EquipmentConst>
        {
            EquipmentConst.KukaC4_1,
            EquipmentConst.KukaC4_2,
            EquipmentConst.KukaC2_1,
            EquipmentConst.KukaC2_2,
            EquipmentConst.Efort_1,
            EquipmentConst.Fanuc_1,
        }.Select(x => x.ToString()).ToList();
    }
}