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

Compare commits

base: ljydlv:69f81addebdaaa8c9aac8bb39e5757250295d3df
Branches Tags
ljydlv:master
ljydlv:feat-d
ljydlv:feat-dl
ljydlv:feat-sync-map
..
compare: ljydlv:8de01c90c2ae2db3a6b351228ad9c72eb42f5e02
Branches Tags
ljydlv:master
ljydlv:feat-d
ljydlv:feat-dl
ljydlv:feat-sync-map

6 Commits
69f81addeb ... 8de01c90c2

Author SHA1 Message Date
lijun 8de01c90c2 使用分离方式，修改视频流返回方式 2025-01-18 19:41:34 +08:00
lijun 5150e1134e 使用分离方式，修改视频流返回方式 2025-01-18 19:39:40 +08:00
lijun 940a92dea5 使用分离方式，修改视频流返回方式 2025-01-18 19:36:35 +08:00
lijun 09e3dce33a 使用分离方式，修改视频流返回方式 2025-01-18 19:32:07 +08:00
lijun 529d6b9282 使用分离方式，修改视频流返回方式 2025-01-18 19:29:47 +08:00
lijun 48d509eea7 使用分离方式，修改视频流返回方式 2025-01-18 19:22:16 +08:00
5 changed files with 98 additions and 170 deletions
Show Stats Expand all files Collapse all files

1
go.mod
View File

