JiShe.CollectBus/services/JiShe.CollectBus.Application/ScheduledMeterReading/EnergySystemScheduledMeterReadingService.cs

using JiShe.CollectBus.Application.Contracts;
using JiShe.CollectBus.Common.BuildSendDatas;
using JiShe.CollectBus.Common.Consts;
using JiShe.CollectBus.Common.DeviceBalanceControl;
using JiShe.CollectBus.Common.Encrypt;
using JiShe.CollectBus.Common.Enums;
using JiShe.CollectBus.Common.Extensions;
using JiShe.CollectBus.Common.Helpers;
using JiShe.CollectBus.FreeSql;
using JiShe.CollectBus.GatherItem;
using JiShe.CollectBus.IoTDB.Context;
using JiShe.CollectBus.IoTDB.Interface;
using JiShe.CollectBus.IotSystems.Ammeters;
using JiShe.CollectBus.IotSystems.MeterReadingRecords;
using JiShe.CollectBus.IotSystems.Watermeter;
using JiShe.CollectBus.Kafka.Internal;
using JiShe.CollectBus.Kafka.Producer;
using JiShe.CollectBus.Protocol.Contracts.Interfaces;
using JiShe.CollectBus.Protocol.Contracts.Models;
using JiShe.CollectBus.Protocol.Contracts.SendData;
using Microsoft.AspNetCore.Authorization;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;


namespace JiShe.CollectBus.ScheduledMeterReading
{
    /// <summary>
    /// 能耗系统定时采集服务
    /// </summary>
    [AllowAnonymous]
    //[Route($"/energy/app/scheduled")]
    public class EnergySystemScheduledMeterReadingService : BasicScheduledMeterReadingService
    {
        string serverTagName = string.Empty;
        private readonly ILogger<BasicScheduledMeterReadingService> _logger;
        private readonly IIoTDbProvider _dbProvider;
        private readonly IProtocolService _protocolService;

        public EnergySystemScheduledMeterReadingService(
            ILogger<EnergySystemScheduledMeterReadingService> logger,
            IIoTDbProvider dbProvider,
            IOptions<KafkaOptionConfig> kafkaOptions,
            IoTDBRuntimeContext runtimeContext,
            IProducerService producerService,
            IProtocolService protocolService,
            IRedisDataCacheService redisDataCacheService)
            : base(logger,
                  producerService,
                  redisDataCacheService,
                  dbProvider,
                  runtimeContext,
                  protocolService,
                  kafkaOptions)
        {
            serverTagName = kafkaOptions.Value.ServerTagName;
            _dbProvider = dbProvider;
            _logger = logger;
            _protocolService = protocolService;
        }

        public sealed override string SystemType => SystemTypeConst.Energy;

        public sealed override string ServerTagName => serverTagName;

        /// <summary>
        /// 获取采集项列表
        /// </summary>
        /// <returns></returns>
        public override async Task<List<GatherItemInfo>> GetGatherItemByDataTypes()
        {
            try
            {
                string sql = $"SELECT DataType,ItemCode FROM TB_GatherItem(NOLOCK) WHERE [State]=0";
                return await SqlProvider.Instance.Change(DbEnum.EnergyDB)
               .Ado
               .QueryAsync<GatherItemInfo>(sql, null);
            }
            catch
            {
                return null;
            }
        }

