MES / DataAcquisition · Files

using System.Collections.ObjectModel;
using System.Configuration;
using System.Diagnostics;
using System.Windows;
using System.Windows.Controls;
using CommunityToolkit.Mvvm.ComponentModel;
using CommunityToolkit.Mvvm.Input;
using FreeSql;
using HHECS.DAQClient.Common;
using HHECS.DAQClient.Model;
using HHECS.DAQClient.Services;
using HHECS.DAQClient.View;
using HHECS.DAQClient.View.CommunicationView;
using HHECS.DAQClient.View.EquipmentView;
using HHECS.DAQShared.Models;
using MessageBox = HandyControl.Controls.MessageBox;

namespace HHECS.DAQClient.ViewModel
{
    internal partial class MainVM : ObservableObject
    {
        [ObservableProperty]
        private bool btnStartEnabled = true;

        [ObservableProperty]
        private bool btnStopEnabled = false;

        [ObservableProperty]
        private ObservableCollection<LogModel> logModels = new();

        [ObservableProperty]
        private Page communicationPage = new CommunicationPage();

        [ObservableProperty]
        private Page equipmentPage = new EquipmentPage();

        [ObservableProperty]
        private Page equipmentDataQueuePage = new EquipmentDataQueuePage();

        [ObservableProperty]
        private int queueCount;

        /// <summary>
        /// 自动上传数据
        /// </summary>
        [ObservableProperty]
        private bool autoCommit = true;

        /// <summary>
        /// 数据压缩
        /// </summary>
        /// <remarks>为True时，对于持续不变的数据，记录开始和结束时间再上传</remarks>
        [ObservableProperty]
        private bool dataCompression;

        /// <summary>
        /// 每次提交的数据量
        /// </summary>
        private int _commitCount = 100;
        private readonly SystemLog _log = SystemLog.GetInstance();
        private readonly CenterService _centerService;
        private readonly HttpService _httpService;
        private readonly IFreeSql _freeSql;

        public MainVM(IFreeSql freeSql, CenterService centerService, HttpService httpService)
        {
            _centerService = centerService;
            _httpService = httpService;
            _freeSql = freeSql;
            Initial();
        }

        private void Initial()
        {
            //SyncEquipmentDataToIotCloud();
            var commitCountConfig = ConfigurationManager.AppSettings["CommitCount"];
            _ = bool.TryParse(ConfigurationManager.AppSettings["AutoExecute"], out var autoExecute);
            _ = bool.TryParse(ConfigurationManager.AppSettings["DataCompression"], out var _dataCompression);
            _ = int.TryParse(commitCountConfig, out _commitCount);
            DataCompression = _dataCompression;
            if (_commitCount <= 0)
            {
                //未配置，则默认100
                _commitCount = 100;
            }
            RefreshLog();
            UploadEquipmentDataToCloud();
            UpdateClientStatus();
            //UpdateEquipmentStatus();
            if (autoExecute)
            {
                Start();
            }
        }

