gqc
257 lines
7.8 KiB
Go
257 lines
7.8 KiB
Go
package httpreader
|
|
|
|
import (
|
|
"FireLeave_tool/connect"
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http"
|
|
"os"
|
|
"sync"
|
|
"time"
|
|
|
|
"FireLeave_tool/logger"
|
|
)
|
|
|
|
// InferenceResult 推理结果
|
|
type InferenceResult struct {
|
|
MessageType string `json:"message_type"`
|
|
Serial string `json:"serial"`
|
|
At int64 `json:"at"`
|
|
Type int `json:"type"`
|
|
Params []InferenceParam `json:"params"`
|
|
}
|
|
|
|
type InferenceParam struct {
|
|
ClassIdx int `json:"class_idx"`
|
|
Name string `json:"name"`
|
|
Number int `json:"number"`
|
|
}
|
|
|
|
type HeartbeatMessage struct {
|
|
MessageType string `json:"message_type"`
|
|
SchemaVersion int `json:"schema_version"`
|
|
Serial string `json:"serial"`
|
|
ProcessUUID string `json:"process_uuid"`
|
|
Timestamp int64 `json:"timestamp"`
|
|
Status string `json:"status"`
|
|
}
|
|
|
|
type inferenceStats struct {
|
|
windowStart time.Time
|
|
count int
|
|
}
|
|
|
|
// InferenceServer 推理数据接收服务器
|
|
type InferenceServer struct {
|
|
port int
|
|
deviceUUID string
|
|
server *http.Server
|
|
dataChan chan InferenceResult
|
|
mu sync.RWMutex
|
|
running bool
|
|
statsMu sync.Mutex
|
|
stats inferenceStats
|
|
}
|
|
|
|
// NewInferenceServer 创建推理数据接收服务器
|
|
func NewInferenceServer(deviceUUID string, port int) *InferenceServer {
|
|
return &InferenceServer{
|
|
port: port,
|
|
deviceUUID: deviceUUID,
|
|
dataChan: make(chan InferenceResult, 100), // 缓冲通道，避免阻塞
|
|
running: false,
|
|
}
|
|
}
|
|
|
|
// Start 启动HTTP服务器
|
|
func (is *InferenceServer) Start() error {
|
|
is.mu.Lock()
|
|
defer is.mu.Unlock()
|
|
|
|
if is.running {
|
|
return fmt.Errorf("服务器已在运行")
|
|
}
|
|
|
|
mux := http.NewServeMux()
|
|
mux.HandleFunc("/video/post", is.handleVideoPost)
|
|
mux.HandleFunc("/heartbeat", is.handleHeartbeat)
|
|
mux.HandleFunc("/health", is.handleHealthCheck)
|
|
|
|
is.server = &http.Server{
|
|
Addr: fmt.Sprintf(":%d", is.port),
|
|
Handler: mux,
|
|
ReadTimeout: 10 * time.Second,
|
|
WriteTimeout: 10 * time.Second,
|
|
}
|
|
|
|
go func() {
|
|
logger.Logger.Printf("Start the inference data receiving server (DeviceUUID=%s, Port=%d)", is.deviceUUID, is.port)
|
|
if err := is.server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
|
|
logger.Logger.Printf("The inference data receiving server failed to start (DeviceUUID=%s): %v", is.deviceUUID, err)
|
|
}
|
|
}()
|
|
|
|
// 等待服务器启动
|
|
time.Sleep(100 * time.Millisecond)
|
|
is.running = true
|
|
|
|
logger.Logger.Printf("The inference data receiving server has been successfully started (DeviceUUID=%s, Port=%d)", is.deviceUUID, is.port)
|
|
return nil
|
|
}
|
|
|
|
// handleVideoPost 处理视频数据POST请求
|
|
func (is *InferenceServer) handleVideoPost(w http.ResponseWriter, r *http.Request) {
|
|
startTime := time.Now()
|
|
|
|
if r.Method != "POST" {
|
|
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
|
|
return
|
|
}
|
|
|
|
var result InferenceResult
|
|
if err := json.NewDecoder(r.Body).Decode(&result); err != nil {
|
|
logger.Logger.Printf("Parse inference_result JSON failed: %v (DeviceUUID=%s)", err, is.deviceUUID)
|
|
http.Error(w, `{"error":"Invalid JSON"}`, http.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
if result.MessageType != "inference_result" {
|
|
logger.Logger.Printf("Inference result validation failed: message_type=%q (DeviceUUID=%s)", result.MessageType, is.deviceUUID)
|
|
http.Error(w, `{"error":"Invalid message_type"}`, http.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
if result.Serial == "" {
|
|
logger.Logger.Printf("Inference result validation failed: serial is empty (DeviceUUID=%s)", is.deviceUUID)
|
|
http.Error(w, `{"error":"Missing serial field"}`, http.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
// 发送到处理通道
|
|
select {
|
|
case is.dataChan <- result:
|
|
w.Header().Set("Content-Type", "application/json")
|
|
w.WriteHeader(http.StatusOK)
|
|
w.Write([]byte(`{"status":"ok", "message":"data received"}`))
|
|
// 使用 startTime 记录处理时间
|
|
is.recordInferenceProcessed(time.Since(startTime))
|
|
default:
|
|
logger.Logger.Printf("Data channel full, discard data")
|
|
http.Error(w, `{"error":"Channel full"}`, http.StatusServiceUnavailable)
|
|
}
|
|
}
|
|
|
|
func (is *InferenceServer) recordInferenceProcessed(processCost time.Duration) {
|
|
is.statsMu.Lock()
|
|
defer is.statsMu.Unlock()
|
|
|
|
now := time.Now()
|
|
if is.stats.windowStart.IsZero() {
|
|
is.stats.windowStart = now
|
|
}
|
|
is.stats.count++
|
|
|
|
if now.Sub(is.stats.windowStart) >= 10*time.Second {
|
|
logger.Logger.Printf("Inference data processed summary | UUID=%s | count=%d | window=%v | last_cost=%v",
|
|
is.deviceUUID, is.stats.count, now.Sub(is.stats.windowStart).Round(time.Millisecond), processCost)
|
|
is.stats.windowStart = now
|
|
is.stats.count = 0
|
|
}
|
|
}
|
|
|
|
func (is *InferenceServer) handleHeartbeat(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != "POST" {
|
|
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
|
|
return
|
|
}
|
|
|
|
var heartbeat HeartbeatMessage
|
|
if err := json.NewDecoder(r.Body).Decode(&heartbeat); err != nil {
|
|
logger.Logger.Printf("Parse heartbeat JSON failed: %v (DeviceUUID=%s)", err, is.deviceUUID)
|
|
http.Error(w, `{"error":"Invalid JSON"}`, http.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
if heartbeat.MessageType != "heartbeat" {
|
|
logger.Logger.Printf("Heartbeat validation failed: message_type=%q (DeviceUUID=%s)", heartbeat.MessageType, is.deviceUUID)
|
|
http.Error(w, `{"error":"Invalid message_type"}`, http.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
if heartbeat.SchemaVersion != 1 {
|
|
logger.Logger.Printf("Heartbeat validation failed: schema_version=%d (DeviceUUID=%s)", heartbeat.SchemaVersion, is.deviceUUID)
|
|
http.Error(w, `{"error":"Unsupported schema_version"}`, http.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
if heartbeat.Serial == "" {
|
|
logger.Logger.Printf("Heartbeat validation failed: serial is empty (DeviceUUID=%s)", is.deviceUUID)
|
|
http.Error(w, `{"error":"Missing serial field"}`, http.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
if heartbeat.ProcessUUID == "" {
|
|
logger.Logger.Printf("Heartbeat validation failed: process_uuid is empty (DeviceUUID=%s)", is.deviceUUID)
|
|
http.Error(w, `{"error":"Missing process_uuid field"}`, http.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
if heartbeat.ProcessUUID != is.deviceUUID {
|
|
logger.Logger.Printf("Heartbeat process_uuid mismatch: got=%s expected=%s", heartbeat.ProcessUUID, is.deviceUUID)
|
|
http.Error(w, `{"error":"process_uuid mismatch"}`, http.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
connect.UpdateInferenceHeartbeat(is.deviceUUID)
|
|
now := time.Now().Format("2006/01/02 15:04:05.000")
|
|
logger.Logger.Printf("Inference heartbeat received | Time: %s | UUID: %s | Serial: %s | Status: %s",
|
|
now, is.deviceUUID, heartbeat.Serial, heartbeat.Status)
|
|
|
|
if f, err := os.OpenFile("/data/heartbeat.log", os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644); err == nil {
|
|
defer f.Close()
|
|
fmt.Fprintf(f, "[%s] HEARTBEAT OK | %s | serial=%s | status=%s | ts=%d\n", now, is.deviceUUID, heartbeat.Serial, heartbeat.Status, heartbeat.Timestamp)
|
|
}
|
|
|
|
w.Header().Set("Content-Type", "application/json")
|
|
w.WriteHeader(http.StatusOK)
|
|
w.Write([]byte(`{"status":"ok","message":"heartbeat received"}`))
|
|
}
|
|
|
|
func (is *InferenceServer) handleHealthCheck(w http.ResponseWriter, r *http.Request) {
|
|
w.Header().Set("Content-Type", "application/json")
|
|
|
|
is.mu.RLock()
|
|
running := is.running
|
|
is.mu.RUnlock()
|
|
|
|
if running {
|
|
w.WriteHeader(http.StatusOK)
|
|
w.Write([]byte(`{
|
|
"status": "healthy",
|
|
"device_uuid": "` + is.deviceUUID + `",
|
|
"port": ` + fmt.Sprintf("%d", is.port) + `,
|
|
"timestamp": "` + time.Now().Format(time.RFC3339) + `"
|
|
}`))
|
|
} else {
|
|
w.WriteHeader(http.StatusServiceUnavailable)
|
|
w.Write([]byte(`{"status": "unavailable"}`))
|
|
}
|
|
}
|
|
|
|
func (is *InferenceServer) GetDataChan() <-chan InferenceResult {
|
|
return is.dataChan
|
|
}
|
|
|
|
// StopWithContext 带上下文的停止方法
|
|
func (s *InferenceServer) StopWithContext(ctx context.Context) {
|
|
if s.server != nil {
|
|
// 使用带超时的关闭
|
|
shutdownCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
|
|
defer cancel()
|
|
|
|
s.server.Shutdown(shutdownCtx)
|
|
logger.Logger.Printf("The HTTP server has been shut down (DeviceUUID=%s)", s.deviceUUID)
|
|
}
|
|
}
|