smart_streetlights/edge/camera/ipc_event.go

package main

import (
	"context"
	"runtime/debug"
	"time"

	"github.com/sirupsen/logrus"

	"lc/common/mqtt"
	"lc/common/onvif/profiles/event"
	"lc/common/onvif/soap"
	"lc/common/protocol"
)

type EventActuator struct {
	parent *LcDevice
	ctx    context.Context
	cancel context.CancelFunc
}

func NewEventActuator(parent *LcDevice) *EventActuator {
	ctx, cancel := context.WithCancel(context.Background())
	ea := EventActuator{
		parent: parent,
		ctx:    ctx,
		cancel: cancel,
	}
	return &ea
}

func (o *EventActuator) Cancel() {
	o.cancel()
}

func (o *EventActuator) EventHandle(XAddr string) {
	defer func() {
		if err := recover(); err != nil {
			logrus.Errorf("EventActuator.EventHandle发生异常:%v", err)
			logrus.Errorf("EventActuator.EventHandle发生异常,堆栈信息:%s", string(debug.Stack()))
		}
		if o.parent.EventActuator != nil {
			o.parent.EventActuator.Cancel()
			o.parent.EventActuator = nil
		}
	}()
Retry:
	//onvif事件
	client := soap.NewClient(soap.WithTimeout(time.Second * 0))
	client.AddHeader(soap.NewWSSSecurityHeader(o.parent.onvifDev.User, o.parent.onvifDev.Password))
	epy := event.NewEventPortType(client, XAddr)
	resp0, err0 := epy.GetServiceCapabilities(&event.GetServiceCapabilities{})
	if err0 != nil {
		logrus.Errorf("GetServiceCapabilities:设备[%s]获取事件能力集失败:%s", o.parent.onvifDev.Code, err0.Error())
		return
	}
	if !resp0.Capabilities.WSPullPointSupport {
		logrus.Errorf("EventsCoroutine:设备[%s]不支持事件拉点通知方式", o.parent.onvifDev.Code)
		return
	}
	resp1, err1 := epy.GetEventProperties(&event.GetEventProperties{})
	if err1 != nil {
		logrus.Errorf("GetEventProperties:设备[%s]获取事件属性失败:%s", o.parent.onvifDev.Code, err1.Error())
		return
	}

	//订阅在onvif.json中Event配置的事件主题
	var TopicExpressionDialect event.AnyURI = "https://www.onvif.org/ver10/tev/topicExpression/ConcreteSet"
	if len(resp1.TopicExpressionDialect) > 0 && len(resp1.TopicExpressionDialect[0]) > 0 {
		TopicExpressionDialect = event.AnyURI(resp1.TopicExpressionDialect[0])
	}
	cps := event.CreatePullPointSubscription{
		Filter: event.FilterType{
			TopicExpression: event.TopicExpressionType{
				TopicKinds: o.parent.onvifDev.Event,
				Dialect:    TopicExpressionDialect,
			},
		},
	}
	resp2, err2 := epy.CreatePullPointSubscription(&cps)
	if err2 != nil {
		logrus.Errorf("CreatePullPointSubscription:设备[%s]创建拉点失败:%s", o.parent.onvifDev.Code, err2.Error())
		return
	}

	pps := event.NewPullPointSubscription(client, string(resp2.SubscriptionReference.Address))
	pm := event.PullMessages{MessageLimit: resp0.Capabilities.MaxPullPoints}
	errCnt := 0
	for {
		select {
		case <-o.ctx.Done():
			logrus.Errorf("设备[%s]的EventActuator.EventHandle退出,原因:%v", o.parent.onvifDev.Code, o.ctx.Err())
			return
		default:
			resp, err := pps.PullMessages(&pm)
			if err != nil {
				if serr, ok := err.(*soap.SOAPFault); ok {
					logrus.Errorf("PullMessages:设备[%s]拉取事件失败:%s", o.parent.onvifDev.Code, serr.Error())
				} else {
					logrus.Errorf("PullMessages:设备[%s]拉取事件失败:%s", o.parent.onvifDev.Code, err.Error())
				}
				//发生错误，则停3秒再重试
				time.Sleep(3 * time.Second)
				if errCnt++; errCnt > 5 {
					logrus.Errorf("PullMessages连续5次调用发生错误，重启事件订阅服务.[code=%s]", o.parent.onvifDev.Code)
					goto Retry
				}
				break
			} else {
				errCnt = 0
			}
			if resp == nil {
				break
			}
			for _, v := range resp.NotificationMessage {
				logrus.Warnf("收到设备[%s]的告警信息:%v", o.parent.onvifDev.Code, v)
				//一键告警按键告警,v.Topic.TopicKinds == "tns1:Device/Trigger/tnshik:AlarmIn"
				t, err := time.Parse("2006-01-02T15:04:05Z", v.Message.Message.UtcTime)
				if err != nil {
					continue
				}
				ca := protocol.OnvifAlarmInfo{
					AlarmTopic: v.Topic.TopicKinds,
					Time:       protocol.ToBJTime(t).Format("2006-01-02 15:04:05"),
				}
				if len(v.Message.Message.Source.SimpleItem) > 0 {
					ca.Source = make(map[string]string)
					for _, si := range v.Message.Message.Source.SimpleItem {
						ca.Source[si.Name] = si.Value
					}
				}
				if len(v.Message.Message.Key.SimpleItem) > 0 {
					ca.Key = make(map[string]string)
					for _, si := range v.Message.Message.Key.SimpleItem {
						ca.Key[si.Name] = si.Value
					}
				}
				if len(v.Message.Message.Data.SimpleItem) > 0 {
					ca.Data = make(map[string]string)
					for _, si := range v.Message.Message.Data.SimpleItem {
						ca.Data[si.Name] = si.Value
					}
				}
				seq := GetNextUint64()
				var obj protocol.Pack_OnvifAlarm
				if str, err := obj.EnCode(o.parent.onvifDev.Code, appConfig.GID, seq, &ca); err == nil {
					topic := GetTopic(o.parent.GetDevType(), o.parent.onvifDev.Code, protocol.TP_ONVIF_ALARM)
					GetMQTTMgr().Publish(topic, str, mqtt.AtMostOnce, ToAll)
				}
				switch v.Topic.TopicKinds {
				case "tns1:Device/Trigger/tnshik:AlarmIn": //海康一键报警
					o.HandleDeviceTriggerAlarmin(seq, &ca) //录像抓图
				case "tns1:RuleEngine/LineDetector/Crossed": //遮挡告警
					o.HandleRuleengineCrossedLinedetector(seq, &ca)
				}
			}
		}
	}
}

func (o *EventActuator) HandleRuleengineCrossedLinedetector(seq uint64, ca *protocol.OnvifAlarmInfo) {
	go o.parent.Snapshot("")
}

func (o *EventActuator) HandleDeviceTriggerAlarmin(seq uint64, ca *protocol.OnvifAlarmInfo) {
	if v, ok := ca.Data["State"]; ok && v == "true" {
		go o.parent.RecordToFLV("", 20) //告警开始，抓拍视频
	}
	go o.parent.Snapshot("") //告警开始或结束，抓拍图片
}