iot/projects/IoT.Shared/Services/IoTCenter/IoTCenterHub.cs

using Application.Domain.Entities;
using Application.Models;
using Infrastructure.Data;
using Infrastructure.Events;
using Infrastructure.Extensions;
using Infrastructure.Web.SignalR;
using IoT.Shared.Services;
using Microsoft.AspNetCore.SignalR;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Configuration;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Vibrant.InfluxDB.Client;
using Vibrant.InfluxDB.Client.Rows;

namespace IoTCenter.Services
{
    public class IoTCenterHub : BasePageHub
    {
        private readonly IConfiguration _cfg;
        private readonly IEventPublisher _eventPublisher;
        private readonly IRepository<Node> _nodeRepo;
        private readonly IRepository<Device> _deviceRepo;
        private readonly IRepository<Data> _dataRepo;
        private readonly IRepository<Api> _apiRepo;
        private readonly IRepository<Command> _commandRepo;
        private readonly IRepository<Category> _categoryRepo;
        private readonly IRepository<Product> _productRepo;
        private readonly IRepository<Scene> _sceneRepo;
        private readonly IRepository<SceneCommand> _sceneCommandRepo;
        private readonly IRepository<IoTTimer> _iotTimerRepo;
        private readonly IRepository<IoTTigger> _iotTiggerRepo;
        private readonly DataService _dataService;

        public IoTCenterHub(IConfiguration cfg,
            IEventPublisher eventPublisher,
            IRepository<Node> nodeRepo,
            IRepository<Device> deviceRepo,
            IRepository<Data> dataRepo,
            IRepository<Api> apiRepo,
            IRepository<Command> commandRepo,
            IRepository<Category> categoryRepo,
            IRepository<Product> productRepo,
            IRepository<Scene> sceneRepo,
            IRepository<SceneCommand> sceneCommandRepo,
            IRepository<IoTTimer> iotTimerRepo,
            IRepository<IoTTigger> iotTiggerRepo,
            DataService dataService)
        {
            this._cfg = cfg;
            this._eventPublisher = eventPublisher;
            this._nodeRepo = nodeRepo;
            this._deviceRepo = deviceRepo;
            this._dataRepo = dataRepo;
            this._apiRepo = apiRepo;
            this._commandRepo = commandRepo;
            this._categoryRepo = categoryRepo;
            this._productRepo = productRepo;
            this._sceneRepo = sceneRepo;
            this._sceneCommandRepo = sceneCommandRepo;
            this._iotTimerRepo = iotTimerRepo;
            this._iotTiggerRepo = iotTiggerRepo;
            this._dataService = dataService;
        }

