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

using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Apache.IoTDB.DataStructure;
using Apache.IoTDB;
using Confluent.Kafka;
using JiShe.CollectBus.Ammeters;
using JiShe.CollectBus.FreeSql;
using JiShe.CollectBus.IoTDBProvider;
using JiShe.CollectBus.IotSystems.PrepayModel;
using Microsoft.AspNetCore.Authorization;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Extensions.Options;
using JiShe.CollectBus.IoTDBProvider.Context;
using Microsoft.Extensions.Logging;
using JiShe.CollectBus.IotSystems.AFNEntity;
using JiShe.CollectBus.Protocol.Contracts.Interfaces;
using Microsoft.Extensions.DependencyInjection;
using JiShe.CollectBus.Common.Consts;
using JiShe.CollectBus.Common.Enums;
using System.Diagnostics.Metrics;
using JiShe.CollectBus.Common.DeviceBalanceControl;
using JiShe.CollectBus.Kafka.Attributes;
using System.Text.Json;
using JiShe.CollectBus.Kafka;

namespace JiShe.CollectBus.Samples;

public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaSubscribe
{
    private readonly ILogger<SampleAppService> _logger;
    private readonly IIoTDBProvider _iotDBProvider;
    private readonly IoTDBRuntimeContext _dbContext;
    private readonly IoTDBOptions _options;

    public SampleAppService(IIoTDBProvider iotDBProvider, IOptions<IoTDBOptions> options,
        IoTDBRuntimeContext dbContext, ILogger<SampleAppService> logger)
    {
        _iotDBProvider = iotDBProvider;
        _options = options.Value;
        _dbContext = dbContext;
        _logger = logger;
    }


    [HttpGet]
    public async Task UseSessionPool(long timestamps)
    {
        string? messageHexString = null;
        if (timestamps == 0)
        {
            timestamps = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
            _logger.LogError($"timestamps_{timestamps}");
        }
        else
        {
            messageHexString = messageHexString + timestamps;
        }

        ElectricityMeter meter = new ElectricityMeter()
        {
            SystemName = "energy",
            DeviceId = "402440506",
            DeviceType = "Ammeter",
            Current = 10,
            MeterModel = "DDZY-1980",
            ProjectCode = "10059",
            Voltage = 10,
            IssuedMessageHexString = messageHexString,
            Timestamps = timestamps,
        };
        await _iotDBProvider.InsertAsync(meter);
    }


    [HttpGet]
    public async Task UseTableSessionPool()
    {
        ElectricityMeter meter2 = new ElectricityMeter()
        {
            SystemName = "energy",
            DeviceId = "402440506",
            DeviceType = "Ammeter",
            Current = 10,
            MeterModel = "DDZY-1980",
            ProjectCode = "10059",
            Voltage = 10,
            Timestamps = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(),
        };

        await _iotDBProvider.InsertAsync(meter2);

        _dbContext.UseTableSessionPool = true;

        ElectricityMeter meter = new ElectricityMeter()
        {
            SystemName = "energy",
            DeviceId = "402440506",
            DeviceType = "Ammeter",
            Current = 10,
            MeterModel = "DDZY-1980",
            ProjectCode = "10059",
            Voltage = 10,
            Timestamps = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(),
        };
        await _iotDBProvider.InsertAsync(meter);
    }

    /// <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.CacheMeterInfoKey, "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>
    /// 测试单个测点数据项
    /// </summary>
    /// <param name="measuring"></param>
    /// <returns></returns>
    [HttpGet]
    public async Task TestSingleMeasuringAFNData(string measuring, string value)
    {
        var meter = new SingleMeasuringAFNDataEntity<string>()
        {
            SystemName = "energy",
            DeviceId = "402440506",
            DeviceType = "Ammeter",
            ProjectCode = "10059",
            Timestamps = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(),
            SingleMeasuring = new Tuple<string, string>(measuring, value)
        };
        await _iotDBProvider.InsertAsync(meter);
    }


    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(["test-topic"])]

    public async Task<bool> KafkaSubscribeAsync(string obj)
    {
        _logger.LogWarning($"收到订阅消息： {obj}");
       return await Task.FromResult(true);
    }
}