VideoStream/service/tool.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 GetDeviceCurrentFrameV5(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 -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()
	frameCount, ok := DeviceFrameCount[deviceId]
	if !ok {
		return -1
	}
	if frameCount == curCount {
		return frameCount
	}
	//获取当前帧
	var tempBuf []byte
	switch deviceId {
	case 1:
		if Device1CurrentFrame.Empty() {
			return -1
		}
		tempBuf = Device1CurrentFrame.ToBytes()
	case 50:
		if Device50CurrentFrame.Empty() {
			return -1
		}
		tempBuf = Device50CurrentFrame.ToBytes()
	case 73:
		if Device73CurrentFrame.Empty() {
			return -1
		}
		tempBuf = Device73CurrentFrame.ToBytes()
	default:
		return -1
	}
	// 避免直接返回大数组，可考虑复制必要的数据
	if len(tempBuf) > 0 {
		copy(*buf, tempBuf)
		// 释放临时变量
		tempBuf = nil
	}
	return frameCount
}

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)
}