        /// <summary>
        /// 获取电表信息
        /// </summary>
        /// <param name="gatherCode">采集端Code</param>
        /// <returns></returns>
        //[HttpGet]
        //[Route($"ammeter/list")]
        public override async Task<List<AmmeterInfo>> GetAmmeterInfoList(string gatherCode = "V4-Gather-8890")
        {

            //List<AmmeterInfo> ammeterInfos = new List<AmmeterInfo>();
            //ammeterInfos.Add(new AmmeterInfo()
            //{
            //    Baudrate = 2400,
            //    FocusAddress = "402440506",
            //    Name = "张家祠工务(三相电表)",
            //    FocusId = 95780,
            //    DatabaseBusiID = 1,
            //    MeteringCode = 1,
            //    AmmerterAddress = "402410040506",
            //    MeterId = 127035,
            //    TypeName = 3,
            //    DataTypes = "449,503,581,582,583,584,585,586,587,588,589,590,591,592,593,594,597,598,599,600,601,602,603,604,605,606,607,608,661,663,677,679",
            //    TimeDensity = 15,
            //    BrandType = "",
            //});
            //ammeterInfos.Add(new AmmeterInfo()
            //{
            //    Baudrate = 2400,
            //    FocusAddress = "542400504",
            //    Name = "五号配（长芦二所四排）(单相电表)",
            //    FocusId = 69280,
            //    DatabaseBusiID = 1,
            //    MeteringCode = 2,
            //    AmmerterAddress = "542410000504",
            //    MeterId = 95594,
            //    TypeName = 1,
            //    DataTypes = "581,589,592,597,601",
            //    TimeDensity = 15,
            //    BrandType = "",
            //});

            //return ammeterInfos;

            string sql = $@"SELECT C.ID as MeterId,C.Name,C.FocusID as FocusId,C.SingleRate,C.MeteringCode,C.Code AS BrandType,C.Baudrate,C.Password,C.MeteringPort,C.[Address] AS AmmerterAddress,C.TypeName,C.Protocol,C.TripState,C.[State],B.[Address],B.AreaCode,B.AutomaticReport,D.DataTypes,B.TimeDensity,A.GatherCode,C.Special,C.[ProjectID],B.AbnormalState,B.LastTime,CONCAT(B.AreaCode, B.[Address]) AS FocusAddress,(select top 1 DatabaseBusiID from TB_Project where ID = B.ProjectID) AS DatabaseBusiID
              FROM TB_GatherInfo(NOLOCK) AS A
              INNER JOIN TB_FocusInfo(NOLOCK) AS B ON A.ID = B.GatherInfoID AND B.RemoveState >= 0 AND B.State>=0
              INNER JOIN TB_AmmeterInfo(NOLOCK) AS C ON B.ID = C.FocusID AND C.State>= 0 AND C.State<100
              INNER JOIN TB_AmmeterGatherItem(NOLOCK) AS D ON C.ID = D.AmmeterID AND D.State>=0
              WHERE 1=1 and C.Special = 0 ";
            //TODO 记得移除特殊表过滤

            if (!string.IsNullOrWhiteSpace(gatherCode))
            {
                sql = $@"{sql} AND A.GatherCode = '{gatherCode}'";
            }
            return await SqlProvider.Instance.Change(DbEnum.EnergyDB)
                .Ado
                .QueryAsync<AmmeterInfo>(sql);
        }


        /// <summary>
        /// 获取电表阀控配置
        /// </summary>
        /// <param name="currentTime">阀控的时间</param>
        /// <returns></returns>
        public override async Task<List<AmmeterAutoValveControlSetting>> GetAmmeterAutoValveControlSetting(string currentTime)
        {
            string sql = $@"SELECT A.*,B.AmmeterID as MeterId,E.GatherCode,D.AreaCode,D.[Address],CONCAT(D.AreaCode, D.[Address]) AS FocusAddress,D.ID as FocusId
                FROM TB_AutoTripTask(nolock) AS A
                INNER JOIN TB_AutoTripAmmeter(nolock) AS B ON A.ID=B.TripTaskID
                INNER JOIN TB_AmmeterInfo(nolock) AS C ON B.AmmeterID=C.ID
                INNER JOIN TB_FocusInfo(nolock) AS D ON D.ID=C.FocusID
                INNER JOIN TB_GatherInfo(NOLOCK) AS E ON E.ID=D.GatherInfoID
                WHERE A.IsForbid=0 and A.State<>-1 and E.GatherCode LIKE '%V4%'  and A.TripTime";

            if (!string.IsNullOrWhiteSpace(currentTime))
            {
                sql = $@"{sql} AND A.TripTime = '{currentTime}'";
            }

            return await SqlProvider.Instance.Change(DbEnum.EnergyDB)
                .Ado
                .QueryAsync<AmmeterAutoValveControlSetting>(sql);
        }


