ljydlv
/
VideoStream
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Compare commits

base: ljydlv:master
Branches Tags
ljydlv:master
ljydlv:feat-d
ljydlv:feat-dl
ljydlv:feat-sync-map
..
compare: ljydlv:feat-d
Branches Tags
ljydlv:master
ljydlv:feat-d
ljydlv:feat-dl
ljydlv:feat-sync-map

No commits in common. "master" and "feat-d" have entirely different histories.
master ... feat-d

5 changed files with 211 additions and 33 deletions
Show Stats Expand all files Collapse all files

2
go.mod
View File

@ -12,7 +12,7 @@ require (
github.com/gorilla/websocket v1.5.3
github.com/robfig/cron/v3 v3.0.1
github.com/ulule/deepcopier v0.0.0-20200430083143-45decc6639b6
gocv.io/x/gocv v0.40.0
gocv.io/x/gocv v0.39.0
gorm.io/driver/mysql v1.5.7
gorm.io/driver/postgres v1.5.11
gorm.io/gorm v1.25.12

2
go.sum
View File

@ -110,8 +110,6 @@ github.com/ulule/deepcopier v0.0.0-20200430083143-45decc6639b6 h1:TtyC78WMafNW8Q
github.com/ulule/deepcopier v0.0.0-20200430083143-45decc6639b6/go.mod h1:h8272+G2omSmi30fBXiZDMkmHuOgonplfKIKjQWzlfs=
gocv.io/x/gocv v0.39.0 h1:vWHupDE22LebZW6id2mVeT767j1YS8WqGt+ZiV7XJXE=
gocv.io/x/gocv v0.39.0/go.mod h1:zYdWMj29WAEznM3Y8NsU3A0TRq/wR/cy75jeUypThqU=
gocv.io/x/gocv v0.40.0 h1:kGBu/UVj+dO6A9dhQmGOnCICSL7ke7b5YtX3R3azdXI=
gocv.io/x/gocv v0.40.0/go.mod h1:zYdWMj29WAEznM3Y8NsU3A0TRq/wR/cy75jeUypThqU=
golang.org/x/arch v0.0.0-20210923205945-b76863e36670/go.mod h1:5om86z9Hs0C8fWVUuoMHwpExlXzs5Tkyp9hOrfG7pp8=
golang.org/x/arch v0.8.0 h1:3wRIsP3pM4yUptoR96otTUOXI367OS0+c9eeRi9doIc=
golang.org/x/arch v0.8.0/go.mod h1:FEVrYAQjsQXMVJ1nsMoVVXPZg6p2JE2mx8psSWTDQys=

82
handler/tool.go
View File

@ -8,6 +8,7 @@ import (
"github.com/gin-gonic/gin"
"github.com/google/uuid"
"github.com/gorilla/websocket"
"gocv.io/x/gocv"
"io"
"log"
"net/http"
@ -112,41 +113,57 @@ func GetVideoStream(c *gin.Context) {
//设备流
c.Stream(func(w io.Writer) bool {
var count int
frame := gocv.NewMat()
defer func() {
//关闭帧
err4 := frame.Close()
if err4 != nil {
log.Printf("device:%d frame close err:%v", deviceID, err4)
}
}()
errCount := 0
for {
if errCount > 10 {
log.Printf("stream device:%d errCount > 10", deviceID)
return false
}
buf := make([]byte, 1024*1024)
cnt := -1
cnt := service.GetDeviceCurrentFrameV2(&frame, deviceID)
if cnt == count || cnt == -1 {
time.Sleep(50 * time.Millisecond)
log.Printf("stream device:%d ,cnt =%d,count=%d,errCount=%d", deviceID, cnt, count, errCount)
errCount++
continue
}
count = cnt
if frame.Empty() {
log.Printf("stream device:%d frame is empty", deviceID)
time.Sleep(100 * time.Millisecond)
errCount++
continue
}
//gocv.Matrix转为jpeg
img, err2 := gocv.IMEncode(".jpg", frame)
if err2 != nil {
log.Printf("stream img encode err:%v", err2)
return false
}
frame_ := img.GetBytes()
_, err = w.Write([]byte("--frame\r\nContent-Type: image/jpeg\r\n\r\n"))
if err != nil {
fmt.Printf("写入头部信息错误: %v\n", err)
buf = nil
return false
}
_, err = w.Write(buf)
_, err = w.Write(frame_)
if err != nil {
fmt.Printf("写入帧数据错误: %v\n", err)
buf = nil
return false
}
_, err = w.Write([]byte("\r\n"))
if err != nil {
fmt.Printf("写入帧结束标记错误: %v\n", err)
buf = nil
return false
}
time.Sleep(50 * time.Millisecond) // 控制帧率，模拟每秒约20帧，可按实际调整
buf = nil
}
})
}
@ -176,42 +193,60 @@ func GetRealTimeImage(c *gin.Context) {
return
}
worker.SetRedisWithExpire(strconv.Itoa(int(device.ID))+"_is_play", "1", time.Minute*5)
log.Printf("device_id:%d has set is_play to 1", deviceIdInt)
isGetting = worker.GetRedis(fmt.Sprintf("device_%d_is_getting", device.ID))
if isGetting != "true" {
c.JSON(http.StatusOK, gin.H{"code": 5, "message": "device is not getting or not exist"})
log.Printf("device_id:%d is not getting or not exist", deviceIdInt)
return
}
subscribeAndHandleMessagesV6(ws, deviceIdInt)
go subscribeAndHandleMessagesV3(ws, deviceIdInt)
}
func subscribeAndHandleMessagesV6(ws *websocket.Conn, deviceId int) {
func subscribeAndHandleMessagesV3(ws *websocket.Conn, device_id int) {
// 生成唯一连接 uuid
con_id := uuid.New().String()
online_conn_key := "device_" + strconv.Itoa(deviceId) + "_online_conn_ids"
online_conn_key := "device_" + strconv.Itoa(device_id) + "_online_conn_ids"
// 加入设备在线连接集合
worker.SetRedisSetAddWithExpire(online_conn_key, con_id, time.Minute*5)
//图片计数器
count := 0
//定时器，发送计数器
tCount := 0
//计算帧率
t_count := 0
img := gocv.NewMat()
for {
//从service获取当前帧
c := service.DeviceFrameCount[deviceId]
c := service.GetDeviceCurrentFrameV2(&img, device_id)
if c != count {
count = c
buf := service.Device1CurrentFrame
err2 := ws.WriteMessage(websocket.BinaryMessage, buf)
if c == -1 {
log.Printf("device:%d get frame err!", device_id)
worker.SetRedisSetRemove(online_conn_key, con_id)
break
}
//将img转[]byte
if img.Empty() {
log.Printf("device:%d img is empty! count = %d \n", device_id, c)
} else {
//gocv.Matrix转为jpeg
buf, err := gocv.IMEncode(".jpg", img)
if err != nil {
log.Printf("img encode err:%v", err)
worker.SetRedisSetRemove(online_conn_key, con_id)
break
}
buf1 := buf.GetBytes()
err2 := ws.WriteMessage(websocket.BinaryMessage, buf1)
if err2 != nil {
log.Printf("send message to client err:%v", err2)
worker.SetRedisSetRemove(online_conn_key, con_id)
break
}
c = count
}
} else {
//每秒发送一次心跳检测
if tCount%10 == 0 {
if t_count%10 == 0 {
err := ws.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(time.Second))
if err != nil {
log.Printf("Connection check failed:%v", err)
@ -223,12 +258,17 @@ func subscribeAndHandleMessagesV6(ws *websocket.Conn, deviceId int) {
}
}
time.Sleep(100 * time.Millisecond)
tCount++
t_count++
}
// 关闭img
err := img.Close()
if err != nil {
log.Printf("device:%d img close err:%v", device_id, err)
}
// 查看是否还有其他连接，没有则设置 is_play 为 0
if worker.IsContainKey(online_conn_key) == false {
worker.SetRedisWithExpire(strconv.Itoa(deviceId)+"_is_play", "1", time.Minute*5)
log.Printf("device_id: %d has set is_play to 0", deviceId)
worker.SetRedisWithExpire(strconv.Itoa(device_id)+"_is_play", "1", time.Minute*5)
log.Printf("device_id: %d has set is_play to 0", device_id)
}
}

2
main.go
View File

@ -125,6 +125,8 @@ func ReadConfigAndSetSystem() {
}
if isGetting == "false" && device.NextStop == false { //如果设备流已经停止且不暂停，则开启
switch device.ID {
case 1:
service.Device1CurrentFrame = gocv.NewMat()
case 50:
service.Device50CurrentFrame = gocv.NewMat()
case 73:

148
service/tool.go
View File

@ -18,7 +18,7 @@ var DeviceRWMap = &sync.Map{}
var DeviceCurrentFrameMap = &sync.Map{}
var DeviceFrameCount map[int]int
var DeviceIsGettingFrame = &sync.Map{}
var Device1CurrentFrame []byte
var Device1CurrentFrame gocv.Mat
var Device50CurrentFrame gocv.Mat
var Device73CurrentFrame gocv.Mat
@ -65,7 +65,10 @@ func SetDeviceCurrentFrameV2(frame *gocv.Mat, deviceId int) error {
//设置当前帧
switch deviceId {
case 1:
Device1CurrentFrame = (*frame).ToBytes()
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())
@ -82,14 +85,149 @@ func SetDeviceCurrentFrameV2(frame *gocv.Mat, deviceId int) error {
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 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() {