        /// <summary>
        /// 上传数据至IOTClound
        /// </summary>
        private void UploadEquipmentDataToCloud()
        {
            Task.Run(async () =>
            {
                //缓存时间，分钟
                const int cacheMinutesTime = 3;
                var _lastAExecution = DateTime.MinValue;
                while (true)
                {
                    try
                    {
                        await Task.Delay(1000);
                        var haveRecord = _freeSql.Queryable<EquipmentDataQueue>().Any();
                        if (!_centerService.EquipmentDataQueues.IsEmpty)
                        {
                            var temps = new List<EquipmentDataQueue>();
                            var commitFailureRecords = new List<EquipmentDataQueue>();
                            //启用数据压缩
                            if (DataCompression)
                            {
                                while (!_centerService.EquipmentDataQueues.IsEmpty)
                                {
                                    var result = _centerService.EquipmentDataQueues.TryDequeue(out var item);
                                    if (item != null)
                                    {
                                        temps.Add(item);
                                    }
                                }
                            }
                            //未启用数据压缩
                            else
                            {
                                for (int i = 0; i < _commitCount; i++)
                                {
                                    var result = _centerService.EquipmentDataQueues.TryDequeue(out var item);
                                    if (!result) break;//队列已清空
                                    if (item != null)
                                    {
                                        temps.Add(item);
                                    }
                                }
                            }

                            //自动上传启用，且数据库无未上传的数据，则直接推送
                            if (AutoCommit && !DataCompression && !haveRecord)
                            {
                                var tasks = new List<Task<List<EquipmentDataQueue>>>();
                                foreach (var item in temps.GroupBy(x => x.EquipmentCode))
                                {
                                    tasks.Add(Task.Run(() =>
                                    {
                                        Stopwatch stopwatch = Stopwatch.StartNew();
                                        var records = item.OrderBy(x => x.SourceTimestamp).ToList();
                                        var result = _httpService.SendEquipmentData(records);
                                        if (!result.Success)
                                        {
                                            _log.LogError($"推送设备[{item.Key}]数据失败，{result.Msg}，耗时：{stopwatch.ElapsedMilliseconds}ms");
                                            return records;
                                        }
                                        _log.LogSuccess($"成功推送{records.Count}条设备[{item.Key}]数据，耗时：{stopwatch.ElapsedMilliseconds}ms");
                                        return new List<EquipmentDataQueue>();
                                    }));
                                }
                                Task.WaitAll(tasks.ToArray());
                                commitFailureRecords = tasks.SelectMany(x => x.Result).ToList();
                            }
                            //自动上传关闭或数据库存在未上传的记录，则需要存入数据库
                            else
                            {
                                commitFailureRecords.AddRange(temps);
                            }

                            //将上传失败的数据存入数据库
                            if (commitFailureRecords.Count > 0)
                            {
                                //节流模式
                                if (DataCompression)
                                {
                                    var dictionaryUpdateTemps = new Dictionary<string, EquipmentDataQueue>();
                                    var dictionaryAddTemps = new Dictionary<string, List<EquipmentDataQueue>>();
                                    foreach (var item in commitFailureRecords.GroupBy(x => x.EquipmentCode))
                                    {
                                        var lastRecord = _freeSql.Queryable<EquipmentDataQueue>().Where(x => x.EquipmentCode == item.Key).OrderByDescending(x => x.SourceTimestamp).First();
                                        var addTemps = new List<EquipmentDataQueue>();
                                        dictionaryAddTemps[item.Key] = addTemps;//传递集合引用
                                        foreach (var record in item.OrderBy(x => x.SourceTimestamp).ToList())
                                        {
                                            var lastTemp = lastRecord;
                                            if (addTemps.Count > 0)
                                            {
                                                lastTemp = addTemps.OrderByDescending(x => x.SourceTimestamp).First();
                                            }

                                            if (lastTemp != null)
                                            {
                                                var currentRecordTime = DateTimeOffset.FromUnixTimeMilliseconds(record.SourceTimestamp).LocalDateTime;
                                                var lastRecordTime = (lastTemp.Updated ?? lastTemp.Created).Value;
                                                if (lastTemp.Reported == record.Reported && currentRecordTime.Date == lastRecordTime.Date
                                                && (currentRecordTime - lastRecordTime).TotalSeconds <= 5)
                                                {
                                                    lastTemp.Updated = currentRecordTime;

                                                    //是数据库里面的数据，有变化需要更新
                                                    if (lastTemp.Id != Guid.Empty && lastTemp.Id == lastRecord.Id)
                                                    {
                                                        dictionaryUpdateTemps[item.Key] = lastTemp;
                                                    }
                                                    continue;
                                                }
                                            }
                                            //其他情况，新增记录
                                            addTemps.Add(record);
                                        }
                                    }

                                    var allUpdateTemps = dictionaryUpdateTemps.Select(x => x.Value).ToList();
                                    if (allUpdateTemps.Count > 0)
                                    {
                                        _freeSql.Update<EquipmentDataQueue>().SetSource(allUpdateTemps).UpdateColumns(x => x.Updated).ExecuteAffrows();
                                        _log.LogInfo($"更新[{allUpdateTemps.Count}]条设备记录");
                                    }

                                    var allAddTemps = dictionaryAddTemps.SelectMany(x => x.Value).ToList();
                                    if (allAddTemps.Count > 0)
                                    {
                                        _freeSql.Insert(allAddTemps).ExecuteAffrows();
                                        _log.LogInfo($"新增{allAddTemps.Count}条设备记录");
                                    }
                                }
                                //实时存储
                                else
                                {
                                    _freeSql.Insert(commitFailureRecords).ExecuteAffrows();
                                    _log.LogInfo($"新增{commitFailureRecords.Count}条数据记录");
                                }
                                //更新状态
                                haveRecord = _freeSql.Queryable<EquipmentDataQueue>().Any();
                            }
                        }

                        var timeSpan = TimeSpan.FromSeconds(5);//默认5秒一次
                        var equipmentTotal = _freeSql.Queryable<Equipment>().Where(x => !x.Disable).Count();
                        var recordsTotal = _freeSql.Queryable<EquipmentDataQueue>().Count();
                        if (recordsTotal >= equipmentTotal * _commitCount)
                        {
                            //数据较多时，1秒一次
                            timeSpan = TimeSpan.FromSeconds(1);
                        }

                        //推送数据
                        if (AutoCommit && (DateTime.Now - _lastAExecution) >= timeSpan && haveRecord)
                        {
                            _lastAExecution = DateTime.Now;
                            var equipmentCodes = _freeSql.Queryable<EquipmentDataQueue>().Distinct().ToList(x => x.EquipmentCode);
                            var commitSuccessRecordIds = new List<Guid>();

                            //是否存在更新时间为空的记录，最后一条记录不算
                            var haveUpdateIsNull = equipmentCodes.Where(equipmentCode =>
                            {
                                //第一条更新时间为空的记录
                                var firstTemp = _freeSql.Queryable<EquipmentDataQueue>().Where(x => x.EquipmentCode == equipmentCode && x.Updated == null).OrderBy(x => x.SourceTimestamp).First(x => x.Id);
                                if (firstTemp == default)
                                {
                                    //无记录
                                    return false;
                                }
                                //最后一条记录
                                var lastTemp = _freeSql.Queryable<EquipmentDataQueue>().Where(x => x.EquipmentCode == equipmentCode).OrderByDescending(x => x.SourceTimestamp).First(x => x.Id);
                                //第一条与最后一条是同一条，则排除
                                if (firstTemp == lastTemp)
                                {
                                    return false;
                                }
                                //存在多条
                                return true;
                            }).Any();

                            //节流模式
                            if (DataCompression && !haveUpdateIsNull)
                            {
                                //超过设定时间的数据
                                var records = _freeSql.Queryable<EquipmentDataQueue>().Where(x => DateTime.Now.AddMinutes(-(cacheMinutesTime + 1)) >= x.Created).OrderBy(x => x.SourceTimestamp).Take(_commitCount).OrderBy(x => x.Created).ToList();
                                if (records.Count == 0)
                                {
                                    continue;
                                }
                                var stopwatch = Stopwatch.StartNew();
                                var result = _httpService.SendEquipmentDataV2(records);
                                var currentEquipmentCodes = records.Select(x => x.EquipmentCode).Distinct().ToList();
                                if (!result.Success)
                                {
                                    _log.LogError($"推送设备[{string.Join(',', currentEquipmentCodes)}]数据失败，{result.Msg}，耗时：{stopwatch.ElapsedMilliseconds}ms");
                                    continue;
                                }
                                _log.LogSuccess($"成功推送{records.Count}条设备[{string.Join(',', currentEquipmentCodes)}]数据，耗时：{stopwatch.ElapsedMilliseconds}ms");
                                commitSuccessRecordIds = records.Select(x => x.Id).ToList();
                                _freeSql.Delete<EquipmentDataQueue>().Where(x => commitSuccessRecordIds.Contains(x.Id)).ExecuteAffrows();
                                continue;
                            }

                            //推送成功的数据集合
                            var tasks = new List<Task<List<EquipmentDataQueue>>>();
                            foreach (var equipmentCode in equipmentCodes)
                            {
                                tasks.Add(Task.Run(() =>
                                {
                                    Stopwatch stopwatch = Stopwatch.StartNew();
                                    var successRecords = new List<EquipmentDataQueue>();
                                    var records = _freeSql.Queryable<EquipmentDataQueue>().Where(x => x.EquipmentCode == equipmentCode && !x.IsCommit).OrderBy(x => x.SourceTimestamp).Take(_commitCount).ToList();
                                    if (records.Count == 0)
                                    {
                                        return successRecords;
                                    }
                                    var temp1s = new List<List<EquipmentDataQueue>>();
                                    var temp2s = new List<EquipmentDataQueue>();
                                    foreach (var record in records)
                                    {
                                        //是最后一条记录，且启用了节流模式
                                        if (DataCompression && record == records.Last())
                                        {
                                            //创建时间距离现在的时间未超过设定时间，则不推送
                                            if ((DateTime.Now - record.Created.Value).TotalMinutes < cacheMinutesTime)
                                            {
                                                break;
                                            }

                                            //队列还有缓存的数据，且持续时间未达到设定值，则不推送
                                            if (!_centerService.EquipmentDataQueues.IsEmpty && record.Updated != null
                                            && (record.Updated - record.Created).Value.TotalMinutes < cacheMinutesTime)
                                            {
                                                break;
                                            }
                                        }

                                        if (temp2s.Count == 0)
                                        {
                                            temp2s.Add(record);
                                            continue;
                                        }

                                        var lastItem = temp2s.Last();
                                        var lastType = lastItem.Updated > lastItem.Created;
                                        var currentType = record.Updated > record.Created;
                                        if (lastType == currentType)
                                        {
                                            temp2s.Add(record);
                                        }
                                        else
                                        {
                                            temp1s.Add(temp2s);
                                            temp2s = new List<EquipmentDataQueue>
                                            {
                                                    record
                                            };
                                        }
                                    }
                                    if (temp2s.Count > 0)
                                    {
                                        temp1s.Add(temp2s);
                                        temp2s = new List<EquipmentDataQueue>();
                                    }

                                    foreach (var item in temp1s)
                                    {
                                        //某段时间的数据
                                        if (item.All(x => x.Updated > x.Created))
                                        {
                                            var result = _httpService.SendEquipmentDataV2(item);
                                            if (!result.Success)
                                            {
                                                _log.LogError($"推送设备[{equipmentCode}]数据失败，{result.Msg}，耗时：{stopwatch.ElapsedMilliseconds}ms");
                                                break;//推送失败，后续的数据也暂停推送
                                            }
                                            _log.LogSuccess($"成功推送{records.Count}条设备[{equipmentCode}]数据，耗时：{stopwatch.ElapsedMilliseconds}ms");
                                            successRecords.AddRange(item);
                                            continue;
                                        }

                                        //一秒一条记录的数据
                                        var result2 = _httpService.SendEquipmentData(item);
                                        if (!result2.Success)
                                        {
                                            _log.LogError($"推送设备[{equipmentCode}]数据失败，{result2.Msg}，耗时：{stopwatch.ElapsedMilliseconds}ms");
                                            break;//推送失败，后续的数据也暂停推送
                                        }
                                        _log.LogSuccess($"成功推送1条设备[{equipmentCode}]数据，耗时：{stopwatch.ElapsedMilliseconds}ms");
                                        successRecords.AddRange(item);
                                    }
                                    return successRecords;
                                }));
                            }
                            Task.WaitAll(tasks.ToArray());
                            commitSuccessRecordIds = tasks.SelectMany(x => x.Result).Select(x => x.Id).ToList();
                            _freeSql.Delete<EquipmentDataQueue>().Where(x => commitSuccessRecordIds.Contains(x.Id)).ExecuteAffrows();
                        }
                    }
                    catch (Exception ex)
                    {
                        _log.LogException($"数据上传线程异常：{ex.Message}");
                    }
                }
            });
        }

