新增推送功能
This commit is contained in:
parent
3768057203
commit
61e2d780ce
17
angle.go
17
angle.go
@ -116,8 +116,21 @@ func parseBatchData(data string) ([]SensorReading, error) {
|
||||
// 保存批量传感器数据到数据库
|
||||
func SaveBatchSensorData(readings []SensorReading) error {
|
||||
for _, reading := range readings {
|
||||
// 使用现有的SaveSensorData函数保存每个传感器的数据
|
||||
if err := SaveSensorData(reading.SensorID, reading.X, reading.Y, reading.Z, reading.Temperature); err != nil {
|
||||
deviceID := 0
|
||||
if idx := strings.Index(reading.SerialNumber, "-"); idx > 0 {
|
||||
idPart := reading.SerialNumber[:idx]
|
||||
if v, err := strconv.Atoi(idPart); err == nil {
|
||||
deviceID = v
|
||||
}
|
||||
}
|
||||
|
||||
if deviceID > 0 {
|
||||
if err := EnsureDeviceExists(deviceID); err != nil {
|
||||
return fmt.Errorf("确保设备 %d 存在失败: %v", deviceID, err)
|
||||
}
|
||||
}
|
||||
|
||||
if err := SaveSensorData(reading.SensorID, reading.X, reading.Y, reading.Z, reading.Temperature, deviceID); err != nil {
|
||||
return fmt.Errorf("保存传感器 %d 数据失败: %v", reading.SensorID, err)
|
||||
}
|
||||
}
|
||||
|
||||
147
clients.go
Normal file
147
clients.go
Normal file
@ -0,0 +1,147 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// 客户端信息结构
|
||||
type ClientInfo struct {
|
||||
IP string // IP地址
|
||||
Port string // 端口
|
||||
LastSeen time.Time // 最后活跃时间
|
||||
IsConnected bool // 是否当前连接
|
||||
}
|
||||
|
||||
// 客户端列表（使用互斥锁保护的映射）
|
||||
var (
|
||||
clientsMutex sync.Mutex
|
||||
clients = make(map[string]*ClientInfo)
|
||||
)
|
||||
|
||||
// addClient 添加客户端
|
||||
func addClient(addr string) {
|
||||
clientsMutex.Lock()
|
||||
defer clientsMutex.Unlock()
|
||||
|
||||
host, port, err := net.SplitHostPort(addr)
|
||||
if err != nil {
|
||||
Logger.Printf("解析客户端地址失败 %s: %v", addr, err)
|
||||
host = addr
|
||||
port = "unknown"
|
||||
}
|
||||
|
||||
clients[addr] = &ClientInfo{
|
||||
IP: host,
|
||||
Port: port,
|
||||
LastSeen: time.Now(),
|
||||
IsConnected: true,
|
||||
}
|
||||
|
||||
Logger.Printf("添加新客户端: %s", addr)
|
||||
}
|
||||
|
||||
// updateClientLastSeen 更新客户端最后活跃时间
|
||||
func updateClientLastSeen(addr string) {
|
||||
clientsMutex.Lock()
|
||||
defer clientsMutex.Unlock()
|
||||
|
||||
if client, exists := clients[addr]; exists {
|
||||
client.LastSeen = time.Now()
|
||||
}
|
||||
}
|
||||
|
||||
// removeClient 移除客户端（标记断开）
|
||||
func removeClient(addr string) {
|
||||
clientsMutex.Lock()
|
||||
defer clientsMutex.Unlock()
|
||||
|
||||
if client, exists := clients[addr]; exists {
|
||||
client.IsConnected = false
|
||||
Logger.Printf("客户端断开连接: %s", addr)
|
||||
}
|
||||
}
|
||||
|
||||
// getAllClients 获取所有客户端信息
|
||||
func getAllClients() []map[string]interface{} {
|
||||
clientsMutex.Lock()
|
||||
defer clientsMutex.Unlock()
|
||||
|
||||
now := time.Now()
|
||||
result := make([]map[string]interface{}, 0, len(clients))
|
||||
|
||||
for addr, client := range clients {
|
||||
lastSeenDuration := now.Sub(client.LastSeen)
|
||||
|
||||
// 清理24小时前的记录
|
||||
if lastSeenDuration > 24*time.Hour {
|
||||
delete(clients, addr)
|
||||
continue
|
||||
}
|
||||
|
||||
// 连接状态判断：当前连接且2小时内活跃为在线
|
||||
isOnline := client.IsConnected && lastSeenDuration < 2*time.Hour
|
||||
var connectionStatus string
|
||||
if isOnline {
|
||||
connectionStatus = "保持连接"
|
||||
} else if client.IsConnected {
|
||||
connectionStatus = "连接超时"
|
||||
} else {
|
||||
connectionStatus = "已断开"
|
||||
}
|
||||
|
||||
result = append(result, map[string]interface{}{
|
||||
"address": addr,
|
||||
"ip": client.IP,
|
||||
"port": client.Port,
|
||||
"lastSeen": client.LastSeen,
|
||||
"isOnline": isOnline,
|
||||
"connectionStatus": connectionStatus,
|
||||
"lastSeenFormatted": formatDuration(lastSeenDuration),
|
||||
})
|
||||
}
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
// formatDuration 格式化持续时间为友好的字符串
|
||||
func formatDuration(d time.Duration) string {
|
||||
if d < time.Minute {
|
||||
return "刚刚"
|
||||
} else if d < time.Hour {
|
||||
return fmt.Sprintf("%d分钟前", int(d.Minutes()))
|
||||
} else if d < 24*time.Hour {
|
||||
hours := int(d.Hours())
|
||||
minutes := int(d.Minutes()) % 60
|
||||
if minutes == 0 {
|
||||
return fmt.Sprintf("%d小时前", hours)
|
||||
} else {
|
||||
return fmt.Sprintf("%d小时%d分钟前", hours, minutes)
|
||||
}
|
||||
} else {
|
||||
return fmt.Sprintf("%d天前", int(d.Hours()/24))
|
||||
}
|
||||
}
|
||||
|
||||
// startClientCleanup 启动清理过期客户端的goroutine
|
||||
func startClientCleanup() {
|
||||
go func() {
|
||||
for {
|
||||
time.Sleep(1 * time.Hour) // 每小时检查一次
|
||||
|
||||
clientsMutex.Lock()
|
||||
now := time.Now()
|
||||
|
||||
for addr, client := range clients {
|
||||
if now.Sub(client.LastSeen) > 24*time.Hour {
|
||||
delete(clients, addr)
|
||||
Logger.Printf("移除过期客户端: %s", addr)
|
||||
}
|
||||
}
|
||||
|
||||
clientsMutex.Unlock()
|
||||
}
|
||||
}()
|
||||
}
|
||||
52
db.go
52
db.go
@ -14,9 +14,9 @@ const SCALING_FACTOR = 1000 // 浮点数到整数的转换因子
|
||||
|
||||
// 初始化数据库连接
|
||||
func InitDB() error {
|
||||
username := "root"
|
||||
username := "remote"
|
||||
password := "root"
|
||||
host := "localhost"
|
||||
host := "8.134.185.53"
|
||||
port := "3306"
|
||||
dbName := "probe_db"
|
||||
|
||||
@ -48,14 +48,32 @@ func CloseDB() {
|
||||
}
|
||||
}
|
||||
|
||||
// 保存传感器数据 - 将浮点值转换为整数存储，添加温度支持
|
||||
func SaveSensorData(sensorID int, x, y, z, temperature float64) error {
|
||||
// EnsureDeviceExists 确保设备存在，不存在则创建为默认配置
|
||||
func EnsureDeviceExists(deviceID int) error {
|
||||
if deviceID <= 0 {
|
||||
return nil
|
||||
}
|
||||
_, err := db.Exec(
|
||||
"INSERT INTO devices (id, forward_enable, host, port) VALUES (?, 0, NULL, NULL) ON DUPLICATE KEY UPDATE id = id",
|
||||
deviceID,
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
// 保存传感器数据 - 将浮点值转换为整数存储，添加温度与设备ID支持
|
||||
func SaveSensorData(sensorID int, x, y, z, temperature float64, deviceID int) error {
|
||||
xInt := int(x * SCALING_FACTOR)
|
||||
yInt := int(y * SCALING_FACTOR)
|
||||
zInt := int(z * SCALING_FACTOR)
|
||||
tempInt := int(temperature * SCALING_FACTOR)
|
||||
|
||||
query := `INSERT INTO sensor_data (sensor_id, x_value, y_value, z_value, temperature) VALUES (?, ?, ?, ?, ?)`
|
||||
if deviceID > 0 {
|
||||
query := `INSERT INTO sensor_data (sensor_id, x_value, y_value, z_value, temperature, device_id) VALUES (?, ?, ?, ?, ?, ?)`
|
||||
_, err := db.Exec(query, sensorID, xInt, yInt, zInt, tempInt, deviceID)
|
||||
return err
|
||||
}
|
||||
|
||||
query := `INSERT INTO sensor_data (sensor_id, x_value, y_value, z_value, temperature, device_id) VALUES (?, ?, ?, ?, ?, NULL)`
|
||||
_, err := db.Exec(query, sensorID, xInt, yInt, zInt, tempInt)
|
||||
return err
|
||||
}
|
||||
@ -205,3 +223,27 @@ type SensorData struct {
|
||||
Temperature float64 `json:"temperature"`
|
||||
Timestamp time.Time `json:"timestamp"`
|
||||
}
|
||||
|
||||
// Device 表映射
|
||||
type Device struct {
|
||||
ID int
|
||||
ForwardEnable bool
|
||||
Host sql.NullString
|
||||
Port sql.NullInt64
|
||||
RegCodeHex sql.NullString
|
||||
}
|
||||
|
||||
// GetDevice 获取设备配置
|
||||
func GetDevice(deviceID int) (*Device, error) {
|
||||
row := db.QueryRow(`SELECT id, COALESCE(forward_enable, 0) as forward_enable, host, port, reg_code_hex FROM devices WHERE id = ?`, deviceID)
|
||||
var d Device
|
||||
var fe int
|
||||
if err := row.Scan(&d.ID, &fe, &d.Host, &d.Port, &d.RegCodeHex); err != nil {
|
||||
if err == sql.ErrNoRows {
|
||||
return nil, nil
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
d.ForwardEnable = fe != 0
|
||||
return &d, nil
|
||||
}
|
||||
|
||||
63
forwarder.go
Normal file
63
forwarder.go
Normal file
@ -0,0 +1,63 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"net"
|
||||
"time"
|
||||
)
|
||||
|
||||
// ExtractDeviceIDFromBatchRaw 从批量原始数据中提取设备ID（序列号形如 1513343-01）
|
||||
func ExtractDeviceIDFromBatchRaw(data string) int {
|
||||
for i := 0; i < len(data); i++ {
|
||||
if data[i] >= '0' && data[i] <= '9' {
|
||||
j := i
|
||||
for j < len(data) && data[j] >= '0' && data[j] <= '9' {
|
||||
j++
|
||||
}
|
||||
if j < len(data) && data[j] == '-' {
|
||||
var v int
|
||||
for k := i; k < j; k++ {
|
||||
v = v*10 + int(data[k]-'0')
|
||||
}
|
||||
return v
|
||||
}
|
||||
}
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
// ForwardRawData 将原始数据转发到设备配置的 TCP 目标
|
||||
func ForwardRawData(deviceID int, raw string) error {
|
||||
if deviceID <= 0 {
|
||||
return nil
|
||||
}
|
||||
dev, err := GetDevice(deviceID)
|
||||
if err != nil || dev == nil {
|
||||
return err
|
||||
}
|
||||
if !dev.ForwardEnable {
|
||||
return nil
|
||||
}
|
||||
if !dev.Host.Valid || dev.Host.String == "" || !dev.Port.Valid || dev.Port.Int64 <= 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
address := fmt.Sprintf("%s:%d", dev.Host.String, dev.Port.Int64)
|
||||
conn, err := net.DialTimeout("tcp", address, 5*time.Second)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer conn.Close()
|
||||
|
||||
if dev.RegCodeHex.Valid && dev.RegCodeHex.String != "" {
|
||||
bytes, decErr := hex.DecodeString(dev.RegCodeHex.String)
|
||||
if decErr == nil {
|
||||
_, _ = conn.Write(bytes)
|
||||
time.Sleep(1 * time.Second)
|
||||
}
|
||||
}
|
||||
|
||||
_, err = conn.Write([]byte(raw))
|
||||
return err
|
||||
}
|
||||
@ -20,7 +20,6 @@ func StartHTTPServer(address string) error {
|
||||
http.HandleFunc("/api/latest", handleGetLatest)
|
||||
http.HandleFunc("/api/sensors", handleGetSensors)
|
||||
http.HandleFunc("/api/clients", handleGetClients)
|
||||
http.HandleFunc("/api/trigger-query", handleTriggerQuery)
|
||||
fmt.Printf("HTTP服务器已启动，正在监听 %s\n", address)
|
||||
return http.ListenAndServe(address, nil)
|
||||
}
|
||||
@ -184,30 +183,6 @@ func handleGetClients(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
}
|
||||
|
||||
// 处理手动触发查询指令的API
|
||||
func handleTriggerQuery(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
|
||||
if r.Method != "POST" {
|
||||
http.Error(w, "只支持POST请求", http.StatusMethodNotAllowed)
|
||||
return
|
||||
}
|
||||
|
||||
// 触发向所有在线客户端发送查询指令
|
||||
sent := triggerManualQuery()
|
||||
|
||||
response := map[string]interface{}{
|
||||
"success": true,
|
||||
"message": fmt.Sprintf("已向%d个客户端发送查询指令", sent),
|
||||
"sent_count": sent,
|
||||
}
|
||||
|
||||
if err := json.NewEncoder(w).Encode(response); err != nil {
|
||||
log.Printf("错误: 编码响应JSON失败: %v", err)
|
||||
http.Error(w, "编码JSON失败："+err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
}
|
||||
|
||||
// 处理获取最新传感器数据的API
|
||||
func handleGetLatest(w http.ResponseWriter, r *http.Request) {
|
||||
log.Printf("接收到获取最新数据请求")
|
||||
|
||||
30
main.go
30
main.go
@ -8,8 +8,6 @@ import (
|
||||
)
|
||||
|
||||
func main() {
|
||||
serverType := flag.String("server", "tcp", "服务器类型: tcp 或 udp")
|
||||
tcpPort := flag.String("tcpport", "10002", "TCP服务器端口")
|
||||
udpPort := flag.String("udpport", "10002", "UDP服务器端口")
|
||||
httpPort := flag.String("httpport", "10001", "HTTP服务器端口")
|
||||
flag.Parse()
|
||||
@ -26,24 +24,14 @@ func main() {
|
||||
}
|
||||
defer CloseDB()
|
||||
|
||||
|
||||
var wg sync.WaitGroup
|
||||
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
if *serverType == "tcp" {
|
||||
Logger.Printf("使用TCP服务器模式")
|
||||
if err := StartTCPServer(":" + *tcpPort); err != nil {
|
||||
log.Fatalf("TCP服务器启动失败: %v", err)
|
||||
}
|
||||
} else if *serverType == "udp" {
|
||||
Logger.Printf("使用UDP服务器模式")
|
||||
if err := StartUDPServer(":" + *udpPort); err != nil {
|
||||
log.Fatalf("UDP服务器启动失败: %v", err)
|
||||
}
|
||||
} else {
|
||||
log.Fatalf("未知的服务器类型: %s", *serverType)
|
||||
Logger.Printf("使用UDP服务器模式")
|
||||
if err := StartUDPServer(":" + *udpPort); err != nil {
|
||||
log.Fatalf("UDP服务器启动失败: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
@ -55,15 +43,9 @@ func main() {
|
||||
}
|
||||
}()
|
||||
|
||||
if *serverType == "tcp" {
|
||||
fmt.Println("服务器已启动成功")
|
||||
fmt.Printf("- HTTP接口: http://localhost:%s\n", *httpPort)
|
||||
fmt.Printf("- TCP接口: localhost:%s\n", *tcpPort)
|
||||
} else {
|
||||
fmt.Println("服务器已启动成功")
|
||||
fmt.Printf("- HTTP接口: http://localhost:%s\n", *httpPort)
|
||||
fmt.Printf("- UDP接口: localhost:%s\n", *udpPort)
|
||||
}
|
||||
fmt.Println("服务器已启动成功")
|
||||
fmt.Printf("- HTTP接口: http://localhost:%s\n", *httpPort)
|
||||
fmt.Printf("- UDP接口: localhost:%s\n", *udpPort)
|
||||
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
106
parser.go
Normal file
106
parser.go
Normal file
@ -0,0 +1,106 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// parseData 使用正则表达式解析传感器数据，支持新格式 #{1602301014-01,1,1,28.4,-6.884,1.540}!
|
||||
func parseData(data string) (int, float64, float64, float64, float64, error) {
|
||||
// 尝试解析新格式: #{1602301014-01,1,1,28.4,-6.884,1.540}!
|
||||
newPattern := regexp.MustCompile(`#\{[^,]+-(\d+),\d+,(\d+),([-]?\d+\.\d+),([-]?\d+\.\d+),([-]?\d+\.\d+)\}!`)
|
||||
matches := newPattern.FindStringSubmatch(data)
|
||||
|
||||
if len(matches) == 6 {
|
||||
// 新格式解析
|
||||
sensorID, err := strconv.Atoi(matches[2]) // 使用传感器地址编号
|
||||
if err != nil {
|
||||
return 0, 0, 0, 0, 0, fmt.Errorf("解析传感器ID失败: %v", err)
|
||||
}
|
||||
|
||||
temperature, err := strconv.ParseFloat(strings.TrimSpace(matches[3]), 64)
|
||||
if err != nil {
|
||||
return 0, 0, 0, 0, 0, fmt.Errorf("解析温度值失败: %v", err)
|
||||
}
|
||||
|
||||
x, err := strconv.ParseFloat(strings.TrimSpace(matches[4]), 64)
|
||||
if err != nil {
|
||||
return 0, 0, 0, 0, 0, fmt.Errorf("解析X值失败: %v", err)
|
||||
}
|
||||
|
||||
y, err := strconv.ParseFloat(strings.TrimSpace(matches[5]), 64)
|
||||
if err != nil {
|
||||
return 0, 0, 0, 0, 0, fmt.Errorf("解析Y值失败: %v", err)
|
||||
}
|
||||
|
||||
z := 0.0 // 新格式没有Z值，设为0
|
||||
|
||||
return sensorID, x, y, z, temperature, nil
|
||||
}
|
||||
|
||||
// 尝试解析无包装符号的格式: 1602301014-01,1,1,31.1,-6.781,1.542
|
||||
plainPattern := regexp.MustCompile(`([^,]+)-(\d+),\d+,(\d+),([-]?\d+\.\d+),([-]?\d+\.\d+),([-]?\d+\.\d+)`)
|
||||
matches = plainPattern.FindStringSubmatch(data)
|
||||
|
||||
if len(matches) == 7 {
|
||||
// 无包装符号格式解析
|
||||
sensorID, err := strconv.Atoi(matches[3]) // 使用传感器地址编号
|
||||
if err != nil {
|
||||
return 0, 0, 0, 0, 0, fmt.Errorf("解析传感器ID失败: %v", err)
|
||||
}
|
||||
|
||||
temperature, err := strconv.ParseFloat(strings.TrimSpace(matches[4]), 64)
|
||||
if err != nil {
|
||||
return 0, 0, 0, 0, 0, fmt.Errorf("解析温度值失败: %v", err)
|
||||
}
|
||||
|
||||
x, err := strconv.ParseFloat(strings.TrimSpace(matches[5]), 64)
|
||||
if err != nil {
|
||||
return 0, 0, 0, 0, 0, fmt.Errorf("解析X值失败: %v", err)
|
||||
}
|
||||
|
||||
y, err := strconv.ParseFloat(strings.TrimSpace(matches[6]), 64)
|
||||
if err != nil {
|
||||
return 0, 0, 0, 0, 0, fmt.Errorf("解析Y值失败: %v", err)
|
||||
}
|
||||
|
||||
z := 0.0 // 这种格式没有Z值，设为0
|
||||
|
||||
return sensorID, x, y, z, temperature, nil
|
||||
}
|
||||
|
||||
// 尝试解析旧格式: 1:1.000, 2.000, 3.000
|
||||
oldPattern := regexp.MustCompile(`(\d+):([-]?\d+\.\d+),\s*([-]?\d+\.\d+),\s*([-]?\d+\.\d+)`)
|
||||
matches = oldPattern.FindStringSubmatch(data)
|
||||
|
||||
if len(matches) == 5 {
|
||||
// 旧格式解析
|
||||
sensorID, err := strconv.Atoi(matches[1])
|
||||
if err != nil {
|
||||
return 0, 0, 0, 0, 0, fmt.Errorf("解析传感器ID失败: %v", err)
|
||||
}
|
||||
|
||||
x, err := strconv.ParseFloat(strings.TrimSpace(matches[2]), 64)
|
||||
if err != nil {
|
||||
return 0, 0, 0, 0, 0, fmt.Errorf("解析X值失败: %v", err)
|
||||
}
|
||||
|
||||
y, err := strconv.ParseFloat(strings.TrimSpace(matches[3]), 64)
|
||||
if err != nil {
|
||||
return 0, 0, 0, 0, 0, fmt.Errorf("解析Y值失败: %v", err)
|
||||
}
|
||||
|
||||
z, err := strconv.ParseFloat(strings.TrimSpace(matches[4]), 64)
|
||||
if err != nil {
|
||||
return 0, 0, 0, 0, 0, fmt.Errorf("解析Z值失败: %v", err)
|
||||
}
|
||||
|
||||
temperature := 0.0 // 旧格式没有温度值，设为0
|
||||
|
||||
return sensorID, x, y, z, temperature, nil
|
||||
}
|
||||
|
||||
return 0, 0, 0, 0, 0, fmt.Errorf("数据格式不正确: %s", data)
|
||||
}
|
||||
436
tcp_server.go
436
tcp_server.go
@ -1,436 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// 客户端信息结构
|
||||
type ClientInfo struct {
|
||||
IP string // IP地址
|
||||
Port string // 端口
|
||||
LastSeen time.Time // 最后活跃时间
|
||||
IsConnected bool // 是否当前连接
|
||||
}
|
||||
|
||||
// 客户端列表（使用互斥锁保护的映射）
|
||||
var (
|
||||
clientsMutex sync.Mutex
|
||||
clients = make(map[string]*ClientInfo)
|
||||
clientConns = make(map[string]net.Conn) // 存储客户端连接
|
||||
)
|
||||
|
||||
// StartTCPServer 启动TCP服务器
|
||||
func StartTCPServer(address string) error {
|
||||
listener, err := net.Listen("tcp", address)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
startClientCleanup()
|
||||
|
||||
Logger.Printf("TCP服务器已启动，正在监听 %s\n", address)
|
||||
|
||||
for {
|
||||
conn, err := listener.Accept()
|
||||
if err != nil {
|
||||
Logger.Printf("接受连接失败: %v", err)
|
||||
continue
|
||||
}
|
||||
|
||||
go handleConnection(conn)
|
||||
}
|
||||
}
|
||||
|
||||
// handleConnection 处理客户端连接
|
||||
func handleConnection(conn net.Conn) {
|
||||
defer conn.Close()
|
||||
|
||||
remoteAddr := conn.RemoteAddr().String()
|
||||
Logger.Printf("新的客户端连接: %s", remoteAddr)
|
||||
|
||||
addClient(remoteAddr)
|
||||
|
||||
// 存储连接以便手动查询使用
|
||||
clientsMutex.Lock()
|
||||
clientConns[remoteAddr] = conn
|
||||
clientsMutex.Unlock()
|
||||
|
||||
// 注释掉自动发送指令 - 设备刚连接时立即发送一次查询指令
|
||||
go func() {
|
||||
time.Sleep(2 * time.Second) // 等待2秒让连接稳定
|
||||
command := "@1602301014A*0!\n"
|
||||
if _, err := conn.Write([]byte(command)); err != nil {
|
||||
Logger.Printf("发送连接后查询指令到客户端 %s 失败: %v", remoteAddr, err)
|
||||
} else {
|
||||
TCPDataLogger.Printf("发送连接后查询指令到客户端 %s: %s", remoteAddr, strings.TrimSpace(command))
|
||||
}
|
||||
}()
|
||||
|
||||
// 启动定时发送指令的goroutine（30分钟间隔）
|
||||
go sendPeriodicCommand(conn, remoteAddr)
|
||||
|
||||
buffer := make([]byte, 1024)
|
||||
|
||||
for {
|
||||
n, err := conn.Read(buffer)
|
||||
if err != nil {
|
||||
if err != io.EOF {
|
||||
Logger.Printf("从客户端读取失败 %s: %v", remoteAddr, err)
|
||||
} else {
|
||||
Logger.Printf("客户端断开连接 %s", remoteAddr)
|
||||
}
|
||||
removeClient(remoteAddr)
|
||||
// 清理连接映射
|
||||
clientsMutex.Lock()
|
||||
delete(clientConns, remoteAddr)
|
||||
clientsMutex.Unlock()
|
||||
break
|
||||
}
|
||||
|
||||
rawData := string(buffer[:n])
|
||||
TCPDataLogger.Printf("从客户端 %s 接收到原始数据: %s", remoteAddr, rawData)
|
||||
|
||||
// 注释掉心跳包响应 - 检查是否为心跳包 JML610
|
||||
// if strings.Contains(rawData, "JML610") {
|
||||
// TCPDataLogger.Printf("收到心跳包从客户端 %s: %s", remoteAddr, rawData)
|
||||
//
|
||||
// // 立即发送查询指令
|
||||
// command := "@1602301014A*0!\n"
|
||||
// if _, err := conn.Write([]byte(command)); err != nil {
|
||||
// Logger.Printf("响应心跳包发送查询指令到客户端 %s 失败: %v", remoteAddr, err)
|
||||
// } else {
|
||||
// TCPDataLogger.Printf("响应心跳包发送查询指令到客户端 %s: %s", remoteAddr, strings.TrimSpace(command))
|
||||
// }
|
||||
//
|
||||
// updateClientLastSeen(remoteAddr)
|
||||
// continue // 跳过数据解析，继续监听
|
||||
// }
|
||||
|
||||
// 检查是否为心跳包 JML610（仅记录，不发送查询指令）
|
||||
if strings.Contains(rawData, "JML610") {
|
||||
TCPDataLogger.Printf("收到心跳包从客户端 %s: %s", remoteAddr, rawData)
|
||||
updateClientLastSeen(remoteAddr)
|
||||
|
||||
// 发送OK响应
|
||||
resp := "OK\n"
|
||||
if _, err := conn.Write([]byte(resp)); err != nil {
|
||||
Logger.Printf("发送响应到客户端 %s 失败: %v", remoteAddr, err)
|
||||
removeClient(remoteAddr)
|
||||
// 清理连接映射
|
||||
clientsMutex.Lock()
|
||||
delete(clientConns, remoteAddr)
|
||||
clientsMutex.Unlock()
|
||||
break
|
||||
}
|
||||
continue // 跳过数据解析，继续监听
|
||||
}
|
||||
|
||||
sensorID, x, y, z, temperature, err := parseData(rawData)
|
||||
|
||||
if err == nil {
|
||||
TCPDataLogger.Printf("解析成功 - 客户端: %s, 传感器ID: %d, 值: X=%.3f, Y=%.3f, Z=%.3f, 温度=%.1f°C",
|
||||
remoteAddr, sensorID, x, y, z, temperature)
|
||||
|
||||
if err := SaveSensorData(sensorID, x, y, z, temperature); err != nil {
|
||||
Logger.Printf("保存传感器数据失败: %v", err)
|
||||
}
|
||||
} else {
|
||||
TCPDataLogger.Printf("无法解析从客户端 %s 接收到的数据: %s, 错误: %v", remoteAddr, rawData, err)
|
||||
}
|
||||
|
||||
resp := "OK\n"
|
||||
if _, err := conn.Write([]byte(resp)); err != nil {
|
||||
Logger.Printf("发送响应到客户端 %s 失败: %v", remoteAddr, err)
|
||||
removeClient(remoteAddr)
|
||||
// 清理连接映射
|
||||
clientsMutex.Lock()
|
||||
delete(clientConns, remoteAddr)
|
||||
clientsMutex.Unlock()
|
||||
break
|
||||
}
|
||||
|
||||
updateClientLastSeen(remoteAddr)
|
||||
}
|
||||
}
|
||||
|
||||
// sendPeriodicCommand 每30分钟发送一次查询指令
|
||||
func sendPeriodicCommand(conn net.Conn, remoteAddr string) {
|
||||
ticker := time.NewTicker(30 * time.Minute)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
command := "@1602301014A*0!\n"
|
||||
if _, err := conn.Write([]byte(command)); err != nil {
|
||||
Logger.Printf("发送定时指令到客户端 %s 失败: %v", remoteAddr, err)
|
||||
return // 连接断开，退出goroutine
|
||||
}
|
||||
TCPDataLogger.Printf("发送定时指令到客户端 %s: %s", remoteAddr, strings.TrimSpace(command))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// parseData 使用正则表达式解析传感器数据，支持新格式 #{1602301014-01,1,1,28.4,-6.884,1.540}!
|
||||
func parseData(data string) (int, float64, float64, float64, float64, error) {
|
||||
// 尝试解析新格式: #{1602301014-01,1,1,28.4,-6.884,1.540}!
|
||||
newPattern := regexp.MustCompile(`#\{[^,]+-(\d+),\d+,(\d+),([-]?\d+\.\d+),([-]?\d+\.\d+),([-]?\d+\.\d+)\}!`)
|
||||
matches := newPattern.FindStringSubmatch(data)
|
||||
|
||||
if len(matches) == 6 {
|
||||
// 新格式解析
|
||||
sensorID, err := strconv.Atoi(matches[2]) // 使用传感器地址编号
|
||||
if err != nil {
|
||||
return 0, 0, 0, 0, 0, fmt.Errorf("解析传感器ID失败: %v", err)
|
||||
}
|
||||
|
||||
temperature, err := strconv.ParseFloat(strings.TrimSpace(matches[3]), 64)
|
||||
if err != nil {
|
||||
return 0, 0, 0, 0, 0, fmt.Errorf("解析温度值失败: %v", err)
|
||||
}
|
||||
|
||||
x, err := strconv.ParseFloat(strings.TrimSpace(matches[4]), 64)
|
||||
if err != nil {
|
||||
return 0, 0, 0, 0, 0, fmt.Errorf("解析X值失败: %v", err)
|
||||
}
|
||||
|
||||
y, err := strconv.ParseFloat(strings.TrimSpace(matches[5]), 64)
|
||||
if err != nil {
|
||||
return 0, 0, 0, 0, 0, fmt.Errorf("解析Y值失败: %v", err)
|
||||
}
|
||||
|
||||
z := 0.0 // 新格式没有Z值，设为0
|
||||
|
||||
return sensorID, x, y, z, temperature, nil
|
||||
}
|
||||
|
||||
// 尝试解析无包装符号的格式: 1602301014-01,1,1,31.1,-6.781,1.542
|
||||
plainPattern := regexp.MustCompile(`([^,]+)-(\d+),\d+,(\d+),([-]?\d+\.\d+),([-]?\d+\.\d+),([-]?\d+\.\d+)`)
|
||||
matches = plainPattern.FindStringSubmatch(data)
|
||||
|
||||
if len(matches) == 7 {
|
||||
// 无包装符号格式解析
|
||||
sensorID, err := strconv.Atoi(matches[3]) // 使用传感器地址编号
|
||||
if err != nil {
|
||||
return 0, 0, 0, 0, 0, fmt.Errorf("解析传感器ID失败: %v", err)
|
||||
}
|
||||
|
||||
temperature, err := strconv.ParseFloat(strings.TrimSpace(matches[4]), 64)
|
||||
if err != nil {
|
||||
return 0, 0, 0, 0, 0, fmt.Errorf("解析温度值失败: %v", err)
|
||||
}
|
||||
|
||||
x, err := strconv.ParseFloat(strings.TrimSpace(matches[5]), 64)
|
||||
if err != nil {
|
||||
return 0, 0, 0, 0, 0, fmt.Errorf("解析X值失败: %v", err)
|
||||
}
|
||||
|
||||
y, err := strconv.ParseFloat(strings.TrimSpace(matches[6]), 64)
|
||||
if err != nil {
|
||||
return 0, 0, 0, 0, 0, fmt.Errorf("解析Y值失败: %v", err)
|
||||
}
|
||||
|
||||
z := 0.0 // 这种格式没有Z值，设为0
|
||||
|
||||
return sensorID, x, y, z, temperature, nil
|
||||
}
|
||||
|
||||
// 尝试解析旧格式: 1:1.000, 2.000, 3.000
|
||||
oldPattern := regexp.MustCompile(`(\d+):([-]?\d+\.\d+),\s*([-]?\d+\.\d+),\s*([-]?\d+\.\d+)`)
|
||||
matches = oldPattern.FindStringSubmatch(data)
|
||||
|
||||
if len(matches) == 5 {
|
||||
// 旧格式解析
|
||||
sensorID, err := strconv.Atoi(matches[1])
|
||||
if err != nil {
|
||||
return 0, 0, 0, 0, 0, fmt.Errorf("解析传感器ID失败: %v", err)
|
||||
}
|
||||
|
||||
x, err := strconv.ParseFloat(strings.TrimSpace(matches[2]), 64)
|
||||
if err != nil {
|
||||
return 0, 0, 0, 0, 0, fmt.Errorf("解析X值失败: %v", err)
|
||||
}
|
||||
|
||||
y, err := strconv.ParseFloat(strings.TrimSpace(matches[3]), 64)
|
||||
if err != nil {
|
||||
return 0, 0, 0, 0, 0, fmt.Errorf("解析Y值失败: %v", err)
|
||||
}
|
||||
|
||||
z, err := strconv.ParseFloat(strings.TrimSpace(matches[4]), 64)
|
||||
if err != nil {
|
||||
return 0, 0, 0, 0, 0, fmt.Errorf("解析Z值失败: %v", err)
|
||||
}
|
||||
|
||||
temperature := 0.0 // 旧格式没有温度值，设为0
|
||||
|
||||
return sensorID, x, y, z, temperature, nil
|
||||
}
|
||||
|
||||
return 0, 0, 0, 0, 0, fmt.Errorf("数据格式不正确: %s", data)
|
||||
}
|
||||
|
||||
// addClient 添加客户端
|
||||
func addClient(addr string) {
|
||||
clientsMutex.Lock()
|
||||
defer clientsMutex.Unlock()
|
||||
|
||||
host, port, err := net.SplitHostPort(addr)
|
||||
if err != nil {
|
||||
Logger.Printf("解析客户端地址失败 %s: %v", addr, err)
|
||||
host = addr
|
||||
port = "unknown"
|
||||
}
|
||||
|
||||
clients[addr] = &ClientInfo{
|
||||
IP: host,
|
||||
Port: port,
|
||||
LastSeen: time.Now(),
|
||||
IsConnected: true,
|
||||
}
|
||||
|
||||
Logger.Printf("添加新客户端: %s", addr)
|
||||
}
|
||||
|
||||
// updateClientLastSeen 更新客户端最后活跃时间
|
||||
func updateClientLastSeen(addr string) {
|
||||
clientsMutex.Lock()
|
||||
defer clientsMutex.Unlock()
|
||||
|
||||
if client, exists := clients[addr]; exists {
|
||||
client.LastSeen = time.Now()
|
||||
}
|
||||
}
|
||||
|
||||
// removeClient 移除客户端
|
||||
func removeClient(addr string) {
|
||||
clientsMutex.Lock()
|
||||
defer clientsMutex.Unlock()
|
||||
|
||||
if client, exists := clients[addr]; exists {
|
||||
// 标记为断开连接，不更新LastSeen时间
|
||||
client.IsConnected = false
|
||||
Logger.Printf("客户端断开连接: %s", addr)
|
||||
}
|
||||
}
|
||||
|
||||
// getAllClients 获取所有客户端信息
|
||||
func getAllClients() []map[string]interface{} {
|
||||
clientsMutex.Lock()
|
||||
defer clientsMutex.Unlock()
|
||||
|
||||
now := time.Now()
|
||||
result := make([]map[string]interface{}, 0, len(clients))
|
||||
|
||||
for addr, client := range clients {
|
||||
lastSeenDuration := now.Sub(client.LastSeen)
|
||||
|
||||
// 清理24小时前的记录
|
||||
if lastSeenDuration > 24*time.Hour {
|
||||
delete(clients, addr)
|
||||
continue
|
||||
}
|
||||
|
||||
// 连接状态判断：当前连接且2小时内活跃为在线
|
||||
isOnline := client.IsConnected && lastSeenDuration < 2*time.Hour
|
||||
var connectionStatus string
|
||||
if isOnline {
|
||||
connectionStatus = "保持连接"
|
||||
} else if client.IsConnected {
|
||||
connectionStatus = "连接超时"
|
||||
} else {
|
||||
connectionStatus = "已断开"
|
||||
}
|
||||
|
||||
result = append(result, map[string]interface{}{
|
||||
"address": addr,
|
||||
"ip": client.IP,
|
||||
"port": client.Port,
|
||||
"lastSeen": client.LastSeen,
|
||||
"isOnline": isOnline,
|
||||
"connectionStatus": connectionStatus,
|
||||
"lastSeenFormatted": formatDuration(lastSeenDuration),
|
||||
})
|
||||
}
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
// formatDuration 格式化持续时间为友好的字符串
|
||||
func formatDuration(d time.Duration) string {
|
||||
if d < time.Minute {
|
||||
return "刚刚"
|
||||
} else if d < time.Hour {
|
||||
return fmt.Sprintf("%d分钟前", int(d.Minutes()))
|
||||
} else if d < 24*time.Hour {
|
||||
hours := int(d.Hours())
|
||||
minutes := int(d.Minutes()) % 60
|
||||
if minutes == 0 {
|
||||
return fmt.Sprintf("%d小时前", hours)
|
||||
} else {
|
||||
return fmt.Sprintf("%d小时%d分钟前", hours, minutes)
|
||||
}
|
||||
} else {
|
||||
return fmt.Sprintf("%d天前", int(d.Hours()/24))
|
||||
}
|
||||
}
|
||||
|
||||
// startClientCleanup 启动清理过期客户端的goroutine
|
||||
func startClientCleanup() {
|
||||
go func() {
|
||||
for {
|
||||
time.Sleep(1 * time.Hour) // 每小时检查一次
|
||||
|
||||
clientsMutex.Lock()
|
||||
now := time.Now()
|
||||
|
||||
for addr, client := range clients {
|
||||
if now.Sub(client.LastSeen) > 24*time.Hour {
|
||||
delete(clients, addr)
|
||||
delete(clientConns, addr) // 同时清理连接
|
||||
Logger.Printf("移除过期客户端: %s", addr)
|
||||
}
|
||||
}
|
||||
|
||||
clientsMutex.Unlock()
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// triggerManualQuery 手动触发向所有在线客户端发送查询指令
|
||||
func triggerManualQuery() int {
|
||||
clientsMutex.Lock()
|
||||
defer clientsMutex.Unlock()
|
||||
|
||||
sentCount := 0
|
||||
command := "@1602301014A*0!\n"
|
||||
|
||||
for addr, client := range clients {
|
||||
// 检查客户端是否在线（连接状态且2小时内活跃）
|
||||
if client.IsConnected && time.Since(client.LastSeen) < 2*time.Hour {
|
||||
if conn, exists := clientConns[addr]; exists {
|
||||
if _, err := conn.Write([]byte(command)); err != nil {
|
||||
Logger.Printf("手动发送查询指令到客户端 %s 失败: %v", addr, err)
|
||||
// 连接可能已断开，从映射中移除并标记为断开
|
||||
delete(clientConns, addr)
|
||||
client.IsConnected = false
|
||||
} else {
|
||||
TCPDataLogger.Printf("手动发送查询指令到客户端 %s: %s", addr, strings.TrimSpace(command))
|
||||
sentCount++
|
||||
}
|
||||
} else {
|
||||
// 没有连接记录，标记为断开
|
||||
client.IsConnected = false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Logger.Printf("手动查询指令已发送到 %d 个客户端", sentCount)
|
||||
return sentCount
|
||||
}
|
||||
@ -3,6 +3,8 @@ package main
|
||||
import (
|
||||
"fmt"
|
||||
"net"
|
||||
"regexp"
|
||||
"strconv"
|
||||
)
|
||||
|
||||
// StartUDPServer 启动UDP服务器
|
||||
@ -55,6 +57,15 @@ func handleUDPPacket(conn *net.UDPConn, addr *net.UDPAddr, data []byte) {
|
||||
Logger.Printf("保存批量传感器数据失败: %v", err)
|
||||
}
|
||||
|
||||
// 尝试提取设备ID并透传原始数据
|
||||
if deviceID := ExtractDeviceIDFromBatchRaw(rawData); deviceID > 0 {
|
||||
go func(id int, payload string) {
|
||||
if fErr := ForwardRawData(id, payload); fErr != nil {
|
||||
Logger.Printf("转发设备 %d 数据失败: %v", id, fErr)
|
||||
}
|
||||
}(deviceID, rawData)
|
||||
}
|
||||
|
||||
// 记录第一个传感器的数据（用于兼容现有日志格式）
|
||||
firstSensor := readings[0]
|
||||
TCPDataLogger.Printf("解析成功 - UDP客户端: %s, 传感器ID: %d, 值: X=%.3f, Y=%.3f, Z=%.3f, 温度=%.1f°C",
|
||||
@ -67,9 +78,20 @@ func handleUDPPacket(conn *net.UDPConn, addr *net.UDPAddr, data []byte) {
|
||||
TCPDataLogger.Printf("解析成功 - UDP客户端: %s, 传感器ID: %d, 值: X=%.3f, Y=%.3f, Z=%.3f, 温度=%.1f°C",
|
||||
remoteAddr, sensorID, x, y, z, temperature)
|
||||
|
||||
if err := SaveSensorData(sensorID, x, y, z, temperature); err != nil {
|
||||
if err := SaveSensorData(sensorID, x, y, z, temperature, 0); err != nil {
|
||||
Logger.Printf("保存传感器数据失败: %v", err)
|
||||
}
|
||||
|
||||
// 从原始字符串尝试提取设备ID并透传
|
||||
if m := regexp.MustCompile(`([0-9]+)-\d+`).FindStringSubmatch(rawData); len(m) == 2 {
|
||||
if id, convErr := strconv.Atoi(m[1]); convErr == nil && id > 0 {
|
||||
go func(id int, payload string) {
|
||||
if fErr := ForwardRawData(id, payload); fErr != nil {
|
||||
Logger.Printf("转发设备 %d 数据失败: %v", id, fErr)
|
||||
}
|
||||
}(id, rawData)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
TCPDataLogger.Printf("无法解析从UDP客户端 %s 接收到的数据: %s, 错误: %v", remoteAddr, rawData, parseErr)
|
||||
}
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user