iot/projects/IoTCenter/Services/IoTCenterEventHandler.cs

using Application.Domain.Entities;
using Application.Models;
using Hangfire;
using Infrastructure.Application.Services.Settings;
using Infrastructure.Data;
using Infrastructure.Events;
using Infrastructure.Extensions;
using IoT.Shared.Services.IoTCenter;
using Jint;
using Jint.Native;
using Microsoft.AspNetCore.SignalR;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Logging;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Net.Http;
using System.Threading;
using System.Threading.Tasks;
using Vibrant.InfluxDB.Client;
using Vibrant.InfluxDB.Client.Rows;

namespace IoTCenter.Services
{
    public class IoTCenterEventHandler :
        IEventHander<EntityInsertedEvent<Organ>>,
        IEventHander<EntityUpdatedEvent<Organ>>,
        IEventHander<EntityDeletedEvent<Organ>>,
        IEventHander<EntityInsertedEvent<NodeCategory>>,
        IEventHander<EntityUpdatedEvent<NodeCategory>>,
        IEventHander<EntityInsertedEvent<OrganSceneTimer>>,
        IEventHander<EntityUpdatedEvent<OrganSceneTimer>>,
        IEventHander<EntityDeletedEvent<OrganSceneTimer>>,
        IEventHander<EntityInsertedEvent<Node>>,
        IEventHander<EntityUpdatedEvent<Node>>,
        IEventHander<EntityDeletedEvent<Node>>,
        IEventHander<EntityInsertedEvent<Product>>,
        IEventHander<EntityUpdatedEvent<Product>>,
        IEventHander<EntityDeletedEvent<Product>>,
        IEventHander<EntityInsertedEvent<Device>>,
        IEventHander<EntityUpdatedEvent<Device>>,
        IEventHander<EntityDeletedEvent<Device>>,
        IEventHander<EntityInsertedEvent<Data>>,
        IEventHander<EntityUpdatedEvent<Data>>,
        IEventHander<EntityDeletedEvent<Data>>,
        IEventHander<EntityInsertedEvent<Command>>,
        IEventHander<EntityUpdatedEvent<Command>>,
        IEventHander<EntityDeletedEvent<Command>>,
        IEventHander<EntityInsertedEvent<Scene>>,
        IEventHander<EntityUpdatedEvent<Scene>>,
        IEventHander<EntityDeletedEvent<Scene>>,
        IEventHander<NodeClientConnectedEvent>
    {
        private readonly IConfiguration _cfg;
        private readonly ISettingService _settingService;
        private readonly ILogger<IoTCenterEventHandler> _logger;
        private readonly IRepository<Node> _nodeRepo;
        private readonly IRepository<NodeCategory> _nodeCategoryRepo;
        private readonly IRepository<NodeCategoryNode> _nodeCategoryNodeRepo;
        private readonly IRepository<Organ> _organRepo;
        private readonly IRepository<OrganNode> _organNodeRepo;
        private readonly IRepository<Device> _deviceRepo;
        private readonly IRepository<OrganSceneTimer> _organSceneTimerRepo;
        private readonly IRepository<OrganSceneTigger> _sceneTiggerRepo;
        private readonly ISceneTiggerService _sceneTiggerService;
        private readonly IHubContext<IoTCenterHub> _hub;
        private readonly IHttpClientFactory _httpClientFactory;

        public IoTCenterEventHandler(IConfiguration cfg,
            ISettingService settingService,
            ILogger<IoTCenterEventHandler> logger,
            IRepository<Node> nodeRepo,
            IRepository<NodeCategory> nodeCategoryRepo,
            IRepository<NodeCategoryNode> nodeCategoryNodeRepo,
            IRepository<Organ> organRepo,
            IRepository<OrganNode> organNodeRepo,
            IRepository<Device> deviceRepo,
            IRepository<OrganSceneTimer> organSceneTimerRepo,
            IRepository<OrganSceneTigger> sceneTiggerRepo,
            ISceneTiggerService sceneTiggerService,
            IHubContext<IoTCenterHub> hub,
            IHttpClientFactory httpClientFactory)
        {
            this._cfg = cfg;
            this._settingService = settingService;
            this._logger = logger;
            this._nodeRepo = nodeRepo;
            this._nodeCategoryRepo = nodeCategoryRepo;
            this._nodeCategoryNodeRepo = nodeCategoryNodeRepo;
            this._organRepo = organRepo;
            this._organNodeRepo = organNodeRepo;
            this._deviceRepo = deviceRepo;
            this._organSceneTimerRepo = organSceneTimerRepo;
            this._sceneTiggerRepo = sceneTiggerRepo;
            this._sceneTiggerService = sceneTiggerService;
            this._hub = hub;
            this._httpClientFactory = httpClientFactory;
        }

        #region timer

        public void Handle(EntityInsertedEvent<OrganSceneTimer> message)
        {
            var timer = message.Data;
            RecurringJob.AddOrUpdate(timer.Id.ToString(), () => Handle(timer.Id), timer.Cron, TimeZoneInfo.Local);
        }

        public void Handle(EntityUpdatedEvent<OrganSceneTimer> message)
        {
            var timer = message.Data;
            RecurringJob.AddOrUpdate(timer.Id.ToString(), () => Handle(timer.Id), timer.Cron, TimeZoneInfo.Local);
        }

        public void Handle(EntityDeletedEvent<OrganSceneTimer> message)
        {
            var timer = message.Data;
            RecurringJob.RemoveIfExists(timer.Id.ToString());
        }

        #endregion timer

        #region Product

        public void Handle(EntityInsertedEvent<Product> message)
        {
            this.Notify(message);
        }

        public void Handle(EntityUpdatedEvent<Product> message)
        {
            this.Notify(message);
        }

        public void Handle(EntityDeletedEvent<Product> message)
        {
            this.Notify(message);
        }

        #endregion Product

        #region Node

        public void Handle(EntityInsertedEvent<Node> message)
        {
            this.Notify(message);
            //this.UpdateOrganNode(message.Data);
        }

        public void Handle(EntityUpdatedEvent<Node> message)
        {
            this.Notify(message);
            //this.UpdateOrganNode(message.Data);
        }

        public void Handle(EntityDeletedEvent<Node> message)
        {
            this.Notify(message);
        }

        #endregion Node

        #region organ

        public void Handle(EntityInsertedEvent<Organ> message)
        {
            this.Notify(message);
            //this.UpdateOrganNode(message.Data);
        }

        public void Handle(EntityUpdatedEvent<Organ> message)
        {
            this.Notify(message);
            //this.UpdateOrganNode(message.Data);
        }

        public void Handle(EntityDeletedEvent<Organ> message)
        {
            this.Notify(message);
        }

        #endregion organ

        #region nodeCategory

        public void Handle(EntityInsertedEvent<NodeCategory> message)
        {
            this.Notify(message);
        }

        public void Handle(EntityUpdatedEvent<NodeCategory> message)
        {
            this.Notify(message);
        }

        #endregion nodeCategory

        #region Device

        public void Handle(EntityInsertedEvent<Device> message)
        {
            this.Notify(message);
        }

        public void Handle(EntityUpdatedEvent<Device> message)
        {
            this.Notify(message);
        }

        public void Handle(EntityDeletedEvent<Device> message)
        {
            this.Notify(message);
        }

        #endregion Device

        #region Data

        public void Handle(EntityInsertedEvent<Data> message)
        {
            message.Data.Device = _deviceRepo.ReadOnlyTable().Include(o => o.Node).Where(o => o.Id == message.Data.DeviceId).FirstOrDefault();
            this.Notify(message);
            this.TiggerHandle(message);
            this.LogData(message.Data);
        }

        public void Handle(EntityUpdatedEvent<Data> message)
        {
            message.Data.Device = _deviceRepo.ReadOnlyTable().Include(o => o.Node).Where(o => o.Id == message.Data.DeviceId).FirstOrDefault();
            this.Notify(message);
            this.TiggerHandle(message);
            this.LogData(message.Data);
        }

        public void Handle(EntityDeletedEvent<Data> message)
        {
            this.Notify(message);
        }

        #endregion Data

        #region Command

        public void Handle(EntityInsertedEvent<Command> message)
        {
            this.Notify(message);
        }

        public void Handle(EntityUpdatedEvent<Command> message)
        {
            this.Notify(message);
        }

        public void Handle(EntityDeletedEvent<Command> message)
        {
            this.Notify(message);
        }

        #endregion Command

        #region Scene

        public void Handle(EntityInsertedEvent<Scene> message)
        {
            this.Notify(message);
        }

        public void Handle(EntityUpdatedEvent<Scene> message)
        {
            this.Notify(message);
        }

        public void Handle(EntityDeletedEvent<Scene> message)
        {
            this.Notify(message);
        }

        #endregion Scene

        private void TiggerHandle(BaseEvent<Data> message)
        {
            try
            {
                foreach (var tigger in this._sceneTiggerService.GetSceneTiggers())
                {
                    var data = message.Data;
                    if (tigger.DataId == data.Id)
                    {
                        try
                        {
                            var condition = tigger.Condition;
                            var value = data.Value;
                            var engine = new Engine().Execute($"function valid(value){{return {condition};}}");
                            var result = engine.Invoke("valid", value);
                            if (result == JsValue.True)
                            {
                                this.TiggerHandle(tigger.Id);
                            }
                        }
                        catch (Exception ex)
                        {
                            this._logger.LogError(ex.ToString());
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                this._logger.LogError(ex.ToString());
            }
        }

        private void Notify<T>(BaseEvent<T> message)
        {
            Task.Run(() =>
            {
                try
                {
                    _hub.ServerToClient($"{typeof(T).Name}{message.Arg}", message.Data, "page", null);
                }
                catch (Exception ex)
                {
                    ex.PrintStack();
                }
            });
        }

        private void TiggerHandle(Guid tiggerId)
        {
            this._logger.LogDebug($"global tigger exec:{tiggerId}");
            var tigger = this._sceneTiggerRepo.ReadOnlyTable()
                .Include(o => o.OrganScene).ThenInclude(o => o.OrganSceneCommands).ThenInclude(o => o.Command).ThenInclude(o => o.Device).ThenInclude(o => o.Node)
                .FirstOrDefault(o => o.Id == tiggerId);
            foreach (var sceneCommand in tigger.OrganScene.OrganSceneCommands.OrderBy(o => o.Command.DisplayOrder))
            {
                try
                {
                    this._hub.ServerToClient(Methods.ExecCommand, sceneCommand.CommandId, sceneCommand.Command.Device.Node.Number, null);
                    Delay(_settingService, sceneCommand.Command.Delay);
                }
                catch (Exception ex)
                {
                    ex.PrintStack();
                }
            }
        }

        private void LogData(Data data)
        {
            var url = _cfg["influxdb:url"];
            var usr = _cfg["influxdb:usr"];
            var pwd = _cfg["influxdb:pwd"];
            var device = _deviceRepo.ReadOnlyTable()
                .Include(o => o.Node)
                .ThenInclude(o => o.OrganNodes)
                .ThenInclude(o => o.Organ)
                .FirstOrDefault(o => o.Id == data.DeviceId);
            var organ = device.Node.OrganNodes.FirstOrDefault()?.Organ;
            Task.Run(async () =>
            {
                try
                {
                    if (string.IsNullOrEmpty(data.Value))
                    {
                        return;
                    }
                    if (data.Type != DeviceDataType.Int && data.Type != DeviceDataType.Float && data.Type != DeviceDataType.String)
                    {
                        return;
                    }
                    try
                    {
                        var dbName = "iot";
                        var measurementName = "data";
                        using var client = new InfluxClient(new Uri(url), usr, pwd);
                        await client.CreateDatabaseAsync(dbName);
                        var row = new DynamicInfluxRow
                        {
                            Timestamp = DateTimeOffset.FromUnixTimeMilliseconds(data.Timestamp).DateTime
                        };
                        if (organ != null)
                        {
                            row.Fields.Add("OrganName", organ.Name);
                            row.Fields.Add("OrganNumber", organ.Number);
                        }
                        row.Fields.Add("NodeName", device.Node.Name);
                        row.Fields.Add("NodeNumber", device.Node.Number);
                        row.Fields.Add("DeviceNumber", device.Number);
                        row.Fields.Add("DeviceName", device.Name);
                        row.Fields.Add(data.Key, this.GetDataValue(data));
                        await client.WriteAsync(dbName, measurementName, new List<DynamicInfluxRow> { row });
                    }
                    catch (Exception ex)
                    {
                        ex.PrintStack();
                    }
                }
                catch (Exception ex)
                {
                    ex.PrintStack();
                }
            });
        }

        private object GetDataValue(Data model)
        {
            return model.Type switch
            {
                DeviceDataType.Int => Convert.ToInt32(model.Value),

                DeviceDataType.Float => Convert.ToSingle(model.Value),

                _ => model.Value,
            };
        }

        public void Handle(NodeClientConnectedEvent message)
        {
            var organ = this._organRepo.Table().FirstOrDefault(o => o.Number == message.OrganNumber);
            if (organ == null)
            {
                organ = new Organ
                {
                    Name = message.OrganNumber,
                    Number = message.OrganNumber,
                    Image = "/images/classroom.png"
                };
                this._organRepo.Add(organ);
                this._organRepo.SaveChanges();
            }
            if (!this._organNodeRepo.Table().Any(o => o.Organ.Number == message.OrganNumber && o.Node.Number == message.NodeNumber))
            {
                var node = this._nodeRepo.Table().FirstOrDefault(o => o.Number == message.NodeNumber);
                if (node != null)
                {
                    this._organNodeRepo.Add(new OrganNode { NodeId = node.Id, OrganId = organ.Id });
                    this._organNodeRepo.SaveChanges();
                }
            }
        }

        public void Handle(Guid id)
        {
            GlobalTimerExec(_logger, _settingService, _organSceneTimerRepo, _hub, id);
        }

        public static void GlobalTimerExec(ILogger _logger, ISettingService _settingService, IRepository<OrganSceneTimer> _organSceneTimerRepo, IHubContext<IoTCenterHub> _hub, Guid id)
        {
            try
            {
                _logger.LogInformation($"global timer exec at {DateTime.Now.ToString("G")}:{id}");
                var timer = _organSceneTimerRepo.ReadOnlyTable()
                    .Include(o => o.OrganScene).ThenInclude(o => o.OrganSceneCommands).ThenInclude(o => o.Command).ThenInclude(o => o.Device).ThenInclude(o => o.Node)
                    .FirstOrDefault(o => o.Id == id);
                if (timer != null)
                {
                    foreach (var sceneCommand in timer.OrganScene.OrganSceneCommands.OrderBy(o => o.Command.DisplayOrder))
                    {
                        try
                        {
                            _hub.ServerToClient(Methods.ExecCommand, sceneCommand.CommandId, sceneCommand.Command.Device.Node.Number, null);
                            Delay(_settingService, sceneCommand.Command.Delay);
                        }
                        catch (Exception ex)
                        {
                            _logger.LogError(ex.ToString());
                        }
                    }
                }
                else
                {
                    _logger.LogError($"timer {id} does not exist");
                }
            }
            catch (Exception ex)
            {
                _logger.LogError(ex.ToString());
            }
        }

        public static void Delay(ISettingService _settingService, int delay)
        {
            if (int.TryParse(_settingService.GetSetting("delay").Value, out int commandDelay))
            {
                if (commandDelay > 0)
                {
                    delay += commandDelay;
                }
            }
            if (delay > 0)
            {
                Thread.Sleep(delay);
            }
        }
    }
}