        /// <summary>
        /// 电表自动阀控
        /// </summary>
        /// <returns></returns>
        public override async Task AmmeterScheduledAutoValveControl()
        { 
            var currentTime = DateTime.Now;
            string currentTimeStr = $"{currentTime:HH:mm}";

            try
            {
                //获取电表阀控配置
                var settingInfos = await GetAmmeterAutoValveControlSetting(currentTimeStr);

                if (settingInfos == null || settingInfos.Count <= 0)
                {
                    _logger.LogError($"{nameof(AmmeterScheduledAutoValveControl)} 电表自动阀控时，阀控数据为空, -101");
                    return;
                }

                //批量获取对应的缓存电表信息
                var ammeterInfos = new List<AmmeterInfo>();
                List<MeterReadingTelemetryPacketInfo> taskList = new List<MeterReadingTelemetryPacketInfo>();
                var metadata = await _dbProvider.GetMetadata<MeterReadingTelemetryPacketInfo>();
                 
                foreach (var settingInfo in settingInfos)
                {
                    bool isGenerate = false;
                    switch (settingInfo.LoopType)
                    {
                        case "Once":
                            isGenerate = CommonHelper.JudgeIsGenerate_Once(settingInfo.OnceWithDate.Value, currentTime);
                            break;
                        case "EachDay":
                            isGenerate = CommonHelper.JudgeIsGenerate_Day(settingInfo.EachDayWithout, currentTime);
                            break;
                        case "EachWeek":
                            isGenerate = CommonHelper.JudgeIsGenerate_Week(settingInfo.EachWeekWith, currentTime);
                            break;
                        case "EachMonth":
                            isGenerate = CommonHelper.JudgeIsGenerate_Month(settingInfo.EachMonthWith, currentTime);
                            break;
                    }

                    if (!isGenerate)//不生成，跳入下一循环
                    {
                        continue;
                    }

                    //获取对应的缓存电表信息
                    var ammeterInfo = ammeterInfos.First();
                    bool tripStateResult = false;
                    string subItemCode = string.Empty;
                    if (settingInfo.TripType.Equals("on"))
                    {
                        ammeterInfo.TripState = 0;
                        tripStateResult = true;
                        subItemCode = T645PacketItemCodeConst.C1C01C;
                        if (ammeterInfo.TypeName != 1)
                        {
                            subItemCode = T645PacketItemCodeConst.C1C01B;
                        }
                    }
                    else if (settingInfo.TripType.Equals("off"))
                    {
                        ammeterInfo.TripState = 1;
                        tripStateResult = false;
                        subItemCode = T645PacketItemCodeConst.C1C01A;
                    }
                    else
                    {
                        _logger.LogError($"集中器[{settingInfo.FocusAddress}],[{settingInfo.MeterId}]阀控命令错误：{settingInfo.TripType},-102");
                        continue;
                    }
                     
                    var temCode = "10_01_";
                    
                    //根据电表型号获取协议插件
                    var protocolPlugin = await _protocolService.GetProtocolServiceAsync(ammeterInfo.BrandType);
                    if (protocolPlugin == null)
                    {
                        _logger.LogError($"{nameof(AmmeterScheduledAutoValveControl)} 定时阀控运行时间{currentTime}没有找到对应的协议组件,-105");
                        return;
                    }

                    ProtocolBuildResponse builderResponse = await protocolPlugin.BuildAsync(new ProtocolBuildRequest()
                    {
                        FocusAddress = ammeterInfo.FocusAddress,
                        Pn = ammeterInfo.MeteringCode,
                        ItemCode = temCode,
                        SubProtocolRequest = new SubProtocolBuildRequest()
                        {
                            MeterAddress = ammeterInfo.AmmerterAddress,
                            Password = ammeterInfo.Password,
                            ItemCode = subItemCode,
                        }
                    });
                     
                    string taskMark = CommonHelper.GetTaskMark(builderResponse.AFn, builderResponse.Fn, ammeterInfo.MeteringCode, builderResponse.MSA,builderResponse.Seq);
                    var meterReadingRecords = new MeterReadingTelemetryPacketInfo()
                    {
                        SystemName = SystemType,
                        ProjectId = $"{ammeterInfo.ProjectID}",
                        DeviceType = $"{MeterTypeEnum.Ammeter}",
                        DeviceId = $"{ammeterInfo.MeterId}",
                        Timestamps = currentTime.GetDateTimeOffset().ToUnixTimeNanoseconds(),
                        DatabaseBusiID = ammeterInfo.DatabaseBusiID,
                        PendingCopyReadTime = currentTime,
                        CreationTime = currentTime,
                        MeterAddress = ammeterInfo.AmmerterAddress,
                        AFN = builderResponse.AFn,
                        Fn = builderResponse.Fn,
                        Seq = builderResponse.Seq,
                        MSA = builderResponse.MSA,
                        ItemCode = temCode,
                        TaskMark = taskMark,
                        IsSend = false,
                        ManualOrNot = false,
                        Pn = ammeterInfo.MeteringCode,
                        IssuedMessageId = GuidGenerator.Create().ToString(),
                        IssuedMessageHexString = Convert.ToHexString(builderResponse.Data),
                        IsReceived = false,
                        ScoreValue = $"{ammeterInfo.FocusAddress}.{taskMark}".Md5Fun(),
                    };
                    taskList.Add(meterReadingRecords);
                }
                if (taskList == null || taskList.Count <= 0)
                {
                    _logger.LogError($"{nameof(AmmeterScheduledAutoValveControl)} 定时阀控运行时间{currentTime}没有自动阀控任务生成,-106");
                    return;
                }

                //任务记录入库
                _ = _dbProvider.BatchInsertAsync(metadata, taskList);

                //任务信息推送Kafka
                _ = DeviceGroupBalanceControl.ProcessWithThrottleAsync<MeterReadingTelemetryPacketInfo>(
                    items: taskList,
                    deviceIdSelector: data => data.DeviceId,
                    processor: (data, groupIndex) =>
                    {
                        _ = KafkaProducerIssuedMessageAction(ProtocolConst.AmmeterSubscriberWorkerAutoValveControlIssuedEventName, data, groupIndex);
                    }
                );

                //todo 阀控记录入库，推送到新的服务
            }
            catch (Exception)
            {

                throw;
            }
        }

