lighting_control/server/initialize/myData.go

package initialize

import (
	"encoding/hex"
	"fmt"
	"io"
	"net"
	"os"
	"runtime"
	"runtime/debug"
	"server/dao"
	"server/global"
	"server/modbus"
	"server/model"
	"server/service/item"
	"server/utils"
	"strconv"
	"strings"
	"sync"
	"syscall"
	"time"
)

type ModbusHandler struct {
	queue *modbus.MlQueue
}

var _handlerOnce sync.Once
var _handlerSingle *ModbusHandler

func GetHandler() *ModbusHandler {
	_handlerOnce.Do(func() {
		_handlerSingle = &ModbusHandler{
			queue: modbus.NewQueue(10000),
		}
	})
	return _handlerSingle
}

func InitInductanceTCP() {
	lis, err := net.Listen("tcp", ":60001")
	if err != nil {
		global.GVA_LOG.Error(err.Error())
	} else {
		model.InductanceTCP = lis
		global.GVA_LOG.Info(fmt.Sprintf("inductanceTCP启动成功 %v \n", lis))
	}
	go StartInductanceTCP()
}

func StartInductanceTCP() {
	global.GVA_LOG.Info("进入StartInductanceTCP")
	handler := GetHandler()
	for {
		global.GVA_LOG.Info("进入InductanceTCP循环")
		conn, err := model.InductanceTCP.Accept()
		if err != nil {
			global.GVA_LOG.Info(err.Error())
		}

		remoteAddr := conn.RemoteAddr().String()
		global.GVA_LOG.Info(fmt.Sprintf("lis Accept conn = %s\n", remoteAddr))

		// 解析远程地址
		addr, err := net.ResolveTCPAddr("tcp", remoteAddr)
		if err != nil {
			// 处理错误...
			global.GVA_LOG.Error(fmt.Sprintf("解析错误 conn = %s\n", addr.IP.String()))
		}

		model.ConnectionMap1.Store(addr.IP.String(), conn)

		//每次连接 进行设备重连操作
		item.DeviceAdjustment(addr.IP.String())
		// 使用 Load 方法尝试获取连接
		if conn1, ok := model.ConnectionMap1.Load(addr.IP.String()); ok {
			// 成功找到连接
			netConn := conn1.(net.Conn)
			// 在这里处理 netConn
			go handler.ReadAndHandle(netConn, addr.IP.String())
			go handler.Handler()
		} else {
			// 没有找到对应的连接
			global.GVA_LOG.Warn(fmt.Sprintf("启动 Connection for key %s not found", addr.IP.String()))
		}
	}
}

func (o *ModbusHandler) ReadAndHandle(conn net.Conn, remoteAddr string) {
	defer conn.Close()
	for {
		buffer := make([]byte, 1024)
		n, err := conn.Read(buffer)
		if err != nil && err != io.EOF {
			if isConnReset(err) {
				global.GVA_LOG.Error(fmt.Sprintf("连接被远程主机强制关闭conn: %s", conn.RemoteAddr().String()))
			} else if os.IsTimeout(err) {
				global.GVA_LOG.Error(fmt.Sprintf("读取操作超时conn: %s", conn.RemoteAddr().String()))
			} else {
				global.GVA_LOG.Error(fmt.Sprintf("读取错误: %s\n conn: %s", err, conn.RemoteAddr().String()))
			}

			model.ConnectionMap1.Delete(remoteAddr)
			return
		}
		queueData := model.QueueData{
			Ip:    remoteAddr,
			Value: buffer[:n],
		}
		ok, cnt := o.queue.Put(&queueData)
		if ok {
			continue
		} else {
			global.GVA_LOG.Warn(fmt.Sprintf("HandlerData:查询队列失败,队列消息数量:%d", cnt))
			runtime.Gosched()
		}
	}
}

