443 lines
11 KiB
Go
443 lines
11 KiB
Go
package service
|
||
|
||
import (
|
||
"VideoStream/dao"
|
||
"VideoStream/proto"
|
||
"VideoStream/worker"
|
||
"fmt"
|
||
"gocv.io/x/gocv"
|
||
"image"
|
||
"image/color"
|
||
"log"
|
||
"net/http"
|
||
"sync"
|
||
"time"
|
||
)
|
||
|
||
var DeviceRWMap = &sync.Map{}
|
||
var DeviceCurrentFrameMap = &sync.Map{}
|
||
var DeviceFrameCount map[int]int
|
||
var DeviceIsGettingFrame = &sync.Map{}
|
||
var Device1CurrentFrame gocv.Mat
|
||
var Device50CurrentFrame gocv.Mat
|
||
var Device73CurrentFrame gocv.Mat
|
||
|
||
func SetDeviceCurrentFrame(frame gocv.Mat, deviceId int) error {
|
||
//获取读写锁
|
||
mutex_, ok := DeviceRWMap.Load(deviceId)
|
||
if !ok {
|
||
return fmt.Errorf("设备:%s 读写锁不存在", deviceId)
|
||
}
|
||
mutex := mutex_.(*sync.RWMutex)
|
||
mutex.Lock()
|
||
defer mutex.Unlock()
|
||
//获取前一帧，将前一帧释放
|
||
framePrev, ok := DeviceCurrentFrameMap.Load(deviceId)
|
||
if ok {
|
||
frame_, ok2 := framePrev.(gocv.Mat)
|
||
if ok2 {
|
||
err2 := frame_.Close()
|
||
if err2 != nil {
|
||
log.Printf("设备:%d, 错误: 无法关闭帧\n", deviceId)
|
||
}
|
||
}
|
||
}
|
||
//设置当前帧
|
||
DeviceCurrentFrameMap.Store(deviceId, frame)
|
||
frameCount, ok := DeviceFrameCount[deviceId]
|
||
if !ok {
|
||
return fmt.Errorf("设备:%s 当前帧计数不存在", deviceId)
|
||
}
|
||
frameCount++
|
||
DeviceFrameCount[deviceId] = frameCount
|
||
return nil
|
||
}
|
||
|
||
func SetDeviceCurrentFrameV2(frame *gocv.Mat, deviceId int) error {
|
||
//获取读写锁
|
||
mutex_, ok := DeviceRWMap.Load(deviceId)
|
||
if !ok {
|
||
return fmt.Errorf("设备:%s 读写锁不存在", deviceId)
|
||
}
|
||
mutex := mutex_.(*sync.RWMutex)
|
||
mutex.Lock()
|
||
defer mutex.Unlock()
|
||
//设置当前帧
|
||
switch deviceId {
|
||
case 1:
|
||
if Device1CurrentFrame.Empty() {
|
||
Device1CurrentFrame = gocv.NewMatWithSize((*frame).Rows(), (*frame).Cols(), (*frame).Type())
|
||
}
|
||
(*frame).CopyTo(&Device1CurrentFrame)
|
||
case 50:
|
||
if Device50CurrentFrame.Empty() {
|
||
Device50CurrentFrame = gocv.NewMatWithSize((*frame).Rows(), (*frame).Cols(), (*frame).Type())
|
||
}
|
||
(*frame).CopyTo(&Device50CurrentFrame)
|
||
case 73:
|
||
if Device73CurrentFrame.Empty() {
|
||
Device73CurrentFrame = gocv.NewMatWithSize((*frame).Rows(), (*frame).Cols(), (*frame).Type())
|
||
}
|
||
(*frame).CopyTo(&Device73CurrentFrame)
|
||
|
||
}
|
||
frameCount, ok := DeviceFrameCount[deviceId]
|
||
if !ok {
|
||
return fmt.Errorf("设备:%s 当前帧计数不存在", deviceId)
|
||
}
|
||
if frameCount%10 == 0 {
|
||
log.Printf("设备:%d 当前帧: %d 已更新\n", deviceId, frameCount)
|
||
}
|
||
frameCount++
|
||
DeviceFrameCount[deviceId] = frameCount
|
||
return nil
|
||
}
|
||
|
||
func GetDeviceCurrentFrameV4(deviceId int, curCount int) (buf []byte, cnt int) {
|
||
defer func() {
|
||
if err := recover(); err != nil {
|
||
log.Printf("设备:%d 错误: %v\n", deviceId, err)
|
||
// 异常时释放 buf
|
||
buf = nil
|
||
}
|
||
}()
|
||
//获取读写锁
|
||
mutex_, ok := DeviceRWMap.Load(deviceId)
|
||
if !ok {
|
||
log.Printf("DeviceRWMap 读写锁不存在，device_id: %d \n", deviceId)
|
||
return nil, -1
|
||
}
|
||
mutex, ok := mutex_.(*sync.RWMutex)
|
||
if !ok {
|
||
log.Printf("DeviceRWMap 存储的不是 *sync.RWMutex 类型，device_id: %d \n", deviceId)
|
||
return nil, -1
|
||
}
|
||
mutex.RLock()
|
||
defer mutex.RUnlock()
|
||
frameCount, ok := DeviceFrameCount[deviceId]
|
||
if !ok {
|
||
return buf, -1
|
||
}
|
||
if frameCount == curCount {
|
||
return nil, frameCount
|
||
}
|
||
//获取当前帧
|
||
var tempBuf []byte
|
||
switch deviceId {
|
||
case 1:
|
||
if Device1CurrentFrame.Empty() {
|
||
return nil, -1
|
||
}
|
||
tempBuf = Device1CurrentFrame.ToBytes()
|
||
case 50:
|
||
if Device50CurrentFrame.Empty() {
|
||
return nil, -1
|
||
}
|
||
tempBuf = Device50CurrentFrame.ToBytes()
|
||
case 73:
|
||
if Device73CurrentFrame.Empty() {
|
||
return nil, -1
|
||
}
|
||
tempBuf = Device73CurrentFrame.ToBytes()
|
||
default:
|
||
return nil, -1
|
||
}
|
||
// 避免直接返回大数组，可考虑复制必要的数据
|
||
if len(tempBuf) > 0 {
|
||
buf = make([]byte, len(tempBuf))
|
||
copy(buf, tempBuf)
|
||
// 释放临时变量
|
||
tempBuf = nil
|
||
}
|
||
return buf, frameCount
|
||
}
|
||
|
||
func GetDeviceCurrentFrameV2(frame *gocv.Mat, deviceId int) int {
|
||
defer func() {
|
||
if err := recover(); err != nil {
|
||
log.Printf("设备:%d 错误: %v\n", deviceId, err)
|
||
}
|
||
}()
|
||
//获取读写锁
|
||
mutex_, ok := DeviceRWMap.Load(deviceId)
|
||
if !ok {
|
||
log.Printf("DeviceRWMap 读写锁不存在，device_id: %d \n", deviceId)
|
||
return -1
|
||
}
|
||
mutex, ok := mutex_.(*sync.RWMutex)
|
||
if !ok {
|
||
log.Printf("DeviceRWMap 存储的不是 *sync.RWMutex 类型，device_id: %d \n", deviceId)
|
||
return -1
|
||
}
|
||
mutex.RLock()
|
||
defer mutex.RUnlock()
|
||
//获取当前帧
|
||
switch deviceId {
|
||
case 1:
|
||
if (*frame).Empty() {
|
||
*frame = gocv.NewMatWithSize(Device1CurrentFrame.Rows(), Device1CurrentFrame.Cols(), Device1CurrentFrame.Type())
|
||
}
|
||
if Device1CurrentFrame.Empty() {
|
||
return -1
|
||
}
|
||
Device1CurrentFrame.CopyTo(frame)
|
||
case 50:
|
||
if (*frame).Empty() {
|
||
*frame = gocv.NewMatWithSize(Device50CurrentFrame.Rows(), Device50CurrentFrame.Cols(), Device50CurrentFrame.Type())
|
||
}
|
||
if Device50CurrentFrame.Empty() {
|
||
return -1
|
||
}
|
||
Device50CurrentFrame.CopyTo(frame)
|
||
case 73:
|
||
if (*frame).Empty() {
|
||
*frame = gocv.NewMatWithSize(Device73CurrentFrame.Rows(), Device73CurrentFrame.Cols(), Device73CurrentFrame.Type())
|
||
}
|
||
if Device73CurrentFrame.Empty() {
|
||
return -1
|
||
}
|
||
Device73CurrentFrame.CopyTo(frame)
|
||
|
||
}
|
||
frameCount, ok := DeviceFrameCount[deviceId]
|
||
if !ok {
|
||
return -1
|
||
}
|
||
return frameCount
|
||
}
|
||
|
||
func GetDeviceCurrentFrameV3(deviceId int) (gocv.Mat, int) {
|
||
defer func() {
|
||
if err := recover(); err != nil {
|
||
log.Printf("设备:%d 错误: %v\n", deviceId, err)
|
||
}
|
||
}()
|
||
|
||
//获取读写锁
|
||
mutex_, ok := DeviceRWMap.Load(deviceId)
|
||
if !ok {
|
||
log.Printf("DeviceRWMap 读写锁不存在，device_id: %d \n", deviceId)
|
||
return gocv.NewMat(), -1
|
||
}
|
||
mutex, ok := mutex_.(*sync.RWMutex)
|
||
if !ok {
|
||
log.Printf("DeviceRWMap 存储的不是 *sync.RWMutex 类型，device_id: %d \n", deviceId)
|
||
return gocv.NewMat(), -1
|
||
}
|
||
mutex.RLock()
|
||
defer mutex.RUnlock()
|
||
//获取当前帧
|
||
var frame gocv.Mat
|
||
switch deviceId {
|
||
case 1:
|
||
frame = gocv.NewMatWithSize(Device1CurrentFrame.Rows(), Device1CurrentFrame.Cols(), Device1CurrentFrame.Type())
|
||
Device1CurrentFrame.CopyTo(&frame)
|
||
//查看帧状态
|
||
//log.Printf("frame:%v,Device1CurrentFrame:%v \n", frame.Empty(), Device1CurrentFrame.Empty())
|
||
|
||
case 50:
|
||
frame = gocv.NewMatWithSize(Device50CurrentFrame.Rows(), Device50CurrentFrame.Cols(), Device50CurrentFrame.Type())
|
||
Device50CurrentFrame.CopyTo(&frame)
|
||
//查看帧状态
|
||
//log.Printf("frame:%v,Device50CurrentFrame:%v\n", frame.Empty(), Device50CurrentFrame.Empty())
|
||
case 73:
|
||
frame = gocv.NewMatWithSize(Device73CurrentFrame.Rows(), Device73CurrentFrame.Cols(), Device73CurrentFrame.Type())
|
||
Device73CurrentFrame.CopyTo(&frame)
|
||
}
|
||
|
||
frameCount, ok := DeviceFrameCount[deviceId]
|
||
if !ok {
|
||
return gocv.NewMat(), -1
|
||
}
|
||
|
||
//查看地址
|
||
//log.Printf("frame:%p,Device1CurrentFrame:%p,Device50CurrentFrame:%p\n", &frame, &Device1CurrentFrame, &Device50CurrentFrame)
|
||
return frame, frameCount
|
||
}
|
||
func GetDeviceCurrentFrame(deviceId int) (gocv.Mat, int) {
|
||
defer func() {
|
||
if err := recover(); err != nil {
|
||
log.Printf("设备:%d 错误: %v\n", deviceId, err)
|
||
}
|
||
}()
|
||
|
||
//获取读写锁
|
||
mutex_, ok := DeviceRWMap.Load(deviceId)
|
||
if !ok {
|
||
log.Printf("DeviceRWMap 读写锁不存在，device_id: %d \n", deviceId)
|
||
return gocv.NewMat(), -1
|
||
}
|
||
mutex, ok := mutex_.(*sync.RWMutex)
|
||
if !ok {
|
||
log.Printf("DeviceRWMap 存储的不是 *sync.RWMutex 类型，device_id: %d \n", deviceId)
|
||
return gocv.NewMat(), -1
|
||
}
|
||
mutex.RLock()
|
||
defer mutex.RUnlock()
|
||
//获取当前帧
|
||
frameIface, ok := DeviceCurrentFrameMap.Load(deviceId)
|
||
if !ok {
|
||
return gocv.NewMat(), -1
|
||
}
|
||
frame, ok := frameIface.(gocv.Mat)
|
||
if !ok {
|
||
log.Printf("DeviceCurrentFrameMap 存储的不是 gocv.Mat 类型，device_id: %d \n", deviceId)
|
||
}
|
||
frameCount, ok := DeviceFrameCount[deviceId]
|
||
if !ok {
|
||
return gocv.NewMat(), -1
|
||
}
|
||
return frame, frameCount
|
||
}
|
||
|
||
func getVideoFrame(device proto.DeviceInfo) {
|
||
//捕获异常
|
||
defer func() {
|
||
if err := recover(); err != nil {
|
||
log.Printf("设备:%d 错误: %v\n", device.ID, err)
|
||
}
|
||
}()
|
||
|
||
webcam, err := gocv.OpenVideoCapture(device.Stream)
|
||
if err != nil {
|
||
log.Printf("设备:%d 错误: 无法打开视频流，err: %v\n", device.ID, err)
|
||
return
|
||
}
|
||
defer func(webcam *gocv.VideoCapture) {
|
||
err2 := webcam.Close()
|
||
if err2 != nil {
|
||
log.Printf("设备:%d 错误: 无法关闭视频流,%s \n", device.ID, err2.Error())
|
||
}
|
||
}(webcam)
|
||
// 字体相关设置，对应OpenCV默认字体等，这里简化处理，实际可按需求调整
|
||
font := gocv.FontHersheySimplex
|
||
fontScale := 0.5
|
||
fontColor := color.RGBA{G: 255}
|
||
lineType := 2
|
||
z := 0
|
||
var frame gocv.Mat
|
||
for {
|
||
if device.LogFrame > 0 && z%device.LogFrame == 0 {
|
||
log.Printf("设备:%d 当前帧: %d\n", device.ID, z)
|
||
}
|
||
if device.NextStop {
|
||
break
|
||
}
|
||
frame = gocv.NewMat()
|
||
ok := webcam.Read(&frame)
|
||
if !ok {
|
||
log.Printf("设备:%v 错误: 无法从视频流中读取帧\n", device)
|
||
break
|
||
}
|
||
if frame.Empty() {
|
||
log.Printf("设备:%s 错误: 无法从视频流中读取帧\n", device)
|
||
//等待50ms
|
||
time.Sleep(50 * time.Millisecond)
|
||
continue
|
||
}
|
||
height := frame.Rows()
|
||
width := frame.Cols()
|
||
if height < device.CheckFrameHeight || width < device.CheckFrameWidth {
|
||
log.Printf("设备:%s 帧尺寸已改变\n", device)
|
||
break
|
||
}
|
||
currentTime := time.Now().Format("2006-01-02 15:04:05")
|
||
gocv.PutText(&frame, currentTime, image.Point{10, 20}, font, fontScale, fontColor, lineType)
|
||
//需要将帧付给全局变量
|
||
err3 := SetDeviceCurrentFrameV2(&frame, device.ID)
|
||
if err3 != nil {
|
||
log.Printf("设备:%d 错误: 无法设置当前帧,err:%s \n", device.ID, err3.Error())
|
||
}
|
||
z++
|
||
//关闭帧
|
||
err4 := frame.Close()
|
||
if err4 != nil {
|
||
log.Printf("设备:%d 错误: 无法关闭帧,err:%s \n", device.ID, err4.Error())
|
||
}
|
||
}
|
||
//关闭帧
|
||
err4 := frame.Close()
|
||
if err4 != nil {
|
||
log.Printf("设备:%d 错误: 无法关闭帧,err:%s \n", device.ID, err4.Error())
|
||
}
|
||
}
|
||
|
||
// 发起get请求,返回响应状态码
|
||
func Get(url string) int {
|
||
req, err := http.NewRequest("GET", url, nil)
|
||
var client = &http.Client{}
|
||
if err != nil {
|
||
return 500
|
||
}
|
||
resp, err := client.Do(req)
|
||
if err != nil {
|
||
return 500
|
||
}
|
||
return resp.StatusCode
|
||
}
|
||
|
||
func GetVideoStream(id int) {
|
||
defer func() {
|
||
if err := recover(); err != nil {
|
||
log.Printf("设备:%d 错误: %v\n", id, err)
|
||
}
|
||
}()
|
||
isGetting := worker.GetRedis(fmt.Sprintf("device_%d_is_getting", id))
|
||
if isGetting == "true" || isGetting == "" {
|
||
log.Printf("设备:%d 正在运行,isGetting:%s", id, isGetting)
|
||
return
|
||
}
|
||
for {
|
||
var device proto.DeviceInfo
|
||
var index int
|
||
//获取设备信息
|
||
for i, device1 := range proto.Config.DeviceInfo {
|
||
if device1.ID == id {
|
||
device = device1
|
||
break
|
||
}
|
||
index = i
|
||
}
|
||
if index == len(proto.Config.DeviceInfo) {
|
||
//设备不存在
|
||
log.Printf("device: %d not found", id)
|
||
break
|
||
}
|
||
isGetting = worker.GetRedis(fmt.Sprintf("device_%d_is_getting", id))
|
||
if isGetting == "true" {
|
||
log.Printf("设备:%d 正在运行,isGetting:%s", id, isGetting)
|
||
break
|
||
}
|
||
|
||
if device.NextStop {
|
||
DeviceIsGettingFrame.Store(id, false)
|
||
break
|
||
}
|
||
//设置设备控制信息
|
||
status := Get(device.Control)
|
||
DeviceIsGettingFrame.Store(id, true)
|
||
worker.SetRedis(fmt.Sprintf("device_%d_is_getting", id), "true")
|
||
log.Printf("device: %d set control info status: %d", device.ID, status)
|
||
getVideoFrame(device)
|
||
DeviceIsGettingFrame.Store(id, false)
|
||
worker.SetRedis(fmt.Sprintf("device_%d_is_getting", id), "false")
|
||
//等待1s
|
||
time.Sleep(1 * time.Second)
|
||
}
|
||
|
||
}
|
||
|
||
func DoGetVideoStream() {
|
||
for _, device := range proto.Config.DeviceInfo {
|
||
go GetVideoStream(device.ID)
|
||
}
|
||
}
|
||
|
||
func GetDevice(id, auth_id int) dao.Device {
|
||
return dao.FindDeviceByID(id, auth_id)
|
||
}
|
||
|
||
func SetConfigRedis(key, config string) {
|
||
worker.SetRedis(key, config)
|
||
}
|