diff --git a/bin/fireLeave-tool-app-linux-arm64 b/bin/fireleave_tool similarity index 54% rename from bin/fireLeave-tool-app-linux-arm64 rename to bin/fireleave_tool index f5ff4aa..ff98ef5 100644 Binary files a/bin/fireLeave-tool-app-linux-arm64 and b/bin/fireleave_tool differ diff --git a/connect/device_storage.go b/connect/device_storage.go index 578782d..23f52e6 100644 --- a/connect/device_storage.go +++ b/connect/device_storage.go @@ -6,6 +6,7 @@ import ( "fmt" "os" "path/filepath" + "strconv" "strings" ) @@ -25,15 +26,15 @@ type Point struct { // ServiceConfig 服务配置结构体 type ServiceConfig struct { - HostUUID string `json:"host_uuid"` - DeviceUUID string `json:"device_uuid"` - Address string `json:"address"` - TaskID string `json:"task_id"` - CameraRTSP string `json:"camera_rtsp"` - CameraIP string `json:"camera_ip"` - Confidence int `json:"confidence"` + HostUUID string `json:"host_uuid"` + DeviceUUID string `json:"device_uuid"` + Address string `json:"address"` + TaskID string `json:"task_id"` + CameraRTSP string `json:"camera_rtsp"` + CameraIP string `json:"camera_ip"` + Confidence int `json:"confidence"` DetectArea DetectAreaConfig `json:"detect_area"` - Param ParamConfig `json:"param"` + Param ParamConfig `json:"param"` } // DetectAreaConfig 检测区域配置 @@ -56,19 +57,21 @@ type FireCheckParam struct { // FireLeaveParam 离岗参数 type FireLeaveParam struct { - PersonCount string `json:"person_count"` - Confidence string `json:"confidence"` - DetectionTime string `json:"detection_time"` + PersonCount string `json:"person_count"` + Confidence string `json:"confidence"` + DetectionTime string `json:"detection_time"` TemperatureThreshold string `json:"temperature_threshold"` - AlarmTime string `json:"alarm_time"` + AlarmTime string `json:"alarm_time"` } var ( - devicesBaseDir = "/data/devices" - deviceDataList []DeviceData + devicesBaseDir = "/data/devices" + deviceDataList []DeviceData serviceConfigFile = "/usr/local/etc/service.conf" ) +const fixedUploadURL = "http://172.17.0.1/api/v1/upload/video" + // LoadServiceConfig loads multiple device data from service config file func LoadServiceConfig() ([]DeviceData, error) { if _, err := os.Stat(serviceConfigFile); os.IsNotExist(err) { @@ -88,35 +91,43 @@ func LoadServiceConfig() ([]DeviceData, error) { return nil, fmt.Errorf("service config file is empty") } - // Parse as ServiceConfig array + // Parse as the canonical ServiceConfig array. Fall back to the previous + // DeviceData array cache shape so older local service.conf files still boot. var serviceConfigs []ServiceConfig if err := json.Unmarshal(data, &serviceConfigs); err != nil { - logger.Logger.Printf("Failed to parse service config file: %v", err) - return nil, err + var legacyDeviceData []DeviceData + if legacyErr := json.Unmarshal(data, &legacyDeviceData); legacyErr != nil { + logger.Logger.Printf("Failed to parse service config file: %v", err) + return nil, err + } + for i := range legacyDeviceData { + NormalizeDeviceRuntimeFields(&legacyDeviceData[i]) + } + logger.Logger.Printf("Loaded %d legacy device configurations from service config file", len(legacyDeviceData)) + return legacyDeviceData, nil } // Convert ServiceConfig to DeviceData var deviceDataList []DeviceData - const FixedUploadURL = "http://192.168.80.168/api/v1/upload/video" for _, config := range serviceConfigs { // Create DeviceData from ServiceConfig deviceData := DeviceData{ - DeviceUUID: config.DeviceUUID, - TaskID: config.TaskID, - CameraRTSP: config.CameraRTSP, - CameraIP: config.CameraIP, - Confidence: config.Confidence, - HostUUID: config.HostUUID, - Address: config.Address, - UploadURL: FixedUploadURL, + DeviceUUID: config.DeviceUUID, + TaskID: config.TaskID, + CameraRTSP: config.CameraRTSP, + CameraIP: config.CameraIP, + Confidence: config.Confidence, + HostUUID: config.HostUUID, + Address: config.Address, + UploadURL: fixedUploadURL, // Set new fields for new configuration format FireLeaveDetectArea: config.DetectArea.FireLeave, FireCheckDetectArea: config.DetectArea.FireCheck, FireLeaveParam: config.Param.FireLeave, FireCheckParam: config.Param.FireCheck, } - + // Parse and set temperature threshold from FireLeaveParam if tempThresholdStr := config.Param.FireLeave.TemperatureThreshold; tempThresholdStr != "" { var tempThreshold float64 @@ -125,6 +136,7 @@ func LoadServiceConfig() ([]DeviceData, error) { logger.Logger.Printf("Set temperature threshold from config: %.1f°C (DeviceUUID=%s)", tempThreshold, config.DeviceUUID) } } + NormalizeDeviceRuntimeFields(&deviceData) // Filter out configurations with empty DeviceUUID if deviceData.DeviceUUID != "" { @@ -141,6 +153,123 @@ func LoadServiceConfig() ([]DeviceData, error) { return deviceDataList, nil } +func DecodeDeviceDataPayload(data []byte) (DeviceData, error) { + var deviceData DeviceData + if err := json.Unmarshal(data, &deviceData); err == nil && deviceData.DeviceUUID != "" { + NormalizeDeviceRuntimeFields(&deviceData) + if deviceData.UploadURL == "" { + deviceData.UploadURL = fixedUploadURL + } + return deviceData, nil + } + + var serviceConfig ServiceConfig + if err := json.Unmarshal(data, &serviceConfig); err != nil { + return DeviceData{}, err + } + deviceData = ServiceConfigToDeviceData(serviceConfig, fixedUploadURL) + if deviceData.DeviceUUID == "" { + return DeviceData{}, fmt.Errorf("device_uuid is empty") + } + return deviceData, nil +} + +func ServiceConfigToDeviceData(config ServiceConfig, uploadURL string) DeviceData { + deviceData := DeviceData{ + DeviceUUID: config.DeviceUUID, + TaskID: config.TaskID, + CameraRTSP: config.CameraRTSP, + CameraIP: config.CameraIP, + Confidence: config.Confidence, + HostUUID: config.HostUUID, + Address: config.Address, + UploadURL: uploadURL, + FireLeaveDetectArea: config.DetectArea.FireLeave, + FireCheckDetectArea: config.DetectArea.FireCheck, + FireLeaveParam: config.Param.FireLeave, + FireCheckParam: config.Param.FireCheck, + } + NormalizeDeviceRuntimeFields(&deviceData) + return deviceData +} + +func DeviceDataToServiceConfig(deviceData DeviceData) ServiceConfig { + config := ServiceConfig{ + HostUUID: deviceData.HostUUID, + DeviceUUID: deviceData.DeviceUUID, + Address: deviceData.Address, + TaskID: deviceData.TaskID, + CameraRTSP: deviceData.CameraRTSP, + CameraIP: deviceData.CameraIP, + Confidence: deviceData.Confidence, + DetectArea: DetectAreaConfig{ + FireLeave: deviceData.FireLeaveDetectArea, + FireCheck: deviceData.FireCheckDetectArea, + }, + Param: ParamConfig{ + FireLeave: deviceData.FireLeaveParam, + FireCheck: deviceData.FireCheckParam, + }, + } + + if len(config.DetectArea.FireLeave) == 0 { + config.DetectArea.FireLeave = deviceData.DetectArea + } + if config.Param.FireLeave.DetectionTime == "" && deviceData.DetectionTime != "" { + config.Param.FireLeave.DetectionTime = deviceData.DetectionTime.String() + } + if config.Param.FireLeave.AlarmTime == "" && deviceData.AlarmTime != "" { + config.Param.FireLeave.AlarmTime = deviceData.AlarmTime.String() + } + if config.Param.FireLeave.TemperatureThreshold == "" && deviceData.TemperatureThreshold > 0 { + config.Param.FireLeave.TemperatureThreshold = strconv.FormatFloat(deviceData.TemperatureThreshold, 'f', -1, 64) + } + if config.Param.FireLeave.Confidence == "" && deviceData.Confidence > 0 { + config.Param.FireLeave.Confidence = strconv.Itoa(deviceData.Confidence) + } + if config.Param.FireCheck.Confidence == "" && deviceData.Confidence > 0 { + config.Param.FireCheck.Confidence = strconv.Itoa(deviceData.Confidence) + } + + return config +} + +func NormalizeDeviceRuntimeFields(deviceData *DeviceData) { + if deviceData == nil { + return + } + + if len(deviceData.FireLeaveDetectArea) == 0 && len(deviceData.DetectArea) > 0 { + deviceData.FireLeaveDetectArea = deviceData.DetectArea + } + if len(deviceData.DetectArea) == 0 && len(deviceData.FireLeaveDetectArea) > 0 { + deviceData.DetectArea = deviceData.FireLeaveDetectArea + } + + if deviceData.FireLeaveParam.PersonCount != "" { + if count, err := strconv.Atoi(deviceData.FireLeaveParam.PersonCount); err == nil && count >= 0 { + deviceData.PersonCount = count + } + } + if deviceData.FireLeaveParam.DetectionTime != "" { + deviceData.DetectionTime = json.Number(deviceData.FireLeaveParam.DetectionTime) + } + if deviceData.FireLeaveParam.AlarmTime != "" { + deviceData.AlarmTime = json.Number(deviceData.FireLeaveParam.AlarmTime) + } + if deviceData.FireLeaveParam.TemperatureThreshold != "" { + if threshold, err := strconv.ParseFloat(deviceData.FireLeaveParam.TemperatureThreshold, 64); err == nil && threshold > 0 { + deviceData.TemperatureThreshold = threshold + logger.Logger.Printf("Set temperature threshold from config: %.1f掳C (DeviceUUID=%s)", threshold, deviceData.DeviceUUID) + } + } + if deviceData.FireLeaveParam.Confidence != "" { + if confidence, err := strconv.Atoi(deviceData.FireLeaveParam.Confidence); err == nil && confidence > 0 { + deviceData.Confidence = confidence + } + } +} + // ProcessDetectAreaForInference processes detection area data, converts to format required by inference program func ProcessDetectAreaForInference(areas []Area) string { if len(areas) == 0 { @@ -165,20 +294,22 @@ func ProcessDetectAreaForInference(areas []Area) string { func ProcessDeviceDataForInference(data DeviceData) DeviceData { // Create a copy, only process fields that need conversion processed := data - + NormalizeDeviceRuntimeFields(&processed) + // For backward compatibility, process original DetectArea if FireLeaveDetectArea is empty if len(data.FireLeaveDetectArea) > 0 { processed.DetectAreaStr = ProcessDetectAreaForInference(data.FireLeaveDetectArea) } else { processed.DetectAreaStr = ProcessDetectAreaForInference(data.DetectArea) } - + return processed } // ProcessFireLeaveDataForInference processes device data for fire_leave inference func ProcessFireLeaveDataForInference(data DeviceData) DeviceData { processed := data + NormalizeDeviceRuntimeFields(&processed) if len(data.FireLeaveDetectArea) > 0 { processed.DetectAreaStr = ProcessDetectAreaForInference(data.FireLeaveDetectArea) } @@ -188,6 +319,7 @@ func ProcessFireLeaveDataForInference(data DeviceData) DeviceData { // ProcessFireCheckDataForInference processes device data for fire_check inference func ProcessFireCheckDataForInference(data DeviceData) DeviceData { processed := data + NormalizeDeviceRuntimeFields(&processed) if len(data.FireCheckDetectArea) > 0 { processed.DetectAreaStr = ProcessDetectAreaForInference(data.FireCheckDetectArea) } @@ -380,12 +512,25 @@ func UpdateDeviceData(data DeviceData) error { // SaveServiceConfig saves multiple device configurations to service config file func SaveServiceConfig(deviceDataList []DeviceData) error { - configData, err := json.MarshalIndent(deviceDataList, "", " ") + serviceConfigs := make([]ServiceConfig, 0, len(deviceDataList)) + for _, deviceData := range deviceDataList { + if deviceData.DeviceUUID == "" { + continue + } + serviceConfigs = append(serviceConfigs, DeviceDataToServiceConfig(deviceData)) + } + + configData, err := json.MarshalIndent(serviceConfigs, "", " ") if err != nil { logger.Logger.Printf("Failed to serialize service configuration: %v", err) return err } + if err := os.MkdirAll(filepath.Dir(serviceConfigFile), 0755); err != nil { + logger.Logger.Printf("Failed to create service config directory: %v", err) + return err + } + if err := os.WriteFile(serviceConfigFile, configData, 0644); err != nil { logger.Logger.Printf("Failed to write service config file: %v", err) return err diff --git a/connect/heartbeat.go b/connect/heartbeat.go index f8f3a41..2095a63 100644 --- a/connect/heartbeat.go +++ b/connect/heartbeat.go @@ -9,7 +9,6 @@ import ( "time" "FireLeave_tool/logger" - "FireLeave_tool/portmanager" ) var ( @@ -24,9 +23,9 @@ func UpdateInferenceHeartbeat(deviceUUID string) { inferenceLastHeartbeat.Store(deviceUUID, time.Now()) // 输出心跳日志 if strings.Contains(deviceUUID, "_fire_check") { - logger.Logger.Printf("♥ 推理心跳收到 | 时间: %s | UUID: %s | 类型: fire_check", time.Now().Format("2006/01/02 15:04:05.000"), deviceUUID) + logger.Logger.Printf("推理心跳收到 | 时间: %s | UUID: %s | 类型: fire_check", time.Now().Format("2006/01/02 15:04:05.000"), deviceUUID) } else { - logger.Logger.Printf("♥ 推理心跳收到 | 时间: %s | UUID: %s | 类型: fire_leave", time.Now().Format("2006/01/02 15:04:05.000"), deviceUUID) + logger.Logger.Printf("推理心跳收到 | 时间: %s | UUID: %s | 类型: fire_leave", time.Now().Format("2006/01/02 15:04:05.000"), deviceUUID) } } @@ -58,7 +57,7 @@ func StartInferenceProcess(deviceData *DeviceData) { // 重启fire_leave推理程序 StopInferenceProcess(deviceData.DeviceUUID) - portFireLeave, err := portmanager.GlobalPortManager.GetPort(deviceData.DeviceUUID) + portFireLeave, err := GlobalPortManager.GetPort(deviceData.DeviceUUID) if err != nil { logger.Logger.Printf("获取fire_leave端口失败，无法重启推理程序 (DeviceUUID=%s): %v", deviceData.DeviceUUID, err) return @@ -67,7 +66,7 @@ func StartInferenceProcess(deviceData *DeviceData) { // 重启fire_check推理程序 StopInferenceProcess(fmt.Sprintf("%s_fire_check", deviceData.DeviceUUID)) - portFireCheck, err := portmanager.GlobalPortManager.GetPort(fmt.Sprintf("%s_fire_check", deviceData.DeviceUUID)) + portFireCheck, err := GlobalPortManager.GetPort(fmt.Sprintf("%s_fire_check", deviceData.DeviceUUID)) if err != nil { logger.Logger.Printf("获取fire_check端口失败，无法重启推理程序 (DeviceUUID=%s): %v", deviceData.DeviceUUID, err) return diff --git a/connect/ws_channel.go b/connect/ws_channel.go index 45513fc..c2abb73 100644 --- a/connect/ws_channel.go +++ b/connect/ws_channel.go @@ -28,6 +28,12 @@ type PublishParams struct { Content json.RawMessage `json:"content,omitempty"` } +const ( + wsFrameRequest byte = 0x01 + wsFrameReply byte = 0x02 + wsFrameCast byte = 0x03 +) + type InferenceHeartbeat struct { DeviceUUID string `json:"device_uuid"` PersonCount int `json:"person_count"` @@ -47,8 +53,6 @@ func (c *WSClient) IsConnected() bool { return c.connected } - - // DeviceData struct type DeviceData struct { DeviceUUID string `json:"device_uuid"` @@ -71,10 +75,10 @@ type DeviceData struct { HostUUID string `json:"host_uuid"` Address string `json:"address"` // New fields for new configuration format - FireLeaveDetectArea []Area `json:"fire_leave_detect_area,omitempty"` - FireCheckDetectArea []Area `json:"fire_check_detect_area,omitempty"` - FireLeaveParam FireLeaveParam `json:"fire_leave_param,omitempty"` - FireCheckParam FireCheckParam `json:"fire_check_param,omitempty"` + FireLeaveDetectArea []Area `json:"fire_leave_detect_area,omitempty"` + FireCheckDetectArea []Area `json:"fire_check_detect_area,omitempty"` + FireLeaveParam FireLeaveParam `json:"fire_leave_param,omitempty"` + FireCheckParam FireCheckParam `json:"fire_check_param,omitempty"` } type InferenceParams struct { @@ -100,6 +104,12 @@ type WSError struct { Message string `json:"message"` } +type ServiceReply struct { + PacketID uint32 + Result []byte + Error *WSError +} + type RegisterParams struct { ContainerName string `json:"container_name,omitempty"` ServiceID string `json:"service_id,omitempty"` @@ -117,8 +127,8 @@ type WSClient struct { serverID string connected bool mutex sync.Mutex - messageID int64 - callbacks map[int64]chan *WSMessage + messageID uint32 + callbacks map[uint32]chan *ServiceReply subscribers map[string]func([]byte) closeChan chan struct{} reconnectFlag bool @@ -136,7 +146,7 @@ func NewWSClient(url string) (*WSClient, error) { url: url, connected: true, messageID: 1, - callbacks: make(map[int64]chan *WSMessage), + callbacks: make(map[uint32]chan *ServiceReply), subscribers: make(map[string]func([]byte)), closeChan: make(chan struct{}), reconnectFlag: true, @@ -243,15 +253,8 @@ func RegisterService(containerName string) (string, error) { return "", fmt.Errorf("Service registration error: %s", response.Error.Message) } - // Parse response - var resultStr string - if err := json.Unmarshal(response.Result, &resultStr); err != nil { - logger.Logger.Printf("Failed to parse service registration response: %v, Raw response: %s", err, string(response.Result)) - return "", fmt.Errorf("Failed to parse service registration response: %v", err) - } - - if resultStr != "ok" { - logger.Logger.Printf("Service registration failed: Expected result 'ok', got '%s'", resultStr) + if string(response.Result) != "ok" { + logger.Logger.Printf("Service registration failed: Expected result 'ok', got '%s'", string(response.Result)) return "", fmt.Errorf("Service registration failed: Server returned abnormal response") } @@ -286,17 +289,58 @@ func SubscribeTopic(topic string, callback func([]byte)) error { } func (c *WSClient) SendRawMessage(message interface{}) error { + messageData, err := json.Marshal(message) + if err != nil { + logger.Logger.Printf("[WebSocket Send Full JSON] marshal failed: %v", err) + } else { + logger.Logger.Printf("[WebSocket Send Full JSON] %s", string(messageData)) + } + + if metric, ok := extractMetricData(message); ok { + return c.sendMetricData(metric.routeKey, metric.metric) + } + return fmt.Errorf("unsupported websocket message type for protobuf protocol") +} + +type metricDataMessage struct { + routeKey []byte + metric []byte +} + +func extractMetricData(message interface{}) (metricDataMessage, bool) { + msg, ok := message.(map[string]interface{}) + if !ok || msg["method"] != "metric_data" { + return metricDataMessage{}, false + } + params, ok := msg["params"].(map[string]interface{}) + if !ok { + return metricDataMessage{}, false + } + routeKey, ok := params["route_key"].(string) + if !ok || routeKey == "" { + return metricDataMessage{}, false + } + metric, ok := params["metric"].(string) + if !ok { + return metricDataMessage{}, false + } + return metricDataMessage{ + routeKey: []byte(routeKey), + metric: []byte(metric), + }, true +} + +func (c *WSClient) sendBinaryFrame(frameType byte, payload []byte) error { c.mutex.Lock() defer c.mutex.Unlock() if !c.connected { return fmt.Errorf("WebSocket connection disconnected") } - messageData, err := json.Marshal(message) - if err != nil { - return fmt.Errorf("Message serialization failed: %v", err) - } - logger.Logger.Printf("Sending WebSocket message: %s", string(messageData)) - err = c.Conn.WriteMessage(websocket.TextMessage, messageData) + frame := make([]byte, 1+len(payload)) + frame[0] = frameType + copy(frame[1:], payload) + logger.Logger.Printf("Sending WebSocket binary frame: type=0x%02x, payload=%d bytes", frameType, len(payload)) + err := c.Conn.WriteMessage(websocket.BinaryMessage, frame) if err != nil { logger.Logger.Printf("WebSocket message sending failed: %v", err) c.connected = false @@ -306,27 +350,38 @@ func (c *WSClient) SendRawMessage(message interface{}) error { return nil } +func (c *WSClient) sendMetricData(routeKey, metric []byte) error { + payload := encodeServiceCastMetricData(routeKey, metric) + return c.sendBinaryFrame(wsFrameCast, payload) +} + // sendRequest sends request -func (c *WSClient) sendRequest(method string, params interface{}, waitResponse bool) (*WSMessage, error) { +func (c *WSClient) sendRequest(method string, params interface{}, waitResponse bool) (*ServiceReply, error) { c.mutex.Lock() id := c.messageID c.messageID++ c.mutex.Unlock() - message := WSMessage{ - ID: id, - Method: method, - } - if params != nil { - paramsData, err := json.Marshal(params) - if err != nil { - return nil, fmt.Errorf("Parameter serialization failed: %v", err) + var payload []byte + switch method { + case "register": + registerParams, ok := params.(RegisterParams) + if !ok { + return nil, fmt.Errorf("invalid register params") } - message.Params = paramsData + payload = encodeServiceRequestRegister(id, registerParams.ServiceID) + case "subscribe": + subscribeParams, ok := params.(SubscribeParams) + if !ok { + return nil, fmt.Errorf("invalid subscribe params") + } + payload = encodeServiceRequestSubscribe(id, subscribeParams.Topic) + default: + return nil, fmt.Errorf("unsupported request method: %s", method) } if waitResponse { - callback := make(chan *WSMessage, 1) + callback := make(chan *ServiceReply, 1) c.mutex.Lock() c.callbacks[id] = callback c.mutex.Unlock() @@ -338,7 +393,7 @@ func (c *WSClient) sendRequest(method string, params interface{}, waitResponse b logger.Logger.Printf("Cleaning up callback channel: ID=%d", id) }() - if err := c.SendRawMessage(message); err != nil { + if err := c.sendBinaryFrame(wsFrameRequest, payload); err != nil { return nil, err } @@ -355,7 +410,7 @@ func (c *WSClient) sendRequest(method string, params interface{}, waitResponse b return nil, fmt.Errorf("Connection closed") } } - return nil, c.SendRawMessage(message) + return nil, c.sendBinaryFrame(wsFrameRequest, payload) } // readLoop message reading loop @@ -370,7 +425,7 @@ func (c *WSClient) readLoop() { }() for { - _, messageData, err := c.Conn.ReadMessage() + messageType, messageData, err := c.Conn.ReadMessage() if err != nil { logger.Logger.Printf("WebSocket message reading failed: %v", err) c.mutex.Lock() @@ -380,34 +435,51 @@ func (c *WSClient) readLoop() { return } + if messageType != websocket.BinaryMessage { + logger.Logger.Printf("[WebSocket Received Raw] non-binary type=%d size=%d payload=%s", messageType, len(messageData), string(messageData)) + continue + } + if len(messageData) == 0 { + logger.Logger.Printf("Ignoring empty WebSocket binary frame") + continue + } + if string(messageData) == "pong" { logger.Logger.Printf("Received pong response") continue } - logger.Logger.Printf("Received WebSocket message: %s", string(messageData)) + frameType := messageData[0] + payload := messageData[1:] + logger.Logger.Printf("Received WebSocket binary frame: type=0x%02x, payload=%d bytes", frameType, len(payload)) - var message WSMessage - if err := json.Unmarshal(messageData, &message); err != nil { - logger.Logger.Printf("WebSocket message parsing failed: %v", err) - continue - } - - if message.ID > 0 { + switch frameType { + case wsFrameReply: + reply, err := decodeServiceReply(payload) + if err != nil { + logger.Logger.Printf("WebSocket reply parsing failed: %v", err) + continue + } + logServiceReply(reply) c.mutex.Lock() - if callback, exists := c.callbacks[message.ID]; exists { + if callback, exists := c.callbacks[reply.PacketID]; exists { select { - case callback <- &message: - logger.Logger.Printf("Successfully sent message to callback channel: ID=%d", message.ID) + case callback <- reply: + logger.Logger.Printf("Successfully sent reply to callback channel: ID=%d", reply.PacketID) default: - logger.Logger.Printf("Callback channel full, discarding message: ID=%d", message.ID) + logger.Logger.Printf("Callback channel full, discarding reply: ID=%d", reply.PacketID) } + } else { + logger.Logger.Printf("[WebSocket Received Reply] no waiting callback for packet_id=%d", reply.PacketID) } c.mutex.Unlock() - } - - if message.Method == "publish" { - c.handlePublishMessage(message) + case wsFrameCast: + if err := c.handleServiceCast(payload); err != nil { + logger.Logger.Printf("WebSocket cast handling failed: %v", err) + } + default: + logger.Logger.Printf("Ignoring unknown WebSocket frame type: 0x%02x", frameType) + continue } } } @@ -433,42 +505,49 @@ func (c *WSClient) handlePublishMessage(message WSMessage) { } if publishParams.Data != nil { - var deviceData DeviceData - if err := json.Unmarshal(publishParams.Data, &deviceData); err != nil { - logger.Logger.Printf("Failed to parse DeviceData: %v", err) - return - } - - // Check if DeviceUUID is valid - if deviceData.DeviceUUID == "" { - logger.Logger.Printf("Invalid DeviceData: DeviceUUID is empty, ignoring") - return - } - - logger.Logger.Printf("Received valid DeviceData: DeviceUUID=%s", deviceData.DeviceUUID) - - // 1. Save original configuration to service config file (using SaveSingleServiceConfig) - if err := SaveSingleServiceConfig(deviceData); err != nil { - logger.Logger.Printf("Failed to save service configuration: %v", err) - } - - // 2. Process data and save to device data file - processedData := ProcessDeviceDataForInference(deviceData) - if err := UpdateDeviceData(processedData); err != nil { - logger.Logger.Printf("Failed to update DeviceData: %v", err) - return - } - - select { - case ConfigUpdateChan <- processedData: - logger.Logger.Printf("New DeviceData sent to config update channel: DeviceUUID=%s", processedData.DeviceUUID) - default: - logger.Logger.Printf("Config update channel full, discarding message") - } - + logger.Logger.Printf("Ignoring publish.data device configuration payload; service.conf is managed by startup config only") } } +func (c *WSClient) handleServiceCast(payload []byte) error { + topic, content, ok, err := decodeServiceCastTopicEvent(payload) + if err != nil { + return err + } + if !ok { + logger.Logger.Printf("Ignoring unsupported ServiceCast body") + return nil + } + + logger.Logger.Printf("WebSocket received topic_event: Topic=%s | Content=%s", topic, string(content)) + c.mutex.Lock() + callback, exists := c.subscribers[topic] + c.mutex.Unlock() + if !exists || callback == nil { + logger.Logger.Printf("No subscriber for topic_event: Topic=%s", topic) + return nil + } + + wrapped, err := wrapTopicEvent(topic, content) + if err != nil { + return err + } + callback(wrapped) + return nil +} + +func wrapTopicEvent(topic string, content []byte) ([]byte, error) { + msg := map[string]interface{}{ + "topic": topic, + } + if json.Valid(content) { + msg["content"] = json.RawMessage(content) + } else { + msg["content"] = string(content) + } + return json.Marshal(msg) +} + // heartbeatLoop heartbeat loop - fixed version func (c *WSClient) heartbeatLoop() { logger.Logger.Printf("Starting WebSocket heartbeat loop") @@ -575,11 +654,238 @@ func (d *DeviceData) ToInferenceParams(port int) InferenceParams { } } +func appendVarint(dst []byte, v uint64) []byte { + for v >= 0x80 { + dst = append(dst, byte(v)|0x80) + v >>= 7 + } + return append(dst, byte(v)) +} +func appendBytesField(dst []byte, fieldNumber int, value []byte) []byte { + dst = appendVarint(dst, uint64(fieldNumber<<3|2)) + dst = appendVarint(dst, uint64(len(value))) + return append(dst, value...) +} +func appendVarintField(dst []byte, fieldNumber int, value uint64) []byte { + dst = appendVarint(dst, uint64(fieldNumber<<3)) + return appendVarint(dst, value) +} +func encodeServiceRequestRegister(packetID uint32, serviceID string) []byte { + register := appendBytesField(nil, 1, []byte(serviceID)) + payload := appendVarintField(nil, 1, uint64(packetID)) + payload = appendBytesField(payload, 10, register) + return payload +} +func encodeServiceRequestSubscribe(packetID uint32, topic string) []byte { + subscribe := appendBytesField(nil, 1, []byte(topic)) + payload := appendVarintField(nil, 1, uint64(packetID)) + payload = appendBytesField(payload, 11, subscribe) + return payload +} +func encodeServiceCastMetricData(routeKey, metric []byte) []byte { + metricData := appendBytesField(nil, 1, routeKey) + metricData = appendBytesField(metricData, 2, metric) + return appendBytesField(nil, 11, metricData) +} + +func consumeVarint(data []byte, pos *int) (uint64, error) { + var value uint64 + var shift uint + for *pos < len(data) { + b := data[*pos] + *pos += 1 + value |= uint64(b&0x7f) << shift + if b < 0x80 { + return value, nil + } + shift += 7 + if shift >= 64 { + return 0, fmt.Errorf("protobuf varint overflow") + } + } + return 0, fmt.Errorf("unexpected end of protobuf varint") +} + +func consumeBytes(data []byte, pos *int) ([]byte, error) { + length, err := consumeVarint(data, pos) + if err != nil { + return nil, err + } + if length > uint64(len(data)-*pos) { + return nil, fmt.Errorf("protobuf bytes field length out of range") + } + value := data[*pos : *pos+int(length)] + *pos += int(length) + return value, nil +} + +func skipField(data []byte, pos *int, wireType int) error { + switch wireType { + case 0: + _, err := consumeVarint(data, pos) + return err + case 2: + _, err := consumeBytes(data, pos) + return err + default: + return fmt.Errorf("unsupported protobuf wire type: %d", wireType) + } +} + +func decodeServiceReply(payload []byte) (*ServiceReply, error) { + reply := &ServiceReply{} + for pos := 0; pos < len(payload); { + key, err := consumeVarint(payload, &pos) + if err != nil { + return nil, err + } + fieldNumber := int(key >> 3) + wireType := int(key & 0x7) + switch fieldNumber { + case 1: + if wireType != 0 { + return nil, fmt.Errorf("invalid packet_id wire type: %d", wireType) + } + packetID, err := consumeVarint(payload, &pos) + if err != nil { + return nil, err + } + reply.PacketID = uint32(packetID) + case 10: + if wireType != 2 { + return nil, fmt.Errorf("invalid result wire type: %d", wireType) + } + result, err := consumeBytes(payload, &pos) + if err != nil { + return nil, err + } + reply.Result = result + case 11: + if wireType != 2 { + return nil, fmt.Errorf("invalid error wire type: %d", wireType) + } + errorPayload, err := consumeBytes(payload, &pos) + if err != nil { + return nil, err + } + reply.Error = decodeServiceReplyError(errorPayload) + default: + if err := skipField(payload, &pos, wireType); err != nil { + return nil, err + } + } + } + return reply, nil +} + +func logServiceReply(reply *ServiceReply) { + if reply == nil { + logger.Logger.Printf("[WebSocket Received Reply] ") + return + } + errorJSON := "null" + if reply.Error != nil { + if data, err := json.Marshal(reply.Error); err == nil { + errorJSON = string(data) + } else { + errorJSON = fmt.Sprintf(`{"marshal_error":%q}`, err.Error()) + } + } + logger.Logger.Printf("[WebSocket Received Reply] packet_id=%d result=%s error=%s", reply.PacketID, string(reply.Result), errorJSON) +} + +func decodeServiceReplyError(payload []byte) *WSError { + wsErr := &WSError{} + for pos := 0; pos < len(payload); { + key, err := consumeVarint(payload, &pos) + if err != nil { + return wsErr + } + fieldNumber := int(key >> 3) + wireType := int(key & 0x7) + switch fieldNumber { + case 1: + code, err := consumeVarint(payload, &pos) + if err != nil { + return wsErr + } + wsErr.Code = int(int32(code)) + case 2: + msg, err := consumeBytes(payload, &pos) + if err != nil { + return wsErr + } + wsErr.Message = string(msg) + default: + _ = skipField(payload, &pos, wireType) + } + } + return wsErr +} + +func decodeServiceCastTopicEvent(payload []byte) (string, []byte, bool, error) { + for pos := 0; pos < len(payload); { + key, err := consumeVarint(payload, &pos) + if err != nil { + return "", nil, false, err + } + fieldNumber := int(key >> 3) + wireType := int(key & 0x7) + if fieldNumber != 10 { + if err := skipField(payload, &pos, wireType); err != nil { + return "", nil, false, err + } + continue + } + if wireType != 2 { + return "", nil, false, fmt.Errorf("invalid topic_event wire type: %d", wireType) + } + topicPayload, err := consumeBytes(payload, &pos) + if err != nil { + return "", nil, false, err + } + topic, content, err := decodeTopicEvent(topicPayload) + return topic, content, true, err + } + return "", nil, false, nil +} + +func decodeTopicEvent(payload []byte) (string, []byte, error) { + var topic string + var content []byte + for pos := 0; pos < len(payload); { + key, err := consumeVarint(payload, &pos) + if err != nil { + return "", nil, err + } + fieldNumber := int(key >> 3) + wireType := int(key & 0x7) + switch fieldNumber { + case 1: + value, err := consumeBytes(payload, &pos) + if err != nil { + return "", nil, err + } + topic = string(value) + case 2: + value, err := consumeBytes(payload, &pos) + if err != nil { + return "", nil, err + } + content = value + default: + if err := skipField(payload, &pos, wireType); err != nil { + return "", nil, err + } + } + } + return topic, content, nil +} func (c *WSClient) SendAsync(message interface{}) { go func() { diff --git a/docs/current_change_summary.md b/docs/current_change_summary.md new file mode 100644 index 0000000..5f708f1 --- /dev/null +++ b/docs/current_change_summary.md @@ -0,0 +1,180 @@ +# Current Change Summary + +Date: 2026-05-13 + +## 1. Configuration Ownership + +The startup device configuration is owned by: + +```text +/usr/local/etc/service.conf +``` + +This file is the Docker startup configuration source. The service reads it at startup and uses it to initialize devices, inference processes, detection areas, and runtime parameters. + +WebSocket messages must not update this file. + +## 2. service.conf Schema + +`service.conf` is written as an array of `ServiceConfig` objects: + +```json +[ + { + "host_uuid": "", + "device_uuid": "", + "address": "", + "task_id": "", + "camera_rtsp": "", + "camera_ip": "", + "confidence": 70, + "detect_area": { + "fire_leave": [], + "fire_check": [] + }, + "param": { + "fire_check": { + "confidence": "70", + "detection_time": "2" + }, + "fire_leave": { + "person_count": "0", + "confidence": "70", + "detection_time": "20", + "temperature_threshold": "45", + "alarm_time": "300" + } + } + } +] +``` + +The service still tolerates the previous `DeviceData[]` cache shape when reading `service.conf`, but writes back only the canonical `ServiceConfig[]` shape. + +## 3. service.conf Write Boundary + +The remaining write functions are: + +- `connect.SaveServiceConfig()` +- `connect.SaveSingleServiceConfig()` + +They write to: + +```text +/usr/local/etc/service.conf +``` + +The old WebSocket path that wrote `publish.params.data` into `service.conf` is now disabled. If a WebSocket publish message contains `data`, the service logs and ignores it. + +## 4. Runtime Device State + +Runtime state is stored under: + +```text +/data/devices/device_{device_uuid}.json +``` + +This includes operational values such as current temperature, current person count, and alarm cooldown state. + +This is separate from `service.conf`. + +## 5. WebSocket Topic Responsibilities + +The service currently subscribes to: + +```text +/dhlr/alert +/dhlr/forbid_time +``` + +### /dhlr/alert + +This topic controls the local sound/light alarm. + +Handled by: + +```text +connect.HandleBuzzerAlert() +``` + +It reads: + +- `type` +- `buzzer_duration` + +It triggers GPIO109. It does not write `service.conf`. + +### /dhlr/forbid_time + +This topic controls the `fire_leave` reporting pause window. + +It reads: + +- `minute` + +It does not write `service.conf`. + +It writes the pause configuration to: + +```text +/data/cache/forbid_time_config.json +``` + +## 6. forbid_time Persistence + +`forbid_time_config.json` now stores: + +```json +{ + "minute": "15", + "start_time": 1770000000, + "end_time": 1770000900 +} +``` + +Meaning: + +- `minute`: configured pause duration +- `start_time`: Unix timestamp when the current pause window started +- `end_time`: Unix timestamp when the current pause window ends + +If Docker restarts while a pause window is still active, startup restores the pause window from `end_time`. + +If `end_time` is expired, no pause is restored. + +If `minute` is `0`, the pause window is cleared and `end_time` is written as `0`. + +## 7. fire_leave Pause Behavior + +During an active pause window: + +- person detection continues +- temperature reading continues +- `fire_leave` reporting is skipped +- `fire_check` reporting is not affected + +Default pause duration is: + +```text +15 minutes +``` + +## 8. Port Manager Fix + +Inference startup and heartbeat restart now use the same port manager: + +```text +connect.GlobalPortManager +``` + +This avoids restart failures caused by using a separate `portmanager.GlobalPortManager` state. + +## 9. Known Remaining Risks + +1. The protobuf WebSocket protocol migration is not implemented because this repository does not contain `service.proto` or generated Go protobuf types. + +2. The `reporter` package still contains old event-cycle functions. GPIO no longer calls them, but the old reporter upload paths still reference event-cycle behavior. This should be cleaned after confirming no external dependency remains. + +3. `fire_check` heartbeat restart currently calls a shared restart function that restarts both `fire_leave` and `fire_check`. This is functional but broader than necessary. + +4. The local environment currently does not expose `go` or `gofmt`, so compilation and formatting have not been executed in this workspace. diff --git a/fireleave_tool b/fireleave_tool index 7dddf43..f17ff71 100644 Binary files a/fireleave_tool and b/fireleave_tool differ diff --git a/gpio_monitor/gpio_monitor.go b/gpio_monitor/gpio_monitor.go index 5a998ca..d94605c 100644 --- a/gpio_monitor/gpio_monitor.go +++ b/gpio_monitor/gpio_monitor.go @@ -8,7 +8,6 @@ import ( "time" "FireLeave_tool/logger" - "FireLeave_tool/reporter" ) const ( @@ -27,6 +26,7 @@ type GPIOButtonMonitor struct { lastAlertTime time.Time cooldownTime time.Duration cooldownActive bool + onPress func() } // NewGPIOButtonMonitor 创建GPIO按钮监视器 @@ -42,6 +42,10 @@ func NewGPIOButtonMonitor(gpioPin int) *GPIOButtonMonitor { } } +func (m *GPIOButtonMonitor) SetOnPress(handler func()) { + m.onPress = handler +} + // InitGPIO 初始化GPIO func (m *GPIOButtonMonitor) InitGPIO() error { gpioPath := fmt.Sprintf("/sys/class/gpio/gpio%d", m.gpioPin) @@ -182,28 +186,14 @@ func (m *GPIOButtonMonitor) Start(ctx context.Context) error { if time.Since(m.lastAlertTime) >= m.cooldownTime { // 检测到稳定的高电平，切换事件周期 if log := logger.GlobalLoggerManager.GetLogger("gpio"); log != nil { - log.Printf("GPIO %d detected stable high level after debounce, switching event cycle...", m.gpioPin) + log.Printf("GPIO %d detected stable high level after debounce, triggering fire_leave pause...", m.gpioPin) } else if log := logger.GlobalLoggerManager.GetLogger("reporter"); log != nil { - log.Printf("GPIO %d detected stable high level after debounce, switching event cycle...", m.gpioPin) + log.Printf("GPIO %d detected stable high level after debounce, triggering fire_leave pause...", m.gpioPin) } // 切换事件周期 - var targetCycle reporter.EventCycle - switch reporter.GetCurrentEventCycle() { - case reporter.EventCycleFireLeave: - targetCycle = reporter.EventCycleFireCheck - case reporter.EventCycleFireCheck: - targetCycle = reporter.EventCycleFireLeave - default: - targetCycle = reporter.EventCycleFireCheck - } - - if err := reporter.SwitchEventCycle(targetCycle); err != nil { - if log := logger.GlobalLoggerManager.GetLogger("gpio"); log != nil { - log.Printf("Failed to switch event cycle: %v", err) - } else if log := logger.GlobalLoggerManager.GetLogger("reporter"); log != nil { - log.Printf("Failed to switch event cycle: %v", err) - } + if m.onPress != nil { + m.onPress() } // 更新lastAlertTime，开始冷却期 diff --git a/httpreader/httpreader.go b/httpreader/httpreader.go index 208a109..1d95b00 100644 --- a/httpreader/httpreader.go +++ b/httpreader/httpreader.go @@ -116,7 +116,7 @@ func (is *InferenceServer) handleVideoPost(w http.ResponseWriter, r *http.Reques return } now := time.Now().Format("2006/01/02 15:04:05.000") - logger.Logger.Printf("♥ 推理心跳收到 | 时间: %s | UUID: %s | Serial: %s | 人数量: %d", + logger.Logger.Printf("推理心跳收到 | 时间: %s | UUID: %s | Serial: %s | 人数量: %d", now, is.deviceUUID, result.Serial, getPersonCount(result)) // 这一行就是你缺失的救命稻草！ diff --git a/logger/logger.go b/logger/logger.go index e30b21c..6e2d6a0 100644 --- a/logger/logger.go +++ b/logger/logger.go @@ -9,7 +9,7 @@ import ( ) // Logger 全局日志实例 -var Logger *log.Logger +var Logger = log.New(os.Stdout, "", log.LstdFlags) var logFile *os.File var logFileName string var mu sync.Mutex @@ -239,7 +239,7 @@ func CloseLogger() { func CleanupOldLogs() { // 清理system目录的日志文件 cleanupLogDirectory("/data/logs/system") - + // 清理reporter目录的日志文件 cleanupLogDirectory("/data/logs/reporter") } diff --git a/main.go b/main.go index 007857e..000aa08 100644 --- a/main.go +++ b/main.go @@ -27,9 +27,10 @@ import ( ) type deviceAlarmState struct { - InCondition bool - StartTime int64 - AlarmPauseUntil int64 + InCondition bool + StartTime int64 + AlarmPauseUntil int64 + FireLeaveForbidUntil int64 // fire_check火焰检测状态 FireCheckInCondition bool FireCheckStartTime int64 @@ -42,6 +43,9 @@ var mergingAlarms sync.Map // 全局上报管理器 var GlobalReporter = reporter.NewReporter() +const defaultFireLeaveForbidDuration = 15 * time.Minute +const buildVersion = "2026-05-13-logger-init-fix" + // DeviceManager 设备管理器 type DeviceManager struct { mu sync.Mutex @@ -63,6 +67,71 @@ func getDeviceState(deviceUUID string) *deviceAlarmState { return s } +func applyFireLeaveForbidWindow(duration time.Duration) { + if duration <= 0 { + return + } + now := time.Now().Unix() + until := now + int64(duration.Seconds()) + applyFireLeaveForbidWindowUntil(now, until) +} + +func applyFireLeaveForbidWindowUntil(startTime, endTime int64) { + if endTime <= time.Now().Unix() { + return + } + count := 0 + GlobalDeviceManager.RangeDevices(func(uuid string, device *connect.DeviceData) bool { + state := getDeviceState(uuid) + state.FireLeaveForbidUntil = endTime + state.InCondition = false + count++ + return true + }) + logger.Logger.Printf("fire_leave reporting paused: start=%d, end=%d, devices=%d", startTime, endTime, count) +} + +func clearFireLeaveForbidWindow() { + count := 0 + GlobalDeviceManager.RangeDevices(func(uuid string, device *connect.DeviceData) bool { + state := getDeviceState(uuid) + state.FireLeaveForbidUntil = 0 + state.InCondition = false + count++ + return true + }) + logger.Logger.Printf("fire_leave reporting pause cleared, devices=%d", count) +} + +func persistAndApplyFireLeaveForbidWindow(minute string, duration time.Duration) { + if duration <= 0 { + if err := reporter.SaveForbidTimeWindow(minute, 0, 0); err != nil { + logger.Logger.Printf("Failed to clear forbid time config: %v", err) + } + clearFireLeaveForbidWindow() + return + } + startTime := time.Now().Unix() + endTime := startTime + int64(duration.Seconds()) + if err := reporter.SaveForbidTimeWindow(minute, startTime, endTime); err != nil { + logger.Logger.Printf("Failed to save forbid time config: %v", err) + } else { + logger.Logger.Printf("Forbid time config saved successfully: Minute=%s, Start=%d, End=%d", minute, startTime, endTime) + } + applyFireLeaveForbidWindowUntil(startTime, endTime) +} + +func restoreFireLeaveForbidWindow(deviceUUID string) { + config, err := reporter.LoadForbidTimeConfig() + if err != nil || config.EndTime <= time.Now().Unix() { + return + } + state := getDeviceState(deviceUUID) + state.FireLeaveForbidUntil = config.EndTime + state.InCondition = false + logger.Logger.Printf("Restored fire_leave forbid window: DeviceUUID=%s, Start=%d, End=%d", deviceUUID, config.StartTime, config.EndTime) +} + // AddDevice 添加设备 func (dm *DeviceManager) AddDevice(deviceData connect.DeviceData) { dm.mu.Lock() @@ -196,6 +265,7 @@ func processDevice(ctx context.Context, deviceData connect.DeviceData) { // 首先将设备添加到设备管理器 GlobalDeviceManager.AddDevice(deviceData) + restoreFireLeaveForbidWindow(deviceData.DeviceUUID) defer GlobalDeviceManager.StopDevice(deviceData.DeviceUUID) // 创建带取消的子上下文 @@ -588,6 +658,15 @@ func processDevice(ctx context.Context, deviceData connect.DeviceData) { conditionMet := noPerson && tempStableHigh + if now < state.FireLeaveForbidUntil { + if state.InCondition { + state.InCondition = false + } + logger.Logger.Printf("fire_leave report paused: %ds remaining (DeviceUUID=%s)", + state.FireLeaveForbidUntil-now, deviceData.DeviceUUID) + continue + } + // 1. 冷却期检查 if now < state.AlarmPauseUntil { if state.InCondition { @@ -775,6 +854,7 @@ func (dm *DeviceManager) RangeDevices(fn func(uuid string, device *connect.Devic } } func main() { + fmt.Printf("FireLeave Tool build: %s\n", buildVersion) var rtspURL string var deviceUUID string @@ -842,6 +922,11 @@ func main() { go func() { // 创建GPIO按钮监视器 monitor := gpio_monitor.NewGPIOButtonMonitor(gpio_monitor.ButtonGPIO) + monitor.SetOnPress(func() { + minute := reporter.GetForbidTimeConfig() + duration := reporter.ParseForbidDuration(minute, defaultFireLeaveForbidDuration) + persistAndApplyFireLeaveForbidWindow(minute, duration) + }) // 启动GPIO监控 if err := monitor.Start(ctx); err != nil { logger.Logger.Printf("Failed to start GPIO button monitor: %v", err) @@ -908,8 +993,7 @@ func printGlobalDeviceStatus() { i := 1 for uuid, data := range activeDevices { - stateInf, _ := deviceStates.Load(uuid) - state := stateInf.(*deviceAlarmState) + state := getDeviceState(uuid) cooldownLeft := int64(0) if now < state.AlarmPauseUntil { cooldownLeft = state.AlarmPauseUntil - now @@ -998,13 +1082,7 @@ func handleForbidTimeMessage(message []byte) { // 处理禁止时间配置 if content.Minute != "" { logger.Logger.Printf("Forbid time config received: Minute=%s", content.Minute) - - // 保存到缓存文件 - if err := reporter.SaveForbidTimeConfig(content.Minute); err != nil { - logger.Logger.Printf("Failed to save forbid time config: %v", err) - } else { - logger.Logger.Printf("Forbid time config saved successfully: Minute=%s", content.Minute) - } + persistAndApplyFireLeaveForbidWindow(content.Minute, reporter.ParseForbidDuration(content.Minute, 0)) } // 处理事件周期时限配置 diff --git a/reporter/reporter.go b/reporter/reporter.go index 828b8b1..5866b47 100644 --- a/reporter/reporter.go +++ b/reporter/reporter.go @@ -165,12 +165,23 @@ var ( ForbidTimeFile = "/data/cache/forbid_time_config.json" ) +type ForbidTimeConfig struct { + Minute string `json:"minute"` + Minutr string `json:"minutr,omitempty"` + StartTime int64 `json:"start_time"` + EndTime int64 `json:"end_time"` +} + // SaveForbidTimeConfig 保存禁止时间配置 -func SaveForbidTimeConfig(minutr string) error { - config := struct { - Minutr string `json:"minutr"` - }{ - Minutr: minutr, +func SaveForbidTimeConfig(minute string) error { + return SaveForbidTimeWindow(minute, 0, 0) +} + +func SaveForbidTimeWindow(minute string, startTime, endTime int64) error { + config := ForbidTimeConfig{ + Minute: minute, + StartTime: startTime, + EndTime: endTime, } data, err := json.Marshal(config) @@ -186,6 +197,22 @@ func SaveForbidTimeConfig(minutr string) error { return os.WriteFile(ForbidTimeFile, data, 0644) } +func LoadForbidTimeConfig() (ForbidTimeConfig, error) { + data, err := os.ReadFile(ForbidTimeFile) + if err != nil { + return ForbidTimeConfig{}, err + } + + var config ForbidTimeConfig + if err := json.Unmarshal(data, &config); err != nil { + return ForbidTimeConfig{}, err + } + if config.Minute == "" { + config.Minute = config.Minutr + } + return config, nil +} + // SaveEventCycleTimeoutConfig 保存事件周期时限配置 func SaveEventCycleTimeoutConfig(fireLeaveTimeout, fireCheckTimeout time.Duration) error { config := CycleTimeoutConfig{ @@ -208,25 +235,36 @@ func SaveEventCycleTimeoutConfig(fireLeaveTimeout, fireCheckTimeout time.Duratio // GetForbidTimeConfig 获取禁止时间配置 func GetForbidTimeConfig() string { - // 读取缓存文件 - data, err := os.ReadFile(ForbidTimeFile) + config, err := LoadForbidTimeConfig() if err != nil { // 缓存中没有值时，默认返回0 return "0" } - var config struct { - Minutr string `json:"minutr"` + if config.Minute != "" { + return config.Minute } - - if err := json.Unmarshal(data, &config); err != nil { - // 解析失败时，默认返回0 - return "0" - } - return config.Minutr } +func GetForbidDuration(defaultDuration time.Duration) time.Duration { + return ParseForbidDuration(GetForbidTimeConfig(), defaultDuration) +} + +func ParseForbidDuration(value string, defaultDuration time.Duration) time.Duration { + if value == "" || value == "0" { + return defaultDuration + } + if duration, err := time.ParseDuration(value); err == nil && duration > 0 { + return duration + } + var minutes int + if _, err := fmt.Sscanf(value, "%d", &minutes); err == nil && minutes > 0 { + return time.Duration(minutes) * time.Minute + } + return defaultDuration +} + // ShouldReportBothEvents 判断是否应该同时上报两种事件 func ShouldReportBothEvents() bool { // 注意：此逻辑已被修改，现在只在特定情况下使用 @@ -246,6 +284,29 @@ func jsonString(v interface{}) string { return string(b) } +func logJSON(logger *log.Logger, label string, v interface{}) { + if logger == nil { + return + } + b, err := json.Marshal(v) + if err != nil { + logger.Printf("%s JSON编码失败: %v", label, err) + return + } + logger.Printf("%s %s", label, string(b)) +} + +func cleanupMergedFileAfterUpload(logger *log.Logger, mergedFile, deviceUUID string, uploadFailed bool) { + if uploadFailed { + logger.Printf("Nginx上传失败，保留临时文件用于排查: %s (DeviceUUID=%s)", mergedFile, deviceUUID) + return + } + logger.Printf("清理临时文件: %s (DeviceUUID=%s)", mergedFile, deviceUUID) + if err := os.Remove(mergedFile); err != nil { + logger.Printf("清理临时文件失败: %v (DeviceUUID=%s, 文件=%s)", err, deviceUUID, mergedFile) + } +} + // UploadVideoAndReport 上传视频并上报事件 func (r *Reporter) UploadVideoAndReport(deviceData connect.DeviceData, alarmTime int64, mergedFile string) error { // 异步执行视频上传和上报，避免阻塞主线程 @@ -321,7 +382,9 @@ func (r *Reporter) UploadVideoAndReport(deviceData connect.DeviceData, alarmTime logger.Printf("开始上传视频到Nginx (DeviceUUID=%s, 文件=%s, URL=%s)", deviceData.DeviceUUID, mergedFile, deviceData.UploadURL) videoURL, err := video_server.UploadToNginx(mergedFile, nginxParams, deviceData.UploadURL) videoURLStr := "" + uploadFailed := false if err != nil { + uploadFailed = true logger.Printf("Nginx上传失败: %v (DeviceUUID=%s)", err, deviceData.DeviceUUID) } else { videoURLStr = videoURL @@ -336,14 +399,14 @@ func (r *Reporter) UploadVideoAndReport(deviceData connect.DeviceData, alarmTime r.wsClient = connect.GetWSClient() if r.wsClient == nil { logger.Printf("WebSocket客户端仍然为nil，无法上报事件 (DeviceUUID=%s)", deviceData.DeviceUUID) - os.Remove(mergedFile) + cleanupMergedFileAfterUpload(logger, mergedFile, deviceData.DeviceUUID, uploadFailed) return } } if !r.wsClient.IsConnected() { logger.Printf("WS未就绪，延迟上报 (DeviceUUID=%s)", deviceData.DeviceUUID) - os.Remove(mergedFile) + cleanupMergedFileAfterUpload(logger, mergedFile, deviceData.DeviceUUID, uploadFailed) return } logger.Printf("WebSocket客户端连接状态正常 (DeviceUUID=%s)", deviceData.DeviceUUID) @@ -365,8 +428,7 @@ func (r *Reporter) UploadVideoAndReport(deviceData connect.DeviceData, alarmTime } // 清理临时文件 - logger.Printf("清理临时文件: %s (DeviceUUID=%s)", mergedFile, deviceData.DeviceUUID) - os.Remove(mergedFile) + cleanupMergedFileAfterUpload(logger, mergedFile, deviceData.DeviceUUID, uploadFailed) logger.Printf("视频上传和事件上报完成 (DeviceUUID=%s)", deviceData.DeviceUUID) }() @@ -405,7 +467,9 @@ func (r *Reporter) UploadVideoAndReportFireLeave(deviceData connect.DeviceData, logger.Printf("开始上传视频到Nginx (DeviceUUID=%s, 文件=%s, URL=%s)", deviceData.DeviceUUID, mergedFile, deviceData.UploadURL) videoURL, err := video_server.UploadToNginx(mergedFile, nginxParams, deviceData.UploadURL) videoURLStr := "" + uploadFailed := false if err != nil { + uploadFailed = true logger.Printf("Nginx上传失败: %v (DeviceUUID=%s)", err, deviceData.DeviceUUID) } else { videoURLStr = videoURL @@ -420,14 +484,14 @@ func (r *Reporter) UploadVideoAndReportFireLeave(deviceData connect.DeviceData, r.wsClient = connect.GetWSClient() if r.wsClient == nil { logger.Printf("WebSocket客户端仍然为nil，无法上报事件 (DeviceUUID=%s)", deviceData.DeviceUUID) - os.Remove(mergedFile) + cleanupMergedFileAfterUpload(logger, mergedFile, deviceData.DeviceUUID, uploadFailed) return } } if !r.wsClient.IsConnected() { logger.Printf("WS未就绪，延迟上报 (DeviceUUID=%s)", deviceData.DeviceUUID) - os.Remove(mergedFile) + cleanupMergedFileAfterUpload(logger, mergedFile, deviceData.DeviceUUID, uploadFailed) return } logger.Printf("WebSocket客户端连接状态正常 (DeviceUUID=%s)", deviceData.DeviceUUID) @@ -449,8 +513,7 @@ func (r *Reporter) UploadVideoAndReportFireLeave(deviceData connect.DeviceData, } // 清理临时文件 - logger.Printf("清理临时文件: %s (DeviceUUID=%s)", mergedFile, deviceData.DeviceUUID) - os.Remove(mergedFile) + cleanupMergedFileAfterUpload(logger, mergedFile, deviceData.DeviceUUID, uploadFailed) logger.Printf("温度报警视频上传和事件上报完成 (DeviceUUID=%s)", deviceData.DeviceUUID) }() @@ -489,7 +552,9 @@ func (r *Reporter) UploadVideoAndReportFireCheck(deviceData connect.DeviceData, logger.Printf("开始上传视频到Nginx (DeviceUUID=%s, 文件=%s, URL=%s)", deviceData.DeviceUUID, mergedFile, deviceData.UploadURL) videoURL, err := video_server.UploadToNginx(mergedFile, nginxParams, deviceData.UploadURL) videoURLStr := "" + uploadFailed := false if err != nil { + uploadFailed = true logger.Printf("Nginx上传失败: %v (DeviceUUID=%s)", err, deviceData.DeviceUUID) } else { videoURLStr = videoURL @@ -504,14 +569,14 @@ func (r *Reporter) UploadVideoAndReportFireCheck(deviceData connect.DeviceData, r.wsClient = connect.GetWSClient() if r.wsClient == nil { logger.Printf("WebSocket客户端仍然为nil，无法上报事件 (DeviceUUID=%s)", deviceData.DeviceUUID) - os.Remove(mergedFile) + cleanupMergedFileAfterUpload(logger, mergedFile, deviceData.DeviceUUID, uploadFailed) return } } if !r.wsClient.IsConnected() { logger.Printf("WS未就绪，延迟上报 (DeviceUUID=%s)", deviceData.DeviceUUID) - os.Remove(mergedFile) + cleanupMergedFileAfterUpload(logger, mergedFile, deviceData.DeviceUUID, uploadFailed) return } logger.Printf("WebSocket客户端连接状态正常 (DeviceUUID=%s)", deviceData.DeviceUUID) @@ -533,8 +598,7 @@ func (r *Reporter) UploadVideoAndReportFireCheck(deviceData connect.DeviceData, } // 清理临时文件 - logger.Printf("清理临时文件: %s (DeviceUUID=%s)", mergedFile, deviceData.DeviceUUID) - os.Remove(mergedFile) + cleanupMergedFileAfterUpload(logger, mergedFile, deviceData.DeviceUUID, uploadFailed) logger.Printf("火焰检测报警视频上传和事件上报完成 (DeviceUUID=%s)", deviceData.DeviceUUID) }() @@ -729,6 +793,8 @@ func (r *Reporter) GenerateNginxParams(deviceData connect.DeviceData, alarmTime params["temperature"] = deviceData.Temperature } + logJSON(r.getLogger(), fmt.Sprintf("[Nginx Upload Params JSON] DeviceUUID=%s Level=%s", deviceData.DeviceUUID, level), params) + return params } diff --git a/scripts/check_logs.sh b/scripts/check_logs.sh new file mode 100644 index 0000000..751ffaf --- /dev/null +++ b/scripts/check_logs.sh @@ -0,0 +1,133 @@ +#!/bin/sh + +set -eu + +SYSTEM_DIR="${1:-/data/logs/system}" +REPORTER_DIR="${2:-/data/logs/reporter}" +LINES="${LINES:-40}" +DEVICE_UUID="${DEVICE_UUID:-}" + +print_usage() { + cat <<'USAGE' +用法: + sh scripts/check_logs.sh [系统日志目录] [上报日志目录] + +示例: + sh scripts/check_logs.sh + sh scripts/check_logs.sh /data/logs/system /data/logs/reporter + LINES=80 sh scripts/check_logs.sh + DEVICE_UUID=your-device-uuid sh scripts/check_logs.sh + +环境变量: + LINES 每个分组显示最近多少行，默认 40 + DEVICE_UUID 可选，按设备 UUID 过滤 +USAGE +} + +case "${1:-}" in + -h|--help) + print_usage + exit 0 + ;; +esac + +has_logs() { + dir="$1" + [ -d "$dir" ] || return 1 + find "$dir" -maxdepth 1 -type f -name '*.log' 2>/dev/null | grep -q . +} + +grep_logs() { + pattern="$1" + shift + + for dir in "$@"; do + has_logs "$dir" || continue + if [ -n "$DEVICE_UUID" ]; then + grep -hE "$pattern" "$dir"/*.log 2>/dev/null | grep -F "$DEVICE_UUID" || true + else + grep -hE "$pattern" "$dir"/*.log 2>/dev/null || true + fi + done +} + +count_logs() { + pattern="$1" + shift + grep_logs "$pattern" "$@" | wc -l | tr -d ' ' +} + +section() { + title="$1" + pattern="$2" + shift 2 + + count="$(count_logs "$pattern" "$@")" + printf '\n=== %s（%s 条）===\n' "$title" "$count" + if [ "$count" = "0" ]; then + echo "没有匹配日志。" + return + fi + grep_logs "$pattern" "$@" | tail -n "$LINES" +} + +summary_item() { + label="$1" + pattern="$2" + shift 2 + printf '%-34s %s\n' "$label:" "$(count_logs "$pattern" "$@")" +} + +echo "FireLeave 日志快速检查" +echo "系统日志目录 : $SYSTEM_DIR" +echo "上报日志目录 : $REPORTER_DIR" +echo "每组显示行数 : $LINES" +if [ -n "$DEVICE_UUID" ]; then + echo "设备过滤 : $DEVICE_UUID" +fi + +if ! has_logs "$SYSTEM_DIR" && ! has_logs "$REPORTER_DIR"; then + echo "指定目录下没有找到 *.log 日志文件。" + exit 1 +fi + +printf '\n=== 汇总 ===\n' +summary_item "程序启动次数" "Starting FireLeave Tool" "$SYSTEM_DIR" +summary_item "WS 注册成功" "Service registration successful" "$SYSTEM_DIR" +summary_item "Topic 订阅成功" "Topic subscription successful" "$SYSTEM_DIR" +summary_item "收到 topic_event" "WebSocket received topic_event" "$SYSTEM_DIR" +summary_item "收到声光报警命令" "Received platform alert push|Platform alert command" "$SYSTEM_DIR" +summary_item "收到 forbid_time" "Received forbid time message|Forbid time config received" "$SYSTEM_DIR" +summary_item "fire_leave 暂停命中" "fire_leave reporting paused|fire_leave report paused" "$SYSTEM_DIR" +summary_item "火焰报警触发" "Fire detected|FIRE DETECTED ALARM TRIGGERED|Fire detected continuously" "$SYSTEM_DIR" +summary_item "离人报警触发" "ALARM TRIGGERED|Alarm condition met" "$SYSTEM_DIR" +summary_item "WS/上报失败" "WebSocket.*failed|Response timeout|Asynchronous reporting failed|unsupported websocket message type|reply parsing failed|cast handling failed|WebSocket消息发送失败|响应超时|上报失败" "$SYSTEM_DIR" "$REPORTER_DIR" +summary_item "异常/错误" "panic|fatal|killed|segfault|failed|error|Error|timeout|abnormally|not connected|失败|错误|异常|超时|未连接|无法上报" "$SYSTEM_DIR" "$REPORTER_DIR" + +section "启动和设备初始化" "Starting FireLeave Tool|Initialize device data|Loaded .*device|Starting device|Device processing goroutine started|Main program startup completed|System status|初始化设备|设备数据初始化|启动设备" "$SYSTEM_DIR" + +section "推理服务启动" "Port allocation successful|Fire_leave inference server|Fire_check inference server|inference program started|Both inference programs started successfully|Start fire_leave inference program command|Start fire_check inference program command|端口分配成功|推理程序启动|推理服务" "$SYSTEM_DIR" + +section "推理心跳和重启" "推理心跳收到|Inference data processed|fire_leave.*心跳|fire_check.*心跳|heartbeat.*timeout|准备重启|StartInferenceProcess|推理程序心跳超时" "$SYSTEM_DIR" + +section "WebSocket 注册和订阅" "Starting WebSocket connection initialization|WebSocket connection initialized successfully|Starting service registration|Service registration successful|Starting topic subscription|Topic subscription successful|/dhlr/alert|/dhlr/forbid_time|WebSocket连接|服务注册|订阅" "$SYSTEM_DIR" + +section "WebSocket 二进制协议" "Sending WebSocket binary frame|Received WebSocket binary frame|Received response|Response timeout|WebSocket reply parsing failed|WebSocket cast handling failed|Ignoring non-binary|Ignoring unknown|发送WebSocket二进制帧|收到WebSocket二进制帧|响应超时" "$SYSTEM_DIR" + +section "WebSocket 重连" "WebSocket message reading failed|Failed to send ping|Reconnection attempt|WebSocket reconnection successful|Re-registration successful|Re-subscribed|Reconnection failed|read loop exited|重连|重新注册|重新订阅|读取失败" "$SYSTEM_DIR" + +section "下行 topic_event 分发" "WebSocket received topic_event|No subscriber for topic_event|Received forbid time message|Received platform alert push|Forbid time config received|Platform alert command|收到.*topic_event|没有订阅者" "$SYSTEM_DIR" + +section "声光报警 alert" "Received platform alert push|Outer parsing failed|Second parsing failed|Not a sound/light alert command|Platform alert command|Buzzer starting|GPIO109|sound/light alarm|声光报警|蜂鸣器|平台告警" "$SYSTEM_DIR" + +section "forbid_time 和 fire_leave 暂停" "Received forbid time message|Forbid time config received|fire_leave reporting paused|fire_leave report paused|fire_leave reporting pause cleared|Forbid time config saved successfully|Restored fire_leave forbid window|Invalid fire_leave_timeout|Event cycle timeout config|禁止时间|暂停|forbid_time" "$SYSTEM_DIR" + +section "fire_leave 离人报警链路" "Personnel quantity update|Temperature update|Alarm condition met|ALARM TRIGGERED|fire_leave report paused|报警事件已上报|数据快照已上报|level.*fire_leave|Level=fire_leave|人员数量更新|温度更新|离人|温度报警" "$SYSTEM_DIR" "$REPORTER_DIR" + +section "fire_check 火焰报警链路" "Received fire_check inference data|Fire detected|Fire detection started|Fire detected continuously|FIRE DETECTED ALARM TRIGGERED|火焰检测|报警事件已上报|数据快照已上报|level.*fire_check|Level=fire_check" "$SYSTEM_DIR" "$REPORTER_DIR" + +section "视频合并上传和事件上报" "开始合并视频|Merging .* videos|Video merged and moved successfully|开始上传视频|UploadToNginx|Nginx|upload failed|报警事件已上报|数据快照已上报|Asynchronous reporting successful|Asynchronous reporting failed|Sending WebSocket binary frame: type=0x03|视频上传|事件上报|数据快照" "$SYSTEM_DIR" "$REPORTER_DIR" + +section "需要重点排查的问题" "panic|fatal|killed|segfault|WebSocket.*failed|Response timeout|Asynchronous reporting failed|unsupported websocket message type|reply parsing failed|cast handling failed|not connected|无法上报|失败|错误|异常|timeout|abnormally|超时|未连接" "$SYSTEM_DIR" "$REPORTER_DIR" + +printf '\n检查完成。\n' diff --git a/video_server/video_upload.go b/video_server/video_upload.go index f4b1cc5..95c0393 100644 --- a/video_server/video_upload.go +++ b/video_server/video_upload.go @@ -17,6 +17,7 @@ import ( // UploadToNginx 上传视频到 Nginx 服务器，返回 URL func UploadToNginx(filePath string, eventParams map[string]any, uploadURL string) (string, error) { logger.Logger.Printf("开始上传视频到 Nginx: %s → %s", filePath, uploadURL) + fileName := filepath.Base(filePath) // 打开文件 logger.Logger.Printf("打开文件: %s", filePath) @@ -35,8 +36,8 @@ func UploadToNginx(filePath string, eventParams map[string]any, uploadURL string writer := multipart.NewWriter(body) // 创建表单文件字段 - logger.Logger.Printf("创建表单文件字段: %s", filepath.Base(filePath)) - part, err := writer.CreateFormFile("file", filepath.Base(filePath)) + logger.Logger.Printf("创建表单文件字段: %s", fileName) + part, err := writer.CreateFormFile("file", fileName) if err != nil { logger.Logger.Printf("创建表单文件字段失败: %v", err) return "", fmt.Errorf("create form file failed: %w", err) @@ -60,6 +61,7 @@ func UploadToNginx(filePath string, eventParams map[string]any, uploadURL string } logger.Logger.Printf("[Upload Event parameters]eventParams: %v", eventParams) logger.Logger.Printf("[Upload Event parameters]eventJSON: %s", string(eventJSON)) + logger.Logger.Printf("[Nginx Multipart Detail] URL=%s file_field=file event_field=event filename=%s event_json=%s", uploadURL, fileName, string(eventJSON)) // 写入事件字段 logger.Logger.Printf("写入事件字段到表单")