func isConnReset(err error) bool {
	if opErr, ok := err.(*net.OpError); ok {
		if opErr.Err == syscall.ECONNRESET {
			return true // Unix-like 系统上的 ECONNRESET
		} else if runtime.GOOS == "windows" {
			// Windows 上的 WSAECONNRESET 通常是通过错误消息识别的
			if se, ok := opErr.Err.(*os.SyscallError); ok {
				if errno, ok := se.Err.(syscall.Errno); ok {
					if errno == 10054 { // 10054 对应 WSAECONNRESET
						return true
					}
				}
			} else if strings.Contains(opErr.Err.Error(), "WSAECONNRESET") {
				// 如果错误消息包含 WSAECONNRESET，也认为是连接被重置
				return true
			}
		}
	}
	return false
}

func (o *ModbusHandler) Handler() interface{} {
	defer func() {
		if err := recover(); err != nil {
			go GetHandler().Handler()
			global.GVA_LOG.Error(fmt.Sprintf("MqttHandler.Handler:发生异常:%s", string(debug.Stack())))
		}
	}()
	for {
		msg, ok, quantity := o.queue.Get()
		if !ok {
			time.Sleep(10 * time.Millisecond)
			continue
		} else if quantity > 1000 {
			global.GVA_LOG.Error(fmt.Sprintf("数据队列累积过多,请注意优化,当前队列条数:%d", quantity))
		}
		queueData, ok := msg.(*model.QueueData)
		if !ok {
			global.GVA_LOG.Error("Type assertion failed: msg is not of type model.QueueDat")
			return nil
		}
		// 信息处理返回
		parseData(queueData)
		// 对数据进行修改

	}
}

