hn_zw_back_main/Service/Cloud/Client/CloudClient.cs

using System.Text;
using Autofac;
using Autofac.Core;
using Common.Util;
using Entity.DbModel.Station;
using HybirdFrameworkCore.Autofac;
using HybirdFrameworkCore.Autofac.Attribute;
using HybirdFrameworkDriver.Common;
using log4net;
using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Client.Connecting;
using MQTTnet.Client.Disconnecting;
using MQTTnet.Client.Options;
using MQTTnet.Client.Receiving;
using MQTTnet.Formatter;
using MQTTnet.Protocol;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using Repository.Station;
using Service.Cloud.Handler;
using Service.Cloud.Msg;
using Service.Cloud.Msg.Cloud.Req;
using Service.Cloud.Msg.Cloud.Resp;
using Service.Cloud.Msg.Cloud.Resp.OutCharger;
using Service.Cloud.Msg.Host.Req;
using Service.Cloud.Msg.Host.Req.OutCharger;

namespace Service.Cloud.Client;

[Scope("SingleInstance")]
public class CloudClient : IMqttClientConnectedHandler, IMqttApplicationMessageReceivedHandler,
    IMqttClientDisconnectedHandler
{
    private static readonly ILog Log = LogManager.GetLogger(typeof(CloudClient));

    public SwapOrderRepository SwapOrderRepository { get; set;}

    #region tcp param

    public string ServerIp { get; set; }
    public int ServerPort { get; set; }
    public string ClientId { get; set; }
    public string? Username { get; set; }
    public string? Password { get; set; }
    public int KeepalivePeriod { get; set; } = 3000;
    public int Timeout { get; set; } = 60;
    public string MqttVersion { get; set; } = "4.0.0";
    public bool IsCleanSession { get; set; } = false;

    #endregion

    #region property

    public bool Connected { get; set; }
    public bool AutoReConnect { get; set; }

    public string StationNo { get; set; }
    public string SubTopic { get; set; }

    public string PubTopic { get; set; }

    public int Encrypt { get; set; }

    public string? AesKey { get; set; }

    public bool Authed { get; set; }

    #endregion

    #region Cmd msg cache

    public CarCanStart? CarCanStart { get; set; }

    /// <summary>
    ///
    /// </summary>
    public MsgPair<VehicleCertification, VehicleCertificationResp> CarAuth { get; set; } = new();
    public MsgPair<ChannelStatusReporting, ChannelStatusReportingResp> ChannelStatus { get; set; } = new();
    public MsgPair<AirConditioningData, AirConditioningDataResp> AirConditioning { get; set; } = new();
    public MsgPair<ChargeRecordReporting, ChargeRecordReportingResp> ChargeRecord { get; set; } = new();
    public MsgPair<EndLogMessage, EndLogMessageResp> EndLog { get; set; } = new();
    public MsgPair<HostStatusReported, HostStatusReportedResp> HostStatus { get; set; } = new();
    public MsgPair<RealTimeFaultInfo, RealTimeFaultInfoResp> RealTimeFault { get; set; } = new();
    public MsgPair<ReportingDeviceList, ReportingDeviceListResp> ReportingDevice { get; set; } = new();
    public MsgPair<SignIn, SignInResp> Sign { get; set; } = new();
    public MsgPair<StartLogMessage, StartLogMessageResp> StartLog { get; set; } = new();
    public MsgPair<TemperatureHumidityData, TemperatureHumidityDataResp> TemperatureHumidity { get; set; } = new();
    public MsgPair<UploadSwapOrder, UploadSwapOrderResp> UploadPowerChange { get; set; } = new();
    public MsgPair<VehicleDataReporting, VehicleDataReportingResp> VehicleData { get; set; } = new();
    public MsgPair<ChargeRecordUpLoad, ChargeRecordUploadRes> ChargeRecordUpLoad { get; set; } = new();
    public MsgPair<ChargeDevDataInfo, ChargeDevDataInfoRes> ChargeDevDataInfo { get; set; } = new();

    public MsgPair<PileEndCharge, PileEndChargeResp> PileEndCharge { get; set; } = new();

    public MsgPair<PileChargeRealtime, PileChargeRealtimeResp> PileChargeRealtime { get; set; } = new();
    public MsgPair<PileRealtime, PileRealtimeResp> PileRealtime { get; set; } = new();

    #endregion

    #region basic

    private IMqttClient? MqttClient;

    private List<IBaseHandler> handlers = new();

    private static ushort _incrementId;

    private static ushort GetIncrementId()
    {
        if (_incrementId < 65535)
        {
            _incrementId += 1;
        }
        else
        {
            _incrementId = 1;
        }

        return _incrementId;
    }

    public void InitHandler()
    {
        var list = new List<Type>();

        foreach (var reg in AppInfo.Container.ComponentRegistry.Registrations)
        foreach (var service in reg.Services)
            if (service is TypedService ts)
                if (MatchHandlers(ts))
                    list.Add(ts.ServiceType);


        foreach (var type in list)
        {
            var resolve = AppInfo.Container.Resolve(type);
            handlers.Add((IBaseHandler)resolve);
        }
    }

    private bool MatchHandlers(TypedService ts)
    {
        var interfaces = ts.ServiceType.GetInterfaces();
        if (interfaces.Length > 0)
            foreach (var type in interfaces)
            {
                if (type.ToString().Contains("Service.Cloud.Handler"))
                {
                    return true;
                }
            }

        return false;
    }


    public void Connect()
    {
        Log.Info($"begin connect cloud {ServerIp}:{ServerPort} with client={ClientId}");
        if (MqttClient == null)
        {
            MqttClient = new MqttFactory().CreateMqttClient();
            MqttClient.ConnectedHandler = this;
            MqttClient.ApplicationMessageReceivedHandler = this;
            MqttClient.DisconnectedHandler = this;
        }

        try
        {
            var task = MqttClient.ConnectAsync(BuildOptions());
            MqttClientConnectResult result = task.Result;
            Log.Info($"connect cloud {result.ResultCode}");
            if (result.ResultCode == MqttClientConnectResultCode.Success)
            {
                Connected = true;
            }
            else
            {
                Connected = false;
                if (AutoReConnect)
                {
                    Thread.Sleep(5000);
                    Connect();
                }
            }
        }
        catch (Exception e)
        {
            Log.Error("connect cloud error", e);
            if (AutoReConnect)
            {
                Thread.Sleep(5000);
                Connect();
            }
        }
    }

    /// <summary>
    /// 连接成功回调
    /// </summary>
    /// <param name="eventArgs"></param>
    /// <returns></returns>
    public async Task HandleConnectedAsync(MqttClientConnectedEventArgs eventArgs)
    {
        await DoSubTopic(SubTopic);

        /*SendVehicleCertification(new VehicleCertification()
        {
            dt = DateTime.Now,
            en = 1,
            mode =1,
            rfid= "LZ5NB6D33RB000438",
            ty= 2,
            vi ="LZ5NB6D33RB000438"
        });*/
        // SendSignIn(new SignIn()
        // {
        //     sn = StaticStationInfo.StationNo,
        //     st = "01",
        //     ss = StaticStationInfo.StationStatus,
        //     en = 1,
        //     cn = 7
        // });
    }

    private async Task DoSubTopic(string topic)
    {
        List<MqttTopicFilter> list = new List<MqttTopicFilter>();
        string[] topics = topic.Split(new char[] { ',' });
        foreach (string str in topics)
        {
            MqttTopicFilter topicFilter = new MqttTopicFilter
            {
                Topic = str,
                QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce
            };
            list.Add(topicFilter);
        }

        await MqttClient.SubscribeAsync(list.ToArray());

        Log.Info($"subscribe {topic} success ");
    }

    public void Publish<T>(T data) where T : ICmd
    {
        if (MqttClient.IsConnected)
        {
            Model<T> model = new Model<T>
            {
                Header = new Header()
                {
                    cmd = data.GetCmd(),
                    cipherFlag = Encrypt,
                    id = GetIncrementId(),
                    sid = StationNo,
                    timeStamp = (DateTime.Now.ToUniversalTime().Ticks - 621355968000000000) / 10000
                },
                body = data
            };

            model.dataSign = "";//SignData(model);

            var settings = new JsonSerializerSettings()
            {
                ReferenceLoopHandling = ReferenceLoopHandling.Ignore,
                DateFormatString = "yyyy-MM-dd HH:mm:ss",
                NullValueHandling = NullValueHandling.Ignore
            };
            Log.Info($"send {JsonConvert.SerializeObject(model, settings)}");
            var appMsg = new MqttApplicationMessage
            {
                Topic = PubTopic,
                Payload = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(model, settings)),
                QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce,
                Retain = false
            };

            MqttClient.PublishAsync(appMsg);
        }

    }

    private string SignData<T>(Model<T> model) where T : ICmd
    {
        var settings = new JsonSerializerSettings()
        {
            ReferenceLoopHandling = ReferenceLoopHandling.Ignore,
            DateFormatString = "yyyy-MM-dd HH:mm:ss",
            NullValueHandling = NullValueHandling.Ignore
        };
        string body = JsonConvert.SerializeObject(model.body, settings);
        return MD5Util.MD5Encrypt32(body + ":" + model.Header.timeStamp + ":" + model.Header.id).ToLower();
    }

    /// <summary>
    /// 消息接收回调
    /// </summary>
    /// <param name="eventArgs"></param>
    /// <returns></returns>
    public Task HandleApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs eventArgs)
    {
        var message = eventArgs.ApplicationMessage;
        if (message.Topic != SubTopic)
        {
            return Task.CompletedTask;
        }

        if (message.Payload == null)
        {
            return Task.CompletedTask;
        }

        string s = Encoding.UTF8.GetString(message.Payload);
        if (string.IsNullOrWhiteSpace(s))
        {
            return Task.CompletedTask;
        }

        Log.Info($"from cloud receive {s} ");
        JObject objResult = JObject.Parse(s);
        string headerStr = objResult["header"].ToString();
        Header? header = JsonConvert.DeserializeObject<Header>(headerStr);
        if (header == null)
        {
            return Task.CompletedTask;
        }

        foreach (IBaseHandler handler in handlers)
        {
            if (handler.CanHandle(header.cmd))
            {
                string bodyStr = objResult["body"].ToString();
                handler.Handle(bodyStr);
                break;
            }
        }

        return Task.CompletedTask;
    }


    /// <summary>
    /// 断开回调
    /// </summary>
    /// <param name="eventArgs"></param>
    /// <returns></returns>
    public Task HandleDisconnectedAsync(MqttClientDisconnectedEventArgs eventArgs)
    {
        Log.Info("cloud disconnect");
        Connected = false;
        if (AutoReConnect)
        {
            Connect();
        }
        return Task.CompletedTask;
    }

    private IMqttClientOptions BuildOptions()
    {
        MqttClientOptionsBuilder builder =
            new MqttClientOptionsBuilder().WithTcpServer(ServerIp, ServerPort).WithClientId(ClientId);
        if (!string.IsNullOrWhiteSpace(Username))
        {
            builder.WithCredentials(Username, Encoding.UTF8.GetBytes(Password));
        }

        if (IsCleanSession)
        {
            builder.WithCleanSession();
        }

        builder.WithKeepAlivePeriod(TimeSpan.FromSeconds(KeepalivePeriod))
            .WithCommunicationTimeout(TimeSpan.FromSeconds(Timeout));
        switch (MqttVersion)
        {
            case "3.1.0":
                builder.WithProtocolVersion(MqttProtocolVersion.V310);
                break;

            case "5.0.16":
                builder.WithProtocolVersion(MqttProtocolVersion.V500);
                break;

            default:
                builder.WithProtocolVersion(MqttProtocolVersion.V311);
                break;
        }

        return builder.Build();
    }

    #endregion


    #region 主动发送CMD

    public VehicleCertificationResp? SendVehicleCertification(VehicleCertification vehicleCertification,
        TimeSpan? timeSpan = null)
    {
        Log.Info(vehicleCertification);
        this.CarAuth.Req = vehicleCertification;
        this.Publish(vehicleCertification);
        return CarAuth.GetResp(timeSpan);
    }

    /// <summary>
    /// 换电站通道状态上报
    /// </summary>
    /// <param name="channelStatusReporting"></param>
    /// <param name="timeSpan"></param>
    /// <returns></returns>
    public ChannelStatusReportingResp? SendChannelStatusReporting(ChannelStatusReporting channelStatusReporting,
        TimeSpan timeSpan)
    {
        this.ChannelStatus.Req = channelStatusReporting;
        this.Publish(channelStatusReporting);
        return ChannelStatus.GetResp(timeSpan);
    }
    public AirConditioningDataResp? SendAirConditioningData(AirConditioningData airConditioningData,
        TimeSpan timeSpan)
    {
        this.AirConditioning.Req = airConditioningData;
        this.Publish(airConditioningData);
        return AirConditioning.GetResp(timeSpan);
    }
    public ChargeRecordReportingResp? SendChargeRecordReporting(ChargeRecordReporting chargeRecord,
        TimeSpan timeSpan)
    {
        this.ChargeRecord.Req = chargeRecord;
        this.Publish(chargeRecord);
        return ChargeRecord.GetResp(timeSpan);
    }
    public EndLogMessageResp? SendEndLogMessage(EndLogMessage endLogMessage,
        TimeSpan timeSpan)
    {
        this.EndLog.Req = endLogMessage;
        this.Publish(endLogMessage);
        return EndLog.GetResp(timeSpan);
    }
    public HostStatusReportedResp? SendHostStatusReported(HostStatusReported hostStatusReported,
        TimeSpan timeSpan)
    {
        this.HostStatus.Req = hostStatusReported;
        this.Publish(hostStatusReported);
        return HostStatus.GetResp(timeSpan);
    }
    public RealTimeFaultInfoResp? SendRealTimeFaultInfo(RealTimeFaultInfo realTimeFaultInfo,
        TimeSpan timeSpan)
    {
        this.RealTimeFault.Req = realTimeFaultInfo;
        this.Publish(realTimeFaultInfo);
        return RealTimeFault.GetResp(timeSpan);
    }
    public ReportingDeviceListResp? SendReportingDeviceList(ReportingDeviceList reportingDeviceList,
        TimeSpan timeSpan)
    {
        this.ReportingDevice.Req = reportingDeviceList;
        this.Publish(reportingDeviceList);
        return ReportingDevice.GetResp(timeSpan);
    }
    public SignInResp? SendSignIn(SignIn signIn,
        TimeSpan? timeSpan=null)
    {
        this.Sign.Req = signIn;
        this.Publish(signIn);
        return Sign.GetResp(timeSpan);
    }
    public StartLogMessageResp? SendStartLogMessage(StartLogMessage startLogMessage,
        TimeSpan timeSpan)
    {
        this.StartLog.Req = startLogMessage;
        this.Publish(startLogMessage);
        return StartLog.GetResp(timeSpan);
    }
    public TemperatureHumidityDataResp? SendTemperatureHumidityData(TemperatureHumidityData temperatureHumidityData,
        TimeSpan timeSpan)
    {
        this.TemperatureHumidity.Req = temperatureHumidityData;
        this.Publish(temperatureHumidityData);
        return TemperatureHumidity.GetResp(timeSpan);
    }
    /// <summary>
    /// 上传换电订单
    /// </summary>
    /// <param name="uploadSwapOrder"></param>
    /// <param name="timeSpan"></param>
    /// <returns></returns>
    public UploadSwapOrderResp? SendUploadPowerChangeOrder(UploadSwapOrder uploadSwapOrder, TimeSpan? timeSpan = null)
    {
        this.UploadPowerChange.Req = uploadSwapOrder;
        this.Publish(uploadSwapOrder);
        return UploadPowerChange.GetResp(timeSpan);
    }
    public VehicleDataReportingResp? SendVehicleDataReporting(VehicleDataReporting vehicleDataReporting,
        TimeSpan timeSpan)
    {
        this.VehicleData.Req = vehicleDataReporting;
        this.Publish(vehicleDataReporting);
        return VehicleData.GetResp(timeSpan);
    }

    public ChargeRecordUploadRes? SendChargeRecordUpLoad(ChargeRecordUpLoad req, TimeSpan? timeSpan = null)
    {
        this.ChargeRecordUpLoad.Req = req;
        this.Publish(req);
        return ChargeRecordUpLoad.GetResp(timeSpan);
    }

    public ChargeDevDataInfoRes? SendChargeDevDataInfo(ChargeDevDataInfo req, TimeSpan? timeSpan = null)
    {
        this.ChargeDevDataInfo.Req = req;
        this.Publish(req);
        return ChargeDevDataInfo.GetResp(timeSpan);
    }

    public PileEndChargeResp? SendPileEndCharge(PileEndCharge req, TimeSpan? timeSpan = null)
    {
        this.PileEndCharge.Req = req;
        this.Publish(req);
        return PileEndCharge.GetResp(timeSpan);
    }

    public PileChargeRealtimeResp? SendPileChargeRealtime(PileChargeRealtime req, TimeSpan? timeSpan = null)
    {
        this.PileChargeRealtime.Req = req;
        this.Publish(req);
        return PileChargeRealtime.GetResp(timeSpan);
    }

    public PileRealtimeResp? SendPileRealtime(PileRealtime req, TimeSpan? timeSpan = null)
    {
        this.PileRealtime.Req = req;
        this.Publish(req);
        return PileRealtime.GetResp(timeSpan);
    }
    #endregion

    #region business func

    /// <summary>
    ///
    /// </summary>
    /// <param name="chargeOrder"></param>
    /// <param name="op">1： 自动； 2： 人工手动</param>
    /// <param name="timeSpan">超时等待</param>
    /// <returns></returns>
    public bool PublishChargeOrder(List<ChargeOrder> orders, int op, TimeSpan? timeSpan = null)
    {
        ChargeOrder chargeOrder = orders[0];
        ChargeOrder lastChargeOrder = orders[^1];
        string swapOrderSn = chargeOrder.SwapOrderSn;
        SwapOrder? swapOrder = SwapOrderRepository.QueryByClause(it => it.Sn == swapOrderSn);
        ChargeRecordUpLoad req = new ChargeRecordUpLoad()
        {
            chrsn = chargeOrder.CloudChargeOrder,
            son = swapOrder?.CloudSn,
            bid = chargeOrder.BatteryNo,
            st = chargeOrder.StartTime ?? DateTime.Now,
            et = lastChargeOrder.EndTime ?? DateTime.Now,
            ssoc = chargeOrder.StartSoc ?? 0,
            esoc = lastChargeOrder.StopSoc ?? 0,
            //ssoe = chargeOrder.soe
            //esoe
            dcce =0,
            acce =0,
            tp = 0,
            pp = 0,
            fp = 0,
            vp = 0,
            ct = 0,
            cn = orders.Count,
            sfs = op,
            vin = swapOrder?.VehicleVin,
            sfoc = 0,
        };
        foreach (ChargeOrder order in orders)
        {
            req.dcce += Convert.ToSingle(order.StopDcElec ?? 0 - order.StartDcElec ?? 0);
            req.acce += Convert.ToSingle(order.StopAcElec ?? 0 - order.StartAcElec ?? 0);
            req.tp   += Convert.ToSingle(order.SharpElecCount);
            req.pp   += Convert.ToSingle(order.PeakElecCount);
            req.fp   += Convert.ToSingle(order.FlatElecCount);
            req.vp   += Convert.ToSingle(order.ValleyElecCount);
            req.ct   += ((order.EndTime ?? DateTime.Now).Subtract(order.StartTime ?? DateTime.Now)).Minutes;
        }

        this.SendChargeRecordUpLoad(req);
        return true;
    }

    #endregion
}