JiShe.CollectBus/services/JiShe.CollectBus.Application/Samples/SampleAppService.cs

using JiShe.CollectBus.Ammeters;
using JiShe.CollectBus.Common.Consts;
using JiShe.CollectBus.Common.DeviceBalanceControl;
using JiShe.CollectBus.Common.Extensions;
using JiShe.CollectBus.Common.Helpers;
using JiShe.CollectBus.IotSystems.Devices;
using JiShe.CollectBus.IotSystems.PrepayModel;
using JiShe.CollectBus.Protocol.Interfaces;
using JiShe.ServicePro.FreeSqlProvider;
using JiShe.ServicePro.IoTDBManagement.Options;
using JiShe.ServicePro.IoTDBManagement.SessionPools;
using JiShe.ServicePro.Kafka.Internal;
using JiShe.ServicePro.Kafka.Producer;
using Microsoft.AspNetCore.Authorization;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using TouchSocket.Sockets;

namespace JiShe.CollectBus.Samples;

public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaSubscribe
{
    private readonly ILogger<SampleAppService> _logger;
    private readonly IoTDBSessionPoolProvider _iotDBProvider;
    private readonly IoTDbOptions _options;
    private readonly IRedisDataCacheService _redisDataCacheService;

    private readonly IProducerService _producerService;
    private readonly ITcpService _tcpService;
    public SampleAppService(IoTDBSessionPoolProvider iotDBProvider, IOptions<IoTDbOptions> options,
        ILogger<SampleAppService> logger, IRedisDataCacheService redisDataCacheService, IProducerService producerService, ITcpService tcpService)
    {
        _iotDBProvider = iotDBProvider;
        _options = options.Value;
        _logger = logger;
        _redisDataCacheService = redisDataCacheService;
        _producerService = producerService;
        _tcpService = tcpService;
    }

    /// <summary>
    /// 测试 UseSessionPool
    /// </summary>
    /// <param name="testTime"></param>
    /// <returns></returns>
    [HttpGet]
    public async Task UseSessionPool(long testTime)
    {
        var dataTime = DateTime.Now;

        List<string> values = new List<string>() { $"{dataTime:yy}", $"{dataTime:MM}", $"{dataTime:dd}", $"{dataTime:HH}", $"{dataTime:mm}", };


        
        //ElectricityMeterTreeModelExtension.GetCurrent()
        //SourceEntityAccessorFactory.SetCurrent(meter);

        //ElectricityMeterTreeModelAccessor.
        //TableModelSingleMeasuringEntityExtension
        //TableModelSingleMeasuringEntityAccessor.GetSystemName(meter);
        //ElectricityMeterAccessor 
    }

    /// <summary>
    /// 测试Session切换
    /// </summary>
    /// <returns></returns>
    [HttpGet]
    public async Task UseTableSessionPool(DateTime time)
    {
        var testTime = time;
        

    }

    /// <summary>
    /// 测试Session切换3
    /// </summary>
    /// <returns></returns>
    [HttpGet]
    public async Task UseTableSessionPool3(DateTime time)
    {
        var testTime = time;
         
    }

    /// <summary>
    /// 测试树模型单个测点数据项
    /// </summary>
    /// <param name="measuring"></param>
    /// <returns></returns>
    [HttpGet]
    public async Task TestTreeModelSingleMeasuringEntity(string measuring, string value, DateTime time)
    {
        
    }

    /// <summary>
    /// 测试树模型单个测点数据项2
    /// </summary>
    /// <param name="measuring"></param>
    /// <returns></returns>
    [HttpGet]
    public async Task TestTreeModelSingleMeasuringEntity2(string measuring, int value, DateTime time)
    {
        
    }

    /// <summary>
    /// 测试树模型单个测点数据项查询
    /// </summary>
    /// <param name="measuring"></param>
    /// <returns></returns>
    [HttpGet]
    public async Task TestTreeModelSingleMeasuringEntityQuery()
    {

         
    }

    /// <summary>
    /// 测试表模型单个测点数据项
    /// </summary>
    /// <param name="measuring"></param>
    /// <returns></returns>
    [HttpGet]
    public async Task TestTableModelSingleMeasuringEntity(string measuring, string value, DateTime time)
    {
        time = DateTime.Now;

        
    }

    /// <summary>
    /// 测试表模型单个测点数据项2
    /// </summary>
    /// <param name="measuring"></param>
    /// <returns></returns>
    [HttpGet]
    public async Task TestTableModelSingleMeasuringEntity2(string measuring, int value, DateTime time)
    {
        time = DateTime.Now;

         
    }

    /// <summary>
    /// 测试设备分组均衡控制算法
    /// </summary>
    /// <param name="deviceCount"></param>
    /// <returns></returns>
    [HttpGet]
    public async Task TestDeviceGroupBalanceControl(int deviceCount = 200000)
    {
        //var deviceList = new List<string>();
        //for (int i = 0; i < deviceCount; i++)
        //{
        //    deviceList.Add($"Device_{Guid.NewGuid()}");
        //}

        //// 初始化缓存
        //DeviceGroupBalanceControl.InitializeCache(deviceList);

        //var timeDensity = "15";
        ////获取缓存中的电表信息
        //var redisKeyList = $"{string.Format(RedisConst.CacheMeterInfoHashKey, "Energy", "JiSheCollectBus", MeterTypeEnum.Ammeter.ToString(), timeDensity)}*";

        //var oneMinutekeyList = await FreeRedisProvider.Instance.KeysAsync(redisKeyList);
        //var meterInfos = await GetMeterRedisCacheListData<AmmeterInfo>(oneMinutekeyList, "Energy", "JiSheCollectBus", timeDensity, MeterTypeEnum.Ammeter);
        //List<string> focusAddressDataLista = new List<string>();
        //foreach (var item in meterInfos)
        //{
        //    focusAddressDataLista.Add(item.FocusAddress);
        //}

        //DeviceGroupBalanceControl.InitializeCache(focusAddressDataLista);

        //// 打印分布统计
        //DeviceGroupBalanceControl.PrintDistributionStats();

        await Task.CompletedTask;
    }

    /// <summary>
    /// 测试设备分组均衡控制算法获取分组Id
    /// </summary>
    /// <param name="deviceAddress"></param>
    /// <returns></returns>
    [HttpGet]
    public async Task TestGetDeviceGroupBalanceControl(string deviceAddress)
    {
        var groupId = DeviceGroupBalanceControl.GetDeviceGroupId(deviceAddress);
        Console.WriteLine(groupId);

        await Task.CompletedTask;
    }


    /// <summary>
    /// 测试Redis批量读取10万条数据性能
    /// </summary>
    /// <returns></returns>
    [HttpGet]
    public async Task TestRedisCacheGetAllPagedData()
    {
        //var timeDensity = "15";
        //string SystemType = "Energy";
        //string ServerTagName = "JiSheCollectBus2";
        //var redisCacheMeterInfoHashKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoHashKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity)}";
        //var redisCacheMeterInfoSetIndexKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoSetIndexKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity)}";
        //var redisCacheMeterInfoZSetScoresIndexKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoZSetScoresIndexKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity)}";

        //var timer1 = Stopwatch.StartNew();
        //decimal? cursor = null;
        //string member = null;
        //bool hasNext;
        //List<AmmeterInfo> meterInfos = new List<AmmeterInfo>();
        //do
        //{
        //    var page = await _redisDataCacheService.GetAllPagedData<AmmeterInfo>(
        //        redisCacheMeterInfoHashKeyTemp,
        //        redisCacheMeterInfoZSetScoresIndexKeyTemp,
        //        pageSize: 1000,
        //        lastScore: cursor,
        //        lastMember: member);

        //    meterInfos.AddRange(page.Items);
        //    cursor = page.HasNext ? page.NextScore : null;
        //    member = page.HasNext ? page.NextMember : null;
        //    hasNext = page.HasNext;
        //} while (hasNext);

        //timer1.Stop();
        //_logger.LogError($"读取数据更花费时间{timer1.ElapsedMilliseconds}毫秒");

        //List<string> focusAddressDataLista = new List<string>();
        //foreach (var item in meterInfos)
        //{
        //    focusAddressDataLista.Add(item.FocusAddress);
        //}

        //DeviceGroupBalanceControl.InitializeCache(focusAddressDataLista);

        //// 打印分布统计
        //DeviceGroupBalanceControl.PrintDistributionStats();

        await Task.CompletedTask;
    }


    /// <summary>
    /// 下一个采集时间点验证
    /// </summary> 
    /// <returns></returns>
    [HttpGet]
    public async Task<DateTime> TestCalculateNextCollectionTime(string time, int timeDensity)
    {
        DateTime nextTaskTime = Convert.ToDateTime(time);

        return await Task.FromResult(nextTaskTime.CalculateNextCollectionTime(timeDensity));
    }


    public Task<SampleDto> GetAsync()
    {
        return Task.FromResult(
            new SampleDto
            {
                Value = 42
            }
        );
    }

    [Authorize]
    public Task<SampleDto> GetAuthorizedAsync()
    {
        return Task.FromResult(
            new SampleDto
            {
                Value = 42
            }
        );
    }

    [AllowAnonymous]
    public async Task<List<Vi_BaseAmmeterInfo>> Test()
    {

        var ammeterList = await SqlProvider.Instance.Change(DbEnum.PrepayDB).Select<Vi_BaseAmmeterInfo>().Where(d => d.TB_CustomerID == 5).Take(10).ToListAsync();
        return ammeterList;
    }

    [AllowAnonymous]
    public bool GetTestProtocol()
    {
        var aa = LazyServiceProvider.GetKeyedService<IProtocolPlugin>("TestProtocolPlugin");
        return aa == null;
    }

    [KafkaSubscribe(KafkaTopicConsts.TESTTOPIC)]

    public async Task<ISubscribeAck> KafkaSubscribeAsync(object obj)
    {
        _logger.LogWarning($"收到订阅消息： {obj}");
        return SubscribeAck.Success();
    }

    /// <summary>
    /// 测试Kafka下发报文
    /// </summary>
    /// <param name="input"></param>
    /// <returns></returns>
    [AllowAnonymous]
    public async Task<bool> KafkaSendAsync(KafkaSendDto input)
    {
        ArgumentException.ThrowIfNullOrWhiteSpace(input.Address);
        ArgumentException.ThrowIfNullOrWhiteSpace(input.Frame);
        input.Frame = input.Frame.Replace(" ", "");
        await _producerService.ProduceAsync<KafkaSendDto>(KafkaTopicConsts.TESTSENDTOPIC, input);
        return await Task.FromResult(true);
    }


    /// <summary>
    /// 订阅下发的数据
    /// </summary>
    /// <param name="dto"></param>
    /// <returns></returns>
    [KafkaSubscribe(KafkaTopicConsts.TESTSENDTOPIC), ApiExplorerSettings(IgnoreApi = true)]

    public async Task<ISubscribeAck> KafkaSubscribeTestSendAsync(KafkaSendDto dto)
    {
        if (_tcpService.ClientExists(dto.Address))
        {
            // 发送给设备
            await _tcpService.SendAsync(dto.Address, Convert.FromHexString(dto.Frame));
            _logger.LogWarning($"{dto.Address}下发消息报文：{dto.Frame}");
        }
        else
        {
            _logger.LogWarning($"{dto.Address}集中器未上线： {dto.Serialize()}");
        }
        // 测试不管是否上线都ACK
        return SubscribeAck.Success();
    }


    /// <summary>
    /// 测试Redis批量读取10万条数据性能
    /// </summary>
    /// <returns></returns>
    [HttpGet]
    public async Task TestRedisCacheGetData(string scores)
    {
        //var timeDensity = "15";
        //string SystemType = "Energy";
        //string ServerTagName = "JiSheCollectBus5";
        //var redisCacheMeterInfoHashKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoHashKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity)}";
        //var redisCacheMeterInfoSetIndexKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoSetIndexKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity)}";
        //var redisCacheMeterInfoZSetScoresIndexKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoZSetScoresIndexKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity)}";

        //var page = await _redisDataCacheService.GetSingleData<AmmeterInfo>(
        //    redisCacheMeterInfoHashKeyTemp,
        //    redisCacheMeterInfoZSetScoresIndexKeyTemp,
        //    "973219481:17",
        //    pageSize: 1000,
        //    lastScore: 100,
        //    lastMember: "memberId",
        //    descending: true
        //    );

        await Task.CompletedTask;
    }


    /// <summary>
    /// IoTDB空表查询情况
    /// </summary>
    /// <returns></returns>
    [HttpGet]
    public async Task TestIoTDBEmptyTableQuery()
    {
        var meter = new MeterReadingTelemetryPacketInfo() { DevicePath = "MeterReadingTelemetryPacketInfo", DeviceId = "1111" };
        QueryCondition conditions = new QueryCondition()
        {
            Field = "DeviceId",
            Operator = "=",
            Value = meter.DeviceId
        };
        var query = new IoTDBQueryOptions()
        {
            TableNameOrTreePath = meter.DevicePath,
            PageIndex = 1,
            PageSize = 1,
            Conditions = new List<QueryCondition>() { conditions },
        };

        await _iotDBProvider.GetSessionPool(true).InitTableSessionModelAsync();

        var pageResult = await _iotDBProvider.GetSessionPool(true).QueryAsync<MeterReadingTelemetryPacketInfo>(query);

        await Task.CompletedTask;
    }
}