func parseData(data *model.QueueData) {
	dev, err := dao.QueryDeviceByIp(data.Ip)
	if err != nil {
		global.GVA_LOG.Error(fmt.Sprintf("Error getting register and device: %s", err))
		return
	}
	toString := hex.EncodeToString(data.Value)
	switch toString[0:2] {
	case "fe":
		switch toString[2:8] { // 开关灯
		case "050000", "050001", "050002", "050003", "050004", "050005", "050006", "050007":
			relyId, _ := strconv.Atoi(toString[7:8])
			state := 0
			if toString[8:12] == "0000" {
				state = 0
			} else if toString[8:12] == "ff00" {
				state = 1
			}
			err := dao.UpdateDeviceLoopStateByDeviceIdAndRelayId(int(dev.ID), relyId+1, state)
			if err != nil {
				global.GVA_LOG.Error(fmt.Sprintf("Error updating device loop state: %s", err))
			}

		case "0f0000":
			OperationCommand := hex.EncodeToString(modbus.OperationCommand)
			state := 0
			if OperationCommand[14:16] == "00" {
				state = 0
			} else if OperationCommand[14:16] == "ff" {
				state = 1
			}
			err := dao.UpdateDeviceLoopStateByDeviceId(int(dev.ID), state)
			if err != nil {
				global.GVA_LOG.Error(fmt.Sprintf("Error updating device loop state: %s", err))
			}
		}
		switch toString[2:6] {
		case "0101":
			// 将16进制字符串解码为字节切片
			bytes, err := hex.DecodeString(toString[6:8])
			if err != nil {
				global.GVA_LOG.Error(fmt.Sprintf("解码失败: %s", err))
				return
			}

			// 转换为二进制字符串
			binStr := ""
			for _, b := range bytes {
				// 使用fmt.Sprintf将每个字节转换为8位的二进制字符串
				binStr += fmt.Sprintf("%08b", b)
			}
			// 转换为二进制字符串
			binStr = reverseString(binStr)

			data := map[string]interface{}{
				"state":       1,
				"online_time": time.Now(),
			}

			dao.UpdateDeviceByMapAndSn(data, dev.Sn)

			loopCount := dev.LoopNumber
			if loopCount > len(binStr) {
				loopCount = len(binStr)
			}

			// 回路ID从 1 开始
			for i := 0; i < loopCount; i++ {
				relayId := i + 1 // 回路 1、2、3、4...
				state := int(binStr[i] - '0')
				dao.UpdateDeviceLoopStateByDeviceIdAndRelayId(int(dev.ID), relayId, state)
			}

			//for i, device := range reg.Devices {
			//	if device.Ip == data.Ip {
			//		reg.Devices[i].State = 1
			//		reg.Devices[i].OnlineTime = time.Now()
			//		if len(device.DeviceLoops) == 8 {
			//			for j := len(device.DeviceLoops) - 1; j >= 0; j-- {
			//				reg.Devices[i].DeviceLoops[7-j].State = int(binStr[j] - '0')
			//			}
			//		} else {
			//			for j := len(device.DeviceLoops) - 1; j >= 0; j-- {
			//				reg.Devices[i].DeviceLoops[3-j].State = int(binStr[j+4] - '0')
			//			}
			//		}
			//
			//	}
			//}
		}
	case "11":
		switch toString[2:6] {
		case "0336":
			batteryVoltage, _ := strconv.ParseInt(toString[6:10], 16, 64)
			batteryCurrent, _ := strconv.ParseInt(toString[10:14], 16, 64)
			batteryPlateVoltage, _ := strconv.ParseInt(toString[38:42], 16, 64)
			sun := dao.Sun{
				DeviceId:            dev.Sn,
				BatteryVoltage:      float64(batteryVoltage) / 100,
				BatteryCurrent:      int(batteryCurrent),
				BatteryPlateVoltage: float64(batteryPlateVoltage) / 100,
			}
			err := sun.SaveSun()
			if err != nil {
				global.GVA_LOG.Error(fmt.Sprintf("电池信息保存失败: %s", err))
				return
			}
			//电池
			if float64(batteryVoltage)/100 < 5 {
				data1 := modbus.DeviceSwitch(8, 1)
				if conn1, ok := model.ConnectionMap1.Load(data.Ip); ok {
					// 成功找到连接
					netConn := conn1.(net.Conn)
					err := utils.WriteDevice(data1, netConn)
					if err != nil {
						global.GVA_LOG.Error(fmt.Sprintf("电池符合控制 写命令错误: %s -- conn: %v", err, netConn.RemoteAddr().String()))
						return
					}
				} else {
					// 没有找到对应的连接
					global.GVA_LOG.Warn(fmt.Sprintf("电池符合控制Connection for key %s not found", data.Ip))
				}
			} else {
				data1 := modbus.DeviceSwitch(8, 0)
				if conn1, ok := model.ConnectionMap1.Load(data.Ip); ok {
					// 成功找到连接
					netConn := conn1.(net.Conn)
					err := utils.WriteDevice(data1, netConn)
					if err != nil {
						global.GVA_LOG.Error(fmt.Sprintf("电池符合控制 写命令错误: %s -- conn: %v", err, netConn.RemoteAddr().String()))
						return
					}
				} else {
					// 没有找到对应的连接
					global.GVA_LOG.Warn(fmt.Sprintf("电池符合控制Connection for key %s not found", data.Ip))
				}
			}
		}
	}

	switch toString[0:4] {
	case "4c43":
		bytes, err := hex.DecodeString(toString[4:])
		if err != nil {
			global.GVA_LOG.Error(fmt.Sprintf("Error decoding bytes: %s", err))
			return
		}

		data := map[string]interface{}{
			"state":       1,
			"online_time": time.Now(),
		}

		err = dao.UpdateDeviceByMapAndSn(data, string(bytes))
		if err != nil {
			global.GVA_LOG.Error(fmt.Sprintf("修改设备状态失败: %s", err))
		}

	}

}

// 反转字符串
func reverseString(s string) string {
	runes := []rune(s)
	for i, j := 0, len(runes)-1; i < j; i, j = i+1, j-1 {
		runes[i], runes[j] = runes[j], runes[i]
	}
	return string(runes)
}