        private void UpdateClientStatus()
        {
            Task.Run(async () =>
            {
                while (true)
                {
                    await Task.Delay(TimeSpan.FromSeconds(1));
                    if (BtnStartEnabled)
                    {
                        continue;
                    }
                    _ = Guid.TryParse(ConfigurationManager.AppSettings["ClientId"], out var clientId);
                    if (clientId == default)
                    {
                        _log.LogError($"更新客户端状态失败：当前程序[ClientId]未配置！");
                        continue;
                    }

                    var result = _httpService.UpdateClientStatus(clientId);
                    if (!result.Success)
                    {
                        _log.LogError($"更新客户端状态失败：{result.Msg}");
                    }
                    else
                    {
                        //更新成功，则等待
                        await Task.Delay(TimeSpan.FromMinutes(1));
                    }
                }
            });
        }

        [RelayCommand]
        public void Start()
        {
            _ = bool.TryParse(ConfigurationManager.AppSettings["PrivacyInfoStatus"], out var privacyInfoStatus);
            if (!privacyInfoStatus)
            {
                var vm = new PrivacyInfoView();
                var result = vm.ShowDialog();
                if (result != true)
                {
                    return;
                }
            }

            BtnStartEnabled = false;
            BtnStopEnabled = true;
            _centerService.Start();
        }