        /// <summary>
        /// 获取水表信息
        /// </summary>
        /// <param name="gatherCode">采集端Code</param>
        /// <returns></returns>
        //[HttpGet]
        //[Route($"ammeter/list")]
        public override async Task<List<WatermeterInfo>> GetWatermeterInfoList(string gatherCode = "V4-Gather-8890")
        {
            string sql = $@"SELECT 
                            A.ID as MeterId,
                            A.Name,
                            A.FocusID as FocusId,
                            A.MeteringCode,
                            A.Baudrate,
                            A.MeteringPort,
                            A.[Address] AS MeterAddress,
                            A.[Password],
                            A.TypeName,
                            A.Protocol,
                            A.Code,
                            A.LinkType,
                            A.HaveValve,
                            A.MeterType AS MeterTypeName, 
                            A.MeterBrand,
                            A.TimesRate,
                            A.TimeDensity,
                            A.TripState,
                            B.[Address],
                            B.AreaCode,
                            B.AutomaticReport,
                            A.[State],
                            C.GatherCode,
                            A.[ProjectID],
                            B.AbnormalState,
                            B.LastTime,
                            CONCAT(B.AreaCode, B.[Address]) AS FocusAddress,
                            (select top 1 DatabaseBusiID from TB_Project where ID = b.ProjectID) AS DatabaseBusiID
                            FROM [dbo].[TB_WatermeterInfo](NOLOCK) AS A 
                            INNER JOIN [dbo].[TB_FocusInfo](NOLOCK) AS B ON A.FocusID=B.ID AND B.RemoveState >= 0 AND B.State>=0
                            INNER JOIN [dbo].[TB_GatherInfo](NOLOCK) AS C ON B.GatherInfoID=C.ID
                            WHERE A.State>=0 AND A.State<100 ";

            if (!string.IsNullOrWhiteSpace(gatherCode))
            {
                sql = $@"{sql} AND C.GatherCode= '{gatherCode}'";
            }
            return await SqlProvider.Instance.Change(DbEnum.EnergyDB)
            .Ado
            .QueryAsync<WatermeterInfo>(sql);
        }
    }
}