iot/projects/IoTCenter/Services/PageHub.cs

using Application.Domain.Entities;
using Application.Models;
using Infrastructure.Data;
using Infrastructure.Extensions;
using Infrastructure.Web.SignalR;
using IoT.Shared.Services;
using Microsoft.AspNetCore.SignalR;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
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 PageHub : BasePageHub
    {
        private readonly IServiceProvider _applicationService;

        public PageHub(IServiceProvider applicationService)
        {
            this._applicationService = applicationService;
        }

        public override void ClientToServer(string method, string message, string connectionId)
        {
            Console.WriteLine($"iot center> receive message from {connectionId}");
            if (method == Methods.HealthCheckResponse)
            {
                using (var scope = this._applicationService.CreateScope())
                {
                    var nodeRepo = scope.ServiceProvider.GetService<IRepository<Node>>();
                    var node = nodeRepo.Table().FirstOrDefault(o => o.Number == message);
                    node.IsOnline = true;
                    nodeRepo.SaveChanges();
                }
            }
            else if (method == Methods.EditNodeResponse)//上报节点
            {
                this.UpdateNode(message);
            }
            else if (method == Methods.EditProductResponse)//上报产品
            {
                this.UpdateProduct(message);
            }
            else if (method == Methods.EditDeviceResponse)//上报设备
            {
                this.UpdateDevice(message);
            }
            else if (method == Methods.EditDataResponse)//上报数据
            {
                this.UpdateData(message);
            }
            //
            else if (method == Methods.EditNodeResponse)
            {
                this.UpdateNode(message);
            }
            else if (method == Methods.EditDeviceResponse)
            {
                this.UpdateDevice(message);
            }
            else if (method == Methods.DeleteDeviceResponse)
            {
                this.DeleteDevice(message);
            }
            else if (method == Methods.DeleteDataResponse)
            {
                this.DeleteData(message);
            }
            else if (method == Methods.EditSceneResponse)
            {
                this.UpdateScene(message);
            }
            else if (method == Methods.DeleteSceneResponse)
            {
                this.DeleteScene(message);
            }
            else if (method == Methods.EditCommandResponse)
            {
                this.UpdateCommand(message);
            }
            else if (method == Methods.DeleteCommandResponse)
            {
                this.DeleteCommand(message);
            }
        }

        private void UpdateNode(string message)
        {
            try
            {
                Console.WriteLine("iot center> receive node message");
                var nodeDto = message.FromJson<EditNodeModel>();
                using (var scope = this._applicationService.CreateScope())
                {
                    var nodeRepo = scope.ServiceProvider.GetService<IRepository<Node>>();
                    var node = nodeRepo.Table().FirstOrDefault(o => o.Number == nodeDto.Number);
                    if (node == null)
                    {
                        node = new Node();
                        nodeRepo.Add(node);
                    }
                    node.FromDto(nodeDto);
                    nodeRepo.SaveChanges();
                    this.Clients.Group("page").SendAsync("UpdateNode", message);
                }
            }
            catch (Exception ex)
            {
                ex.PrintStack();
            }
        }

        private void UpdateProduct(string message)
        {
            try
            {
                Console.WriteLine("iot center> receive product message");
                var productDto = message.FromJson<EditProductModel>();
                using (var scope = this._applicationService.CreateScope())
                {
                    var categoryRepo = scope.ServiceProvider.GetService<IRepository<Category>>();
                    var category = categoryRepo.ReadOnlyTable().FirstOrDefault(o => o.Number == productDto.CategoryNumber);
                    var productRepo = scope.ServiceProvider.GetService<IRepository<Product>>();
                    var product = productRepo.Table().FirstOrDefault(o => o.Number == productDto.Number);
                    if (product == null)
                    {
                        product = new Product().FromDto(productDto);
                        product.CategoryId = category.Id;
                        productRepo.Add(product);
                        OpenApiService.UpdateApi(product);
                        productRepo.SaveChanges();
                    }
                }
            }
            catch (Exception ex)
            {
                ex.PrintStack();
            }
        }

        private void UpdateDevice(string message)
        {
            try
            {
                Console.WriteLine("iot center> receive device message");
                var deviceDto = message.FromJson<EditDeviceModel>();
                using (var scope = this._applicationService.CreateScope())
                {
                    var productRepo = scope.ServiceProvider.GetService<IRepository<Product>>();
                    var product = productRepo.Table().FirstOrDefault(o => o.Number == deviceDto.ProductNumber);
                    if (product == null)
                    {
                        this.ServerToClient(deviceDto.ConnectId, "GetDeviceInfo", deviceDto.ProductNumber);
                        throw new Exception("need device info");
                    }
                    var nodeRepo = scope.ServiceProvider.GetService<IRepository<Node>>();
                    var node = nodeRepo.Table().FirstOrDefault(o => o.Number == deviceDto.NodeNumber);
                    if (node == null)
                    {
                        node = new Node
                        {
                            Number = deviceDto.NodeNumber,
                            Name = deviceDto.NodeNumber
                        };
                        nodeRepo.Add(node);
                        nodeRepo.SaveChanges();
                    }
                    var deviceRepo = scope.ServiceProvider.GetService<IRepository<Device>>();
                    var device = deviceRepo.Table().FirstOrDefault(o => o.Number == deviceDto.Number);
                    if (device == null)
                    {
                        device = new Device();
                        device.DisplayName = device.Name;
                        device.ProductId = product.Id;
                        device.NodeId = node.Id;
                        deviceRepo.Add(device);
                    }
                    device.FromDto(deviceDto);
                    deviceRepo.SaveChanges();
                    var device2 = deviceRepo.ReadOnlyTable().Include(o => o.Data).FirstOrDefault(o => o.Number == deviceDto.Number);
                    this.Clients.Group("page").SendAsync("UpdateDevice", device2.ToJson());
                }
            }
            catch (Exception ex)
            {
                ex.PrintStack();
            }
        }

        private void DeleteDevice(string message)
        {
            try
            {
                Console.WriteLine("iot center> receive device message");
                using (var scope = this._applicationService.CreateScope())
                {
                    var deviceRepo = scope.ServiceProvider.GetService<IRepository<Device>>();
                    var model = message.FromJson<EditDeviceModel>();
                    var device = deviceRepo.Table().FirstOrDefault(o => o.Number == model.Number);
                    if (device != null)
                    {
                        deviceRepo.Delete(device);
                        deviceRepo.SaveChanges();
                        this.Clients.Group("page").SendAsync("DeleteDevice", model.Number);
                    }
                }
            }
            catch (Exception ex)
            {
                ex.PrintStack();
            }
        }

        private void UpdateData(string message)
        {
            Console.WriteLine("iot center> receive data message");
            var dataDtoList = message.FromJson<List<EditDataModel>>();
            if (dataDtoList.Count > 0)
            {
                using (var scope = this._applicationService.CreateScope())
                {
                    var number = dataDtoList.FirstOrDefault().DeviceNumber;
                    var deviceRepo = scope.ServiceProvider.GetService<IRepository<Device>>();
                    var device = deviceRepo.Table().Include(o => o.Data).FirstOrDefault(o => o.Number == number);
                    if (device != null)
                    {
                        foreach (var dataDto in dataDtoList)
                        {
                            var data = device.Data.FirstOrDefault(o => o.Key == dataDto.Key);
                            if (data == null)
                            {
                                data = new Data();
                                device.Data.Add(data);
                            }
                            data.FromDto(dataDto);
                        }
                        deviceRepo.SaveChanges();
                        this.Clients.Group("page").SendAsync("UpdateDevice", device.ToJson());
                        foreach (var dataDto in dataDtoList)
                        {
                            this.LogToInfluxdbAsync(dataDto).Wait(0);
                        }
                    }
                }
            }
        }

        private void DeleteData(string message)
        {
            try
            {
                Console.WriteLine("iot center> receive delete data message");
                using (var scope = this._applicationService.CreateScope())
                {
                    var dataRepo = scope.ServiceProvider.GetService<IRepository<Data>>();
                    var model = message.FromJson<EditDataModel>();
                    var data = dataRepo.Table().FirstOrDefault(o => o.Device.Number == model.DeviceNumber && o.Key == model.Key);
                    if (data != null)
                    {
                        dataRepo.Delete(data);
                        dataRepo.SaveChanges();
                        this.Clients.Group("page").SendAsync("DeleteData", message);
                    }
                }
            }
            catch (Exception ex)
            {
                ex.PrintStack();
            }
        }

        private void UpdateScene(string message)
        {
            try
            {
                Console.WriteLine("iot center> receive edit scene message");
                using (var scope = this._applicationService.CreateScope())
                {
                    var nodeRepo = scope.ServiceProvider.GetService<IRepository<Node>>();
                    var sceneRepo = scope.ServiceProvider.GetService<IRepository<Scene>>();
                    var model = message.FromJson<EditSceneModel>();
                    var scene = sceneRepo.Table().FirstOrDefault(o => o.Node.Number == model.NodeNumber && o.Name == model.Name);
                    if (scene == null)
                    {
                        scene = new Scene();
                        scene.NodeId = nodeRepo.ReadOnlyTable().FirstOrDefault(o => o.Number == model.NodeNumber).Id;
                        sceneRepo.Add(scene);
                    }
                    scene.FromDto(model);
                    sceneRepo.SaveChanges();
                    this.Clients.Group("page").SendAsync("UpdateScene", message);
                }
            }
            catch (Exception ex)
            {
                ex.PrintStack();
            }
        }

        private void DeleteScene(string message)
        {
            try
            {
                Console.WriteLine("iot center> receive delete scene message");
                using (var scope = this._applicationService.CreateScope())
                {
                    var sceneRepo = scope.ServiceProvider.GetService<IRepository<Scene>>();
                    var model = message.FromJson<EditSceneModel>();
                    var scene = sceneRepo.Table().FirstOrDefault(o => o.Node.Number == model.NodeNumber && o.Name == model.Name);
                    if (scene != null)
                    {
                        sceneRepo.Delete(scene);
                        sceneRepo.SaveChanges();
                        this.Clients.Group("page").SendAsync("DeleteScene", message);
                    }
                }
            }
            catch (Exception ex)
            {
                ex.PrintStack();
            }
        }

        private void UpdateCommand(string message)
        {
            try
            {
                Console.WriteLine("iot center> receive edit command message");
                using (var scope = this._applicationService.CreateScope())
                {
                    var model = message.FromJson<EditCommandModel>();
                    var commandRepo = scope.ServiceProvider.GetService<IRepository<Command>>();
                    var sceneRepo = scope.ServiceProvider.GetService<IRepository<Scene>>();
                    var apiRepo = scope.ServiceProvider.GetService<IRepository<Api>>();
                    var deviceRepo = scope.ServiceProvider.GetService<IRepository<Device>>();
                    var device = deviceRepo.ReadOnlyTable().FirstOrDefault(o => o.Number == model.DeviceNumber);

                    var command = commandRepo.Table()
                        .Where(o => o.Device.Number == model.DeviceNumber)
                        .Where(o => o.Api.ProductId == device.ProductId && o.Api.Name == model.ApiName)
                        .Where(o => o.QueryString == model.QueryString)
                        .FirstOrDefault();
                    if (command == null)
                    {
                        command = new Command();
                        command.SceneId = sceneRepo.ReadOnlyTable().FirstOrDefault(o => o.Name == model.SceneName).Id;
                        command.DeviceId = device.Id;
                        command.ApiId = apiRepo.ReadOnlyTable().Where(o => o.ProductId == device.ProductId && o.Name == model.ApiName).FirstOrDefault().Id;
                        commandRepo.Add(command);
                    }
                    command.FromDto(model);
                    commandRepo.SaveChanges();
                }
            }
            catch (Exception ex)
            {
                ex.PrintStack();
            }
        }

        private void DeleteCommand(string message)
        {
            try
            {
                Console.WriteLine("iot center> receive delete scene message");
                using (var scope = this._applicationService.CreateScope())
                {
                    var commandRepo = scope.ServiceProvider.GetService<IRepository<Command>>();
                    var model = message.FromJson<EditCommandModel>();
                    var deviceRepo = scope.ServiceProvider.GetService<IRepository<Device>>();
                    var device = deviceRepo.ReadOnlyTable().FirstOrDefault(o => o.Number == model.DeviceNumber);
                    var command = commandRepo.Table()
                        .Where(o => o.Scene.Name == model.SceneName)
                        .Where(o => o.Device.Number == model.DeviceNumber)
                        .Where(o => o.Api.Name == model.ApiName)
                        .Where(o => o.QueryString == model.QueryString)
                        .FirstOrDefault();
                    if (command != null)
                    {
                        commandRepo.Delete(command);
                        commandRepo.SaveChanges();
                    }
                }
            }
            catch (Exception ex)
            {
                ex.PrintStack();
            }
        }

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

        private async Task LogToInfluxdbAsync(EditDataModel model)
        {
            if (string.IsNullOrEmpty(model.Value))
            {
                return;
            }
            if (model.Type != DeviceDataType.Int && model.Type != DeviceDataType.Float)
            {
                return;
            }
            using (var scope = this._applicationService.CreateScope())
            {
                try
                {
                    var cfg = scope.ServiceProvider.GetService<IConfiguration>();
                    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", model.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)
        {
            switch (model.Type)
            {
                case DeviceDataType.Int:
                    return Convert.ToInt32(model.Value);

                case DeviceDataType.Float:
                    return Convert.ToSingle(model.Value);

                default:
                    return model.Value;
            }
        }
    }
}