        [RelayCommand]
        public void Stop()
        {
            BtnStartEnabled = true;
            BtnStopEnabled = false;
            _centerService.Stop();
        }

        private void RefreshLog()
        {
            Task.Run(() =>
            {
                Application.Current.Dispatcher.Invoke(async () =>
                {
                    do
                    {
                        while (!_log.IsEmpty)
                        {
                            QueueCount = _centerService.EquipmentDataQueues.Count;
                            var log = _log.GetLog();
                            if (log == null)
                            {
                                await Task.Delay(100);
                                continue;
                            }

                            var oldItem = LogModels.Where(x => x.Messages.Equals(log.Messages)).FirstOrDefault();
                            if (oldItem != null)
                            {
                                oldItem.CreateTime = log.CreateTime;
                                LogModels = new ObservableCollection<LogModel>(LogModels.OrderByDescending(x => x.CreateTime));
                            }
                            else
                            {
                                LogModels.Insert(0, new LogModel()
                                {
                                    LogType = log.LogType,
                                    Messages = log.Messages,
                                    CreateTime = log.CreateTime
                                });
                            }
                            if (LogModels.Count > 50)
                            {
                                LogModels.Remove(LogModels.Last());
                            }
                        }
                        await Task.Delay(100);
                    } while (true);
                });
            });
        }