@ -11,7 +11,6 @@ require (
github.com/google/uuid v1.6.0
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.39.0
gorm.io/driver/mysql v1.5.7
gorm.io/driver/postgres v1.5.11

4
go.sum
View File

@ -106,10 +106,10 @@ github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS
github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08=
github.com/ugorji/go/codec v1.2.12 h1:9LC83zGrHhuUA9l16C9AHXAqEV/2wBQ4nkvumAE65EE=
github.com/ugorji/go/codec v1.2.12/go.mod h1:UNopzCgEMSXjBc6AOMqYvWC1ktqTAfzJZUZgYf6w6lg=
github.com/ulule/deepcopier v0.0.0-20200430083143-45decc6639b6 h1:TtyC78WMafNW8QFfv3TeP3yWNDG+uxNkk9vOrnDu6JA=
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=

43
handler/tool.go
View File

@ -34,7 +34,7 @@ func SetUpToolGroup(router *gin.Engine) {
}
// 跨域访问：cross origin resource share
func CrosHandler() gin.HandlerFunc {
func CorsHandler() gin.HandlerFunc {
return func(context *gin.Context) {
//method := context.Request.Method
context.Writer.Header().Set("Access-Control-Allow-Origin", "*")
@ -68,17 +68,17 @@ func GetVideoStream(c *gin.Context) {
var req videoStreamReq
id, _ := c.Get("id")
id1 := id.(int)
//将query参数绑定到结构体
if err := c.ShouldBind(&req); err != nil {
c.JSON(400, gin.H{"error": err.Error()})
return
} else {
//校验权限
device := service.GetDevice(req.ID, id1)
if device.ID == 0 {
c.JSON(400, gin.H{"error": "device not exist"})
return
}
if err := c.ShouldBind(&req); err != nil {
c.JSON(400, gin.H{"error": err.Error()})
return
} else {
//查看id是否存在
index := -1
for _, device := range proto.Config.DeviceInfo {
@ -99,14 +99,20 @@ func GetVideoStream(c *gin.Context) {
//设备流
c.Stream(func(w io.Writer) bool {
var count int
var frame gocv.Mat
for {
frame, cnt := service.GetDeviceCurrentFrame(req.ID)
cnt := service.GetDeviceCurrentFrameV2(&frame, req.ID)
if cnt == count {
time.Sleep(50 * time.Millisecond)
continue
}
//gocv.Matrix转为jpeg
img, err := gocv.IMEncode(".jpg", frame)
img, err2 := gocv.IMEncode(".jpg", frame)
if err2 != nil {
fmt.Printf("img encode err:%v", err2)
time.Sleep(50 * time.Millisecond)
continue
}
frame_ := img.GetBytes()
_, err = w.Write([]byte("--frame\r\nContent-Type: image/jpeg\r\n\r\n"))
@ -161,14 +167,15 @@ func subscribeAndHandleMessagesV3(ws *websocket.Conn, device_id int) {
//图片计数器
count := 0
//定时器，发送计数器
t_count := 0
for {
tCount := 0
//从service获取当前帧
img, c := service.GetDeviceCurrentFrameV3(device_id)
var img gocv.Mat
for {
c := service.GetDeviceCurrentFrameV2(&img, device_id)
if c != count {
//将img转[]byte
if img.Empty() {
log.Printf("device:%d img is empty! count = %d \n", device_id, c)
log.Printf("device:%d img is empty! count=%d", device_id, c)
} else {
//gocv.Matrix转为jpeg
buf, err := gocv.IMEncode(".jpg", img)
@ -193,19 +200,23 @@ func subscribeAndHandleMessagesV3(ws *websocket.Conn, device_id int) {
}
} else {
//每秒发送一次心跳检测
if t_count%10 == 0 {
if tCount%10 == 0 {
err := ws.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(time.Second))
if err != nil {
log.Printf("Connection check failed:%v", err)
worker.SetRedisSetRemove(online_conn_key, con_id)
break
} else {
log.Printf("Connection check success")
}
}
}
time.Sleep(100 * time.Millisecond)
t_count++
tCount++
}
//关闭img
err := img.Close()
if err != nil {
log.Printf("img close err:%v", err)
}
// 查看是否还有其他连接，没有则设置 is_play 为 0

18
main.go
View File

@ -10,7 +10,6 @@ import (
"github.com/gin-gonic/gin"
"github.com/golang-jwt/jwt"
"github.com/robfig/cron/v3"
"gocv.io/x/gocv"
"log"
"runtime"
"strconv"
@ -35,7 +34,7 @@ func main() {
if err0 != nil {
panic("failed to connect redis:" + err0.Error())
}
r.Use(handler.CrosHandler())
r.Use(handler.CorsHandler())
r.Use(JWTAuthMiddleware()) // 使用 JWT 认证中间件
handler.SetUpToolGroup(r) // Tool
err := worker.InitRedis()
@ -71,13 +70,14 @@ func init() {
if err != nil {
panic("failed to read config file:" + err.Error())
}
service.DeviceFrameCount = make(map[int]int)
for _, device := range proto.Config.DeviceInfo {
//service.DeviceRWMap[device.ID] = sync.RWMutex{}
//service.DeviceCurrentFrameMap[device.ID] = gocv.NewMat()
//service.DeviceIsGettingFrame[device.ID] = false
service.DeviceRWMap.Store(device.ID, &sync.RWMutex{})
service.DeviceCurrentFrameMap.Store(device.ID, gocv.NewMat())
service.DeviceFrameCount.Store(device.ID, 0)
//service.DeviceCurrentFrameMap.Store(device.ID, gocv.NewMat())
service.DeviceFrameCount[device.ID] = 0
service.DeviceIsGettingFrame.Store(device.ID, false)
}
}
@ -99,17 +99,11 @@ func ReadConfigAndSetSystem() {
if !ok {
//说明没有这个设备，需初始化添加
service.DeviceRWMap.Store(device.ID, &sync.RWMutex{})
service.DeviceCurrentFrameMap.Store(device.ID, gocv.NewMat())
service.DeviceFrameCount.Store(device.ID, 0)
//service.DeviceCurrentFrameMap.Store(device.ID, gocv.NewMat())
service.DeviceFrameCount[device.ID] = 0
service.DeviceIsGettingFrame.Store(device.ID, false)
}
if is_get == false && device.NextStop == false { //如果设备流已经停止且不暂停，则开启
switch device.ID {
case 1:
service.Device1CurrentFrame = gocv.NewMat()
case 50:
service.Device50CurrentFrame = gocv.NewMat()
}
go service.GetVideoStream(device.ID)
log.Printf("device:%d has started!\n", device.ID)
}

196
service/tool.go
View File

@ -15,45 +15,15 @@ import (
)
var DeviceRWMap = &sync.Map{}
var DeviceCurrentFrameMap = &sync.Map{}
var DeviceFrameCount = &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, device_id int) error {
//获取读写锁
mutex_, ok := DeviceRWMap.Load(device_id)
if !ok {
return fmt.Errorf("设备:%s 读写锁不存在", device_id)
}
mutex := mutex_.(*sync.RWMutex)
mutex.Lock()
defer mutex.Unlock()
//获取前一帧，将前一帧释放
framePrev, ok := DeviceCurrentFrameMap.Load(device_id)
if ok {
frame_, ok2 := framePrev.(gocv.Mat)
if ok2 {
err2 := frame_.Close()
if err2 != nil {
log.Printf("设备:%d, 错误: 无法关闭帧\n", device_id)
}
}
}
//设置当前帧
DeviceCurrentFrameMap.Store(device_id, frame)
frame_count, ok := DeviceFrameCount.Load(device_id)
if !ok {
return fmt.Errorf("设备:%s 当前帧计数不存在", device_id)
}
frame_count_ := frame_count.(int)
frame_count_++
DeviceFrameCount.Store(device_id, frame_count_)
return nil
}
func SetDeviceCurrentFrameV2(frame *gocv.Mat, device_id int) error {
func SetDeviceCurrentFrame(frame *gocv.Mat, device_id int) error {
//获取读写锁
mutex_, ok := DeviceRWMap.Load(device_id)
if !ok {
@ -65,34 +35,28 @@ func SetDeviceCurrentFrameV2(frame *gocv.Mat, device_id int) error {
//设置当前帧
switch device_id {
case 1:
//err := Device1CurrentFrame.Close()
//if err != nil {
// log.Printf("设备:%d, 错误: 无法关闭帧\n", device_id)
//}
if Device1CurrentFrame.Empty() {
Device1CurrentFrame = gocv.NewMatWithSize((*frame).Rows(), (*frame).Cols(), (*frame).Type())
}
(*frame).CopyTo(&Device1CurrentFrame)
//err := deepcopier.Copy(*frame).To(&Device1CurrentFrame)
//if err != nil {
// log.Printf("设备:%d, 错误: 无法设置帧,err: %s \n", device_id, err.Error())
//}
//查看帧状态
//log.Printf("frame:%v,Device1CurrentFrame:%v \n", (*frame).Empty(), Device1CurrentFrame.Empty())
//Device1CurrentFrame = *frame
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_count, ok := DeviceFrameCount.Load(device_id)
(*frame).CopyTo(&Device73CurrentFrame)
}
frameCount, ok := DeviceFrameCount[device_id]
if !ok {
return fmt.Errorf("设备:%s 当前帧计数不存在", device_id)
}
frame_count_ := frame_count.(int)
frame_count_++
DeviceFrameCount.Store(device_id, frame_count_)
frameCount++
DeviceFrameCount[device_id] = frameCount
return nil
}
@ -115,77 +79,33 @@ func GetDeviceCurrentFrameV2(frame *gocv.Mat, deviceId int) int {
}
mutex.RLock()
defer mutex.RUnlock()
var frame_ gocv.Mat
//获取当前帧
frameIface, ok := DeviceCurrentFrameMap.Load(deviceId)
if !ok {
return -1
}
frame_, ok = frameIface.(gocv.Mat)
if !ok {
log.Printf("DeviceCurrentFrameMap 存储的不是 gocv.Mat 类型，device_id: %d \n", deviceId)
}
*frame = frame_
frame_countIface, ok := DeviceFrameCount.Load(deviceId)
if !ok {
return -1
}
frame_count, ok := frame_countIface.(int)
if !ok {
log.Printf("DeviceFrameCount 存储的不是 int 类型，device_id: %d", deviceId)
}
return frame_count
}
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())
if frame.Empty() {
*frame = gocv.NewMatWithSize(Device1CurrentFrame.Rows(), Device1CurrentFrame.Cols(), Device1CurrentFrame.Type())
}
Device1CurrentFrame.CopyTo(frame)
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())
if frame.Empty() {
*frame = gocv.NewMatWithSize(Device50CurrentFrame.Rows(), Device50CurrentFrame.Cols(), Device50CurrentFrame.Type())
}
Device50CurrentFrame.CopyTo(frame)
case 73:
if frame.Empty() {
*frame = gocv.NewMatWithSize(Device73CurrentFrame.Rows(), Device73CurrentFrame.Cols(), Device73CurrentFrame.Type())
}
Device73CurrentFrame.CopyTo(frame)
}
frame_countIface, ok := DeviceFrameCount.Load(deviceId)
frameCount, ok := DeviceFrameCount[deviceId]
if !ok {
return gocv.NewMat(), -1
return -1
}
frame_count, ok := frame_countIface.(int)
if !ok {
log.Printf("DeviceFrameCount 存储的不是 int 类型，device_id: %d", deviceId)
}
//查看地址
//log.Printf("frame:%p,Device1CurrentFrame:%p,Device50CurrentFrame:%p\n", &frame, &Device1CurrentFrame, &Device50CurrentFrame)
return frame, frame_count
return frameCount
}
func GetDeviceCurrentFrame(deviceId int) (gocv.Mat, int) {
defer func() {
if err := recover(); err != nil {
@ -206,24 +126,34 @@ func GetDeviceCurrentFrame(deviceId int) (gocv.Mat, int) {
}
mutex.RLock()
defer mutex.RUnlock()
var ret gocv.Mat
if ret.Empty() {
}
//获取当前帧
frameIface, ok := DeviceCurrentFrameMap.Load(deviceId)
if !ok {
return gocv.NewMat(), -1
switch deviceId {
case 1:
if ret.Empty() {
ret = gocv.NewMatWithSize(Device1CurrentFrame.Rows(), Device1CurrentFrame.Cols(), Device1CurrentFrame.Type())
}
frame, ok := frameIface.(gocv.Mat)
if !ok {
log.Printf("DeviceCurrentFrameMap 存储的不是 gocv.Mat 类型，device_id: %d \n", deviceId)
Device1CurrentFrame.CopyTo(&ret)
case 50:
if ret.Empty() {
ret = gocv.NewMatWithSize(Device50CurrentFrame.Rows(), Device50CurrentFrame.Cols(), Device50CurrentFrame.Type())
}
frame_countIface, ok := DeviceFrameCount.Load(deviceId)
if !ok {
return gocv.NewMat(), -1
Device50CurrentFrame.CopyTo(&ret)
case 73:
if ret.Empty() {
ret = gocv.NewMatWithSize(Device73CurrentFrame.Rows(), Device73CurrentFrame.Cols(), Device73CurrentFrame.Type())
}
frame_count, ok := frame_countIface.(int)
if !ok {
log.Printf("DeviceFrameCount 存储的不是 int 类型，device_id: %d", deviceId)
Device73CurrentFrame.CopyTo(&ret)
}
return frame, frame_count
frameCount, ok := DeviceFrameCount[deviceId]
if !ok {
log.Printf("设备:%d 当前帧计数不存在\n", deviceId)
frameCount = -1
}
return ret, frameCount
}
func getVideoFrame(device proto.DeviceInfo) {
@ -252,6 +182,7 @@ func getVideoFrame(device proto.DeviceInfo) {
lineType := 2
z := 0
var frame gocv.Mat
frame = gocv.NewMat()
for {
if device.LogFrame > 0 && z%device.LogFrame == 0 {
log.Printf("设备:%d 当前帧: %d\n", device.ID, z)
@ -259,7 +190,6 @@ func getVideoFrame(device proto.DeviceInfo) {
if device.NextStop {
break
}
frame = gocv.NewMat()
ok := webcam.Read(&frame)
if !ok {
log.Printf("设备:%v 错误: 无法从视频流中读取帧\n", device)
@ -278,23 +208,17 @@ func getVideoFrame(device proto.DeviceInfo) {
break
}
currentTime := time.Now().Format("2006-01-02 15:04:05")
gocv.PutText(&frame, currentTime, image.Point{10, 20}, font, fontScale, fontColor, lineType)
gocv.PutText(&frame, currentTime, image.Point{X: 10, Y: 20}, font, fontScale, fontColor, lineType)
//需要将帧付给全局变量
err3 := SetDeviceCurrentFrameV2(&frame, device.ID)
err3 := SetDeviceCurrentFrame(&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())
err2 := frame.Close()
if err2 != nil {
log.Printf("设备:%d 错误: 无法释放帧资源,err:%s \n", device.ID, err.Error())
}
}