package devicebus import ( "encoding/json" "fmt" "time" mqtt "github.com/eclipse/paho.mqtt.golang" "go.uber.org/zap" "wails-app/internal/dao" "wails-app/internal/global" common "wails-app/internal/model/common" parking "wails-app/internal/service/parking" "wails-app/internal/service/uhf" ) // DeviceBus 订阅相机识别事件,并将事件转发到统一通行入口。 // 道闸状态、ACK 和 LWT 由 MQTTGateController 统一处理,避免重叠订阅覆盖回调。 type DeviceBus struct { passage *parking.PassageService } func NewDeviceBus() *DeviceBus { return &DeviceBus{passage: &parking.PassageService{}} } // CameraEvent 是相机识别事件的 MQTT 数据结构。 type CameraEvent struct { Schema string `json:"schema"` DeviceCode string `json:"device_code"` PlateNumber string `json:"plate_number"` RFIDTag string `json:"rfid_tag"` Direction string `json:"direction"` TS int64 `json:"ts"` ImagePath string `json:"image_path"` ImageID string `json:"image_id"` EventID string `json:"event_id"` Confidence float64 `json:"confidence"` } // Subscribe 只订阅相机事件,设备状态类消息由道闸控制器订阅。 func (b *DeviceBus) Subscribe(client mqtt.Client) error { topic := "parking/lot/+/booth/+/channel/+/camera/event" token := client.Subscribe(topic, 1, b.onMessage) if token.Wait() && token.Error() != nil { return token.Error() } if global.GVA_LOG != nil { global.GVA_LOG.Info("MQTT 设备主题订阅成功", zap.String("topic", topic)) } return nil } func (b *DeviceBus) onMessage(_ mqtt.Client, msg mqtt.Message) { var probe struct { Schema string `json:"schema"` } if err := json.Unmarshal(msg.Payload(), &probe); err != nil { if global.GVA_LOG != nil { global.GVA_LOG.Warn("MQTT 设备消息 JSON 无效", zap.String("topic", msg.Topic()), zap.Error(err)) } return } if global.GVA_LOG != nil { global.GVA_LOG.Info("收到设备 MQTT 消息", zap.String("topic", msg.Topic()), zap.String("schema", probe.Schema)) } if probe.Schema == "camera.event.v1" { // 识别处理包含数据库事务和等待道闸 ACK,异步执行避免阻塞 MQTT 回调。 // 回调返回后 MQTT 客户端可以复用消息缓冲区,因此协程必须使用副本。 payload := append([]byte(nil), msg.Payload()...) if global.GVA_LOG != nil { global.GVA_LOG.Info("收到车辆识别 MQTT 事件", zap.String("topic", msg.Topic()), zap.Int("payload_bytes", len(payload))) } go b.onCameraEvent(payload) } } func (b *DeviceBus) onCameraEvent(payload []byte) { startedAt := time.Now() var ev CameraEvent if err := json.Unmarshal(payload, &ev); err != nil { if global.GVA_LOG != nil { global.GVA_LOG.Warn("车辆识别 MQTT 事件解析失败", zap.Int("payload_bytes", len(payload)), zap.Error(err)) } return } if global.GVA_LOG != nil { global.GVA_LOG.Info("车辆识别事件字段已解析", zap.String("device_code", ev.DeviceCode), zap.String("event_id", ev.EventID), zap.String("image_id", ev.ImageID), zap.String("plate_number", ev.PlateNumber), zap.String("direction", ev.Direction), zap.Int64("ts", ev.TS)) } imagePath, err := resolveImagePath(ev.DeviceCode, ev.ImageID, ev.ImagePath) if err != nil { if global.GVA_LOG != nil { global.GVA_LOG.Warn("车辆识别事件引用的图片不可用", zap.String("device_code", ev.DeviceCode), zap.String("image_id", ev.ImageID), zap.String("event_id", ev.EventID), zap.Error(err)) } return } if global.GVA_LOG != nil { global.GVA_LOG.Info("车辆识别事件图片引用校验成功", zap.String("device_code", ev.DeviceCode), zap.String("event_id", ev.EventID), zap.String("image_id", ev.ImageID), zap.Bool("legacy_image_path", ev.ImageID == "")) } req := common.PassageRequest{ DeviceCode: ev.DeviceCode, PlateNumber: ev.PlateNumber, RFIDTag: ev.RFIDTag, Direction: ev.Direction, Image: imagePath, TriggerSource: "camera", } result, err := b.passage.HandlePassage(req) status := "completed" message := "识别通过" fee := 0.0 stayTime := int64(0) if err != nil { status = "failed" message = err.Error() } else if result != nil { fee = result.Fee stayTime = result.StayTime if result.Message != "" { message = result.Message } } if global.GVA_LOG != nil { global.GVA_LOG.Info("处理车辆识别事件", zap.String("device_code", ev.DeviceCode), zap.String("plate_number", ev.PlateNumber), zap.String("direction", ev.Direction), zap.String("event_id", ev.EventID), zap.String("image_id", ev.ImageID), zap.String("status", status), zap.String("message", message), zap.Duration("duration", time.Since(startedAt))) } uhf.PushChannelEvent(uhf.ChannelEvent{ DeviceCode: ev.DeviceCode, RFIDTag: ev.RFIDTag, PlateNumber: ev.PlateNumber, Direction: ev.Direction, Timestamp: ev.TS, Status: status, Message: message, Fee: fee, StayTime: stayTime, }) } // resolveImagePath accepts the new uploaded-image reference. image_path stays // supported for deployed legacy cameras until their firmware is upgraded. func resolveImagePath(deviceCode, imageID, legacyPath string) (string, error) { if imageID == "" { return legacyPath, nil } if global.GVA_DB == nil { return "", fmt.Errorf("数据库未初始化") } var image dao.DeviceImage if err := global.GVA_DB.Where("device_code = ? AND image_id = ?", deviceCode, imageID).First(&image).Error; err != nil { return "", fmt.Errorf("图片未上传或不属于该设备") } return image.URL, nil }