        /// <summary>
        /// 将本地设备数据同步至IOT云平台
        /// </summary>
        /// <remarks>数据库升级，数据变更</remarks>
        private void SyncEquipmentDataToIotCloud()
        {
            var connectionString = "Data Source=172.16.29.88;Initial Catalog=IoT;Persist Security Info=True;User ID=sa;Password=1qaz!QAZ;Trust Server Certificate=True";//云平台数据库连接字符串
            var iotCloundFreeSql = new FreeSqlBuilder().UseConnectionString(DataType.SqlServer, connectionString)
                .UseAutoSyncStructure(false).Build();

            var localEquipments = _freeSql.Queryable<Equipment>().ToList();
            foreach (var item in localEquipments)
            {
                var iotEquipmentId = iotCloundFreeSql.Queryable<Equipment>().Where(x => x.Code == item.Code).First(x => x.Id);
                var iotEquipmentProps = iotCloundFreeSql.Queryable<EquipmentProp>().Where(x => x.EquipmentId == iotEquipmentId).ToList();
                var localEquipmentProps = _freeSql.Queryable<EquipmentProp>().Where(x => x.EquipmentId == item.Id).ToList();
                foreach (var iotPropItem in iotEquipmentProps)
                {
                    var localProp = localEquipmentProps.Where(x => x.Code == iotPropItem.Code).FirstOrDefault();
                    if (localProp == null)
                    {
                        //本地没有，则删除
                        iotCloundFreeSql.Delete<EquipmentProp>().Where(x => x.Id == iotPropItem.Id).ExecuteAffrows();
                        continue;
                    }

                    //本地存在，更新Address地址；
                    iotCloundFreeSql.Update<EquipmentProp>().Set(x => x.Address == localProp.Address).ExecuteAffrows();
                }
            }
        }
    }
}