        public override void ClientToServer(string method, string message, string connectionId)
        {
            Console.WriteLine($"iot center> receive message {method} from {connectionId}");
            try
            {
                //节点上传
                if (method == Methods.UpdateNodeResponse)//接收节点
                {
                    this._dataService.Update<Node>(message);
                }
                else if (method == Methods.UpdateProductResponse)//接收产品
                {
                    this._dataService.Update<Product>(message);
                }
                else if (method == Methods.UpdateApiResponse)//接收接口
                {
                    this._dataService.Update<Api>(message);
                }
                else if (method == Methods.UpdateParameterResponse)//接收参数
                {
                    this._dataService.Update<Parameter>(message);
                }
                else if (method == Methods.UpdateDeviceIdListResponse)//接收设备Id列表
                {
                    this._dataService.UpdateList<Device>(message, o => o.Node.Number == connectionId);
                }
                else if (method == Methods.UpdateDeviceResponse)//接收设备
                {
                    this._dataService.Update<Device>(message);
                }
                else if (method == Methods.UpdateDataResponse)//接收数据
                {
                    this._dataService.Update<Data>(message);
                }
                else if (method == Methods.UpdateCommandIdListResponse)//接收命令Id列表
                {
                    this._dataService.UpdateList<Command>(message, o => o.Device.Node.Number == connectionId);
                }
                else if (method == Methods.UpdateCommandResponse)//接收命令
                {
                    this._dataService.Update<Command>(message);
                }
                else if (method == Methods.UpdateSceneIdListResponse)//接收场景Id列表
                {
                    this._dataService.UpdateList<Scene>(message, o => o.Node.Number == connectionId);
                }
                else if (method == Methods.UpdateSceneResponse)//接收场景
                {
                    this._dataService.Update<Scene>(message);
                }
                else if (method == Methods.UpdateIoTTimerIdListResponse)//接收定时器Id列表
                {
                    this._dataService.UpdateList<IoTTimer>(message, o => o.Node.Number == connectionId);
                }
                else if (method == Methods.UpdateIoTTimerResponse)//接收定时器
                {
                    this._dataService.Update<IoTTimer>(message);
                }
                else if (method == Methods.UpdateIoTTiggerIdListResponse)//接收触发器Id列表
                {
                    this._dataService.UpdateList<IoTTigger>(message, o => o.Node.Number == connectionId);
                }
                else if (method == Methods.UpdateIoTTiggerResponse)//接收触发器
                {
                    this._dataService.Update<IoTTigger>(message);
                }
                else if (method == Methods.UpdateSceneCommandIdListResponse)//接收场景命令Id列表
                {
                    this._dataService.UpdateList<SceneCommand>(message, o => o.Scene.Node.Number == connectionId);
                }
                else if (method == Methods.UpdateSceneCommandResponse)//接收场景命令
                {
                    this._dataService.Update<SceneCommand>(message);
                }
                else if (method == Methods.UpdateIoTTimerCommandIdListResponse)//接收定时器命令Id列表
                {
                    this._dataService.UpdateList<IoTTimerCommand>(message, o => o.IoTTimer.Node.Number == connectionId);
                }
                else if (method == Methods.UpdateIoTTimerCommandResponse)//接收定时器命令
                {
                    this._dataService.Update<IoTTimerCommand>(message);
                }
                else if (method == Methods.UpdateIoTTiggerCommandIdListResponse)//接收触发器命令Id列表
                {
                    this._dataService.UpdateList<IoTTiggerCommand>(message, o => o.IoTTigger.Node.Number == connectionId);
                }
                else if (method == Methods.UpdateIoTTiggerCommandResponse)//接收触发器命令
                {
                    this._dataService.Update<IoTTiggerCommand>(message);
                }
                //后台编辑
                else if (method == $"Edit{nameof(Node)}")//编辑节点返回
                {
                    var model = message.FromJson<EditNodeModel>();
                    this._dataService.Edit<Node, EditNodeModel>(model);
                    //
                    this.Clients.Group("page").SendAsync("UpdateNode", model.ToJson());
                }
                else if (method == $"Edit{nameof(Device)}")//编辑设备返回
                {
                    var model = message.FromJson<EditDeviceModel>();
                    this._dataService.Edit<Device, EditDeviceModel>(model);
                }
                else if (method == $"Delete{nameof(Device)}")//删除设备返回
                {
                    var model = message.FromJson<EditDeviceModel>();
                    this._dataService.Delete<Device, EditDeviceModel>(model);
                }
                else if (method == $"Edit{nameof(Data)}")//编辑数据或设备上报数据
                {
                    var model = message.FromJson<EditDataModel>();
                    this._dataService.Edit<Data, EditDataModel>(model);
                    this._eventPublisher.Publish(new EntityUpdatedEvent<Data>(model.To<Data>()));
                    //
                    var device = _deviceRepo.ReadOnlyTable().Include(o => o.Data).Include(o => o.Commands).Where(o => o.Id == model.DeviceId).FirstOrDefault();
                    this.Clients.Group("page").SendAsync("UpdateDevice", device.ToJson());
                }
                else if (method == $"Delete{nameof(Data)}")//删除数据返回
                {
                    var model = message.FromJson<EditDataModel>();
                    this._dataService.Delete<Data, EditDataModel>(model);
                }
                else if (method == $"Edit{nameof(Command)}")//编辑命令返回
                {
                    var model = message.FromJson<EditCommandModel>();
                    this._dataService.Edit<Command, EditCommandModel>(model);
                }
                else if (method == $"Delete{nameof(Command)}")//删除命令返回
                {
                    var model = message.FromJson<EditCommandModel>();
                    this._dataService.Delete<Command, EditCommandModel>(model);
                }
                else if (method == $"Edit{nameof(Scene)}")//编辑场景返回
                {
                    var model = message.FromJson<EditSceneModel>();
                    this._dataService.Edit<Scene, EditSceneModel>(model);
                }
                else if (method == $"Delete{nameof(Scene)}")//删除场景返回
                {
                    var model = message.FromJson<EditSceneModel>();
                    this._dataService.Delete<Scene, EditSceneModel>(model);
                }
                else if (method == $"Edit{nameof(IoTTimer)}")
                {
                    var model = message.FromJson<EditIoTTimerModel>();
                    this._dataService.Edit<IoTTimer, EditIoTTimerModel>(model);
                }
                else if (method == $"Delete{nameof(IoTTimer)}")
                {
                    var model = message.FromJson<EditIoTTimerModel>();
                    this._dataService.Delete<IoTTimer, EditIoTTimerModel>(model);
                }
                else if (method == $"Edit{nameof(IoTTigger)}")
                {
                    var model = message.FromJson<EditIoTTiggerModel>();
                    this._dataService.Edit<IoTTigger, EditIoTTiggerModel>(model);
                }
                else if (method == $"Delete{nameof(IoTTigger)}")
                {
                    var model = message.FromJson<EditIoTTiggerModel>();
                    this._dataService.Delete<IoTTigger, EditIoTTiggerModel>(model);
                }
                else if (method == $"Edit{nameof(SceneCommand)}")
                {
                    var model = message.FromJson<EditSceneCommandModel>();
                    this._dataService.Edit<SceneCommand, EditSceneCommandModel>(model);
                }
                else if (method == $"Delete{nameof(SceneCommand)}")
                {
                    var model = message.FromJson<EditSceneCommandModel>();
                    this._dataService.Delete<SceneCommand, EditSceneCommandModel>(model);
                }
                else if (method == $"Edit{nameof(IoTTimerCommand)}")
                {
                    var model = message.FromJson<EditIoTTimerCommandModel>();
                    this._dataService.Edit<IoTTimerCommand, EditIoTTimerCommandModel>(model);
                }
                else if (method == $"Delete{nameof(IoTTimerCommand)}")
                {
                    var model = message.FromJson<EditIoTTimerCommandModel>();
                    this._dataService.Delete<IoTTimerCommand, EditIoTTimerCommandModel>(model);
                }
                else if (method == $"Edit{nameof(IoTTiggerCommand)}")
                {
                    var model = message.FromJson<EditIoTTiggerCommandModel>();
                    this._dataService.Edit<IoTTiggerCommand, EditIoTTiggerCommandModel>(model);
                }
                else if (method == $"Delete{nameof(IoTTiggerCommand)}")
                {
                    var model = message.FromJson<EditIoTTiggerCommandModel>();
                    this._dataService.Delete<IoTTiggerCommand, EditIoTTiggerCommandModel>(model);
                }
            }
            catch (Exception ex)
            {
                ex.PrintStack();
            }
        }

        [System.Diagnostics.CodeAnalysis.SuppressMessage("Design", "CA1031:不捕获常规异常类型", Justification = "<挂起>")]
        private void UpdateSceneCommand(string message)
        {
            try
            {
                Console.WriteLine("iot center> receive edit scene command message");
                var model = message.FromJson<EditSceneCommandModel>();
                var sceneCommand = _sceneCommandRepo.Table().FirstOrDefault(o => o.Id == model.Id);
                if (sceneCommand == null)
                {
                    sceneCommand = new SceneCommand
                    {
                        Id = model.Id
                    };
                    _sceneCommandRepo.Add(sceneCommand);
                }
                sceneCommand.SceneId = model.SceneId.Value;
                sceneCommand.CommandId = model.CommandId.Value;
                _sceneCommandRepo.SaveChanges();
            }
            catch (Exception ex)
            {
                ex.PrintStack();
            }
        }

        [System.Diagnostics.CodeAnalysis.SuppressMessage("Design", "CA1031:不捕获常规异常类型", Justification = "<挂起>")]
        private void DeleteSceneCommand(string message)
        {
            try
            {
                Console.WriteLine("iot center> receive delete scene command message");
                var model = message.FromJson<EditSceneCommandModel>();
                var sceneCommand = _sceneCommandRepo.Table().FirstOrDefault(o => o.Id == model.Id);
                if (sceneCommand != null)
                {
                    _sceneCommandRepo.Delete(sceneCommand);
                    _sceneCommandRepo.SaveChanges();
                }
            }
            catch (Exception ex)
            {
                ex.PrintStack();
            }
        }

        public void ApiCallback(string message, string connectionId)
        {
            if (!string.IsNullOrEmpty(connectionId))
            {
                this.ServerToClient(connectionId, Methods.ApiCallback, message);
            }
        }

        [System.Diagnostics.CodeAnalysis.SuppressMessage("Design", "CA1031:不捕获常规异常类型", Justification = "<挂起>")]
        private async Task LogToInfluxdbAsync(EditDataModel model, string deviceNumber)
        {
            if (string.IsNullOrEmpty(model.Value))
            {
                return;
            }
            if (model.Type != DeviceDataType.Int && model.Type != DeviceDataType.Float)
            {
                return;
            }
            try
            {
                var url = _cfg["influxdb:url"];
                var usr = _cfg["influxdb:usr"];
                var pwd = _cfg["influxdb:pwd"];
                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 = DateTime.UtcNow
                };
                row.Fields.Add("DeviceNumber", deviceNumber);
                row.Fields.Add("DeviceName", model.Name);
                row.Fields.Add(model.Key, this.GetDataValue(model));

                await client.WriteAsync(dbName, measurementName, new List<DynamicInfluxRow> { row });
            }
            catch (Exception ex)
            {
                ex.PrintStack();
            }
        }

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

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

                _ => model.Value,
            };
        }
    }
}