Compare commits
No commits in common. "master" and "feat-sync-map" have entirely different histories.
master
...
feat-sync-
3
go.mod
3
go.mod
|
|
@ -11,8 +11,7 @@ require (
|
||||||
github.com/google/uuid v1.6.0
|
github.com/google/uuid v1.6.0
|
||||||
github.com/gorilla/websocket v1.5.3
|
github.com/gorilla/websocket v1.5.3
|
||||||
github.com/robfig/cron/v3 v3.0.1
|
github.com/robfig/cron/v3 v3.0.1
|
||||||
github.com/ulule/deepcopier v0.0.0-20200430083143-45decc6639b6
|
gocv.io/x/gocv v0.39.0
|
||||||
gocv.io/x/gocv v0.40.0
|
|
||||||
gorm.io/driver/mysql v1.5.7
|
gorm.io/driver/mysql v1.5.7
|
||||||
gorm.io/driver/postgres v1.5.11
|
gorm.io/driver/postgres v1.5.11
|
||||||
gorm.io/gorm v1.25.12
|
gorm.io/gorm v1.25.12
|
||||||
|
|
|
||||||
2
go.sum
2
go.sum
|
|
@ -106,8 +106,6 @@ 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/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 h1:9LC83zGrHhuUA9l16C9AHXAqEV/2wBQ4nkvumAE65EE=
|
||||||
github.com/ugorji/go/codec v1.2.12/go.mod h1:UNopzCgEMSXjBc6AOMqYvWC1ktqTAfzJZUZgYf6w6lg=
|
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 h1:vWHupDE22LebZW6id2mVeT767j1YS8WqGt+ZiV7XJXE=
|
||||||
gocv.io/x/gocv v0.39.0/go.mod h1:zYdWMj29WAEznM3Y8NsU3A0TRq/wR/cy75jeUypThqU=
|
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 h1:kGBu/UVj+dO6A9dhQmGOnCICSL7ke7b5YtX3R3azdXI=
|
||||||
|
|
|
||||||
134
handler/tool.go
134
handler/tool.go
|
|
@ -8,6 +8,7 @@ import (
|
||||||
"github.com/gin-gonic/gin"
|
"github.com/gin-gonic/gin"
|
||||||
"github.com/google/uuid"
|
"github.com/google/uuid"
|
||||||
"github.com/gorilla/websocket"
|
"github.com/gorilla/websocket"
|
||||||
|
"gocv.io/x/gocv"
|
||||||
"io"
|
"io"
|
||||||
"log"
|
"log"
|
||||||
"net/http"
|
"net/http"
|
||||||
|
|
@ -33,7 +34,7 @@ func SetUpToolGroup(router *gin.Engine) {
|
||||||
}
|
}
|
||||||
|
|
||||||
// 跨域访问:cross origin resource share
|
// 跨域访问:cross origin resource share
|
||||||
func CorsHandler() gin.HandlerFunc {
|
func CrosHandler() gin.HandlerFunc {
|
||||||
return func(context *gin.Context) {
|
return func(context *gin.Context) {
|
||||||
//method := context.Request.Method
|
//method := context.Request.Method
|
||||||
context.Writer.Header().Set("Access-Control-Allow-Origin", "*")
|
context.Writer.Header().Set("Access-Control-Allow-Origin", "*")
|
||||||
|
|
@ -64,171 +65,152 @@ type videoStreamReq struct {
|
||||||
}
|
}
|
||||||
|
|
||||||
func GetVideoStream(c *gin.Context) {
|
func GetVideoStream(c *gin.Context) {
|
||||||
|
var req videoStreamReq
|
||||||
id, _ := c.Get("id")
|
id, _ := c.Get("id")
|
||||||
id1 := id.(int)
|
id1 := id.(int)
|
||||||
deviceIDSTR := c.Query("id")
|
|
||||||
deviceID, err := strconv.Atoi(deviceIDSTR)
|
|
||||||
if err != nil {
|
|
||||||
c.JSON(400, gin.H{"error": "device_id error"})
|
|
||||||
return
|
|
||||||
}
|
|
||||||
key := c.Query("key")
|
|
||||||
//校验权限
|
//校验权限
|
||||||
device := service.GetDevice(deviceID, id1)
|
device := service.GetDevice(req.ID, id1)
|
||||||
if device.ID == 0 {
|
if device.ID == 0 {
|
||||||
c.JSON(400, gin.H{"error": "device not exist"})
|
c.JSON(400, gin.H{"error": "device not exist"})
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
rKey := worker.GetRedis("video_stream_get_stream_key")
|
|
||||||
if rKey == "" {
|
|
||||||
rKey = "123456"
|
|
||||||
}
|
|
||||||
|
|
||||||
|
if err := c.ShouldBind(&req); err != nil {
|
||||||
|
c.JSON(400, gin.H{"error": err.Error()})
|
||||||
|
return
|
||||||
|
} else {
|
||||||
//查看id是否存在
|
//查看id是否存在
|
||||||
index := -1
|
index := -1
|
||||||
for _, device_ := range proto.Config.DeviceInfo {
|
for _, device := range proto.Config.DeviceInfo {
|
||||||
if device_.ID == deviceID {
|
if device.ID == req.ID {
|
||||||
index = deviceID
|
index = req.ID
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if index == -1 {
|
if index == -1 {
|
||||||
c.JSON(400, gin.H{"error": "id config not exist"})
|
c.JSON(400, gin.H{"error": "id not exist"})
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
//查看key是否正确
|
//查看key是否正确
|
||||||
if key != rKey {
|
if req.Key != "123456" {
|
||||||
c.JSON(400, gin.H{"error": "key error"})
|
c.JSON(400, gin.H{"error": "key error"})
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
//查看设备是否在获取
|
|
||||||
isGetting := worker.GetRedis(fmt.Sprintf("device_%d_is_getting", deviceID))
|
|
||||||
if isGetting != "true" {
|
|
||||||
c.JSON(400, gin.H{"error": "device is not getting or not exist"})
|
|
||||||
log.Printf("stream device_id:%d is not getting or not exist", deviceID)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
//设备流
|
//设备流
|
||||||
c.Stream(func(w io.Writer) bool {
|
c.Stream(func(w io.Writer) bool {
|
||||||
var count int
|
var count int
|
||||||
errCount := 0
|
|
||||||
for {
|
for {
|
||||||
if errCount > 10 {
|
frame, cnt := service.GetDeviceCurrentFrame(req.ID)
|
||||||
log.Printf("stream device:%d errCount > 10", deviceID)
|
if cnt == count {
|
||||||
return false
|
|
||||||
}
|
|
||||||
buf := make([]byte, 1024*1024)
|
|
||||||
cnt := -1
|
|
||||||
if cnt == count || cnt == -1 {
|
|
||||||
time.Sleep(50 * time.Millisecond)
|
time.Sleep(50 * time.Millisecond)
|
||||||
log.Printf("stream device:%d,cnt =%d,count=%d,errCount=%d", deviceID, cnt, count, errCount)
|
|
||||||
errCount++
|
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
count = cnt
|
//gocv.Matrix转为jpeg
|
||||||
|
img, err := gocv.IMEncode(".jpg", frame)
|
||||||
|
frame_ := img.GetBytes()
|
||||||
|
|
||||||
_, err = w.Write([]byte("--frame\r\nContent-Type: image/jpeg\r\n\r\n"))
|
_, err = w.Write([]byte("--frame\r\nContent-Type: image/jpeg\r\n\r\n"))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
fmt.Printf("写入头部信息错误: %v\n", err)
|
fmt.Printf("写入头部信息错误: %v\n", err)
|
||||||
buf = nil
|
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
_, err = w.Write(buf)
|
_, err = w.Write(frame_)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
fmt.Printf("写入帧数据错误: %v\n", err)
|
fmt.Printf("写入帧数据错误: %v\n", err)
|
||||||
buf = nil
|
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
_, err = w.Write([]byte("\r\n"))
|
_, err = w.Write([]byte("\r\n"))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
fmt.Printf("写入帧结束标记错误: %v\n", err)
|
fmt.Printf("写入帧结束标记错误: %v\n", err)
|
||||||
buf = nil
|
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
time.Sleep(50 * time.Millisecond) // 控制帧率,模拟每秒约20帧,可按实际调整
|
time.Sleep(50 * time.Millisecond) // 控制帧率,模拟每秒约20帧,可按实际调整
|
||||||
buf = nil
|
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
|
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// 发送实时视频流
|
// 发送实时视频流
|
||||||
func GetRealTimeImage(c *gin.Context) {
|
func GetRealTimeImage(c *gin.Context) {
|
||||||
id, _ := c.Get("id")
|
id, _ := c.Get("id")
|
||||||
id1 := id.(int)
|
id1 := id.(int)
|
||||||
deviceId := c.Query("device_id")
|
device_id := c.Query("device_id")
|
||||||
deviceIdInt, _ := strconv.Atoi(deviceId)
|
device_id_int, _ := strconv.Atoi(device_id)
|
||||||
device := service.GetDevice(deviceIdInt, id1)
|
device := service.GetDevice(device_id_int, id1)
|
||||||
if device.ID == 0 {
|
if device.ID == 0 {
|
||||||
c.JSON(http.StatusOK, gin.H{"code": 4, "message": "device not found"})
|
c.JSON(http.StatusOK, gin.H{"code": 4, "message": "device not found"})
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
//查看设备是否在获取
|
|
||||||
isGetting := worker.GetRedis(fmt.Sprintf("device_%d_is_getting", device.ID))
|
|
||||||
if isGetting != "true" {
|
|
||||||
c.JSON(http.StatusOK, gin.H{"code": 4, "message": "device is not getting or not exist"})
|
|
||||||
log.Printf("device_id:%d is not getting or not exist", deviceIdInt)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
ws, err := upgrader.Upgrade(c.Writer, c.Request, nil)
|
ws, err := upgrader.Upgrade(c.Writer, c.Request, nil)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Printf("connect wss err:%v", err)
|
log.Printf("connect wss err:%v", err)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
worker.SetRedisWithExpire(strconv.Itoa(int(device.ID))+"_is_play", "1", time.Minute*5)
|
worker.SetRedisWithExpire(strconv.Itoa(int(device.ID))+"_is_play", "1", time.Minute*5)
|
||||||
isGetting = worker.GetRedis(fmt.Sprintf("device_%d_is_getting", device.ID))
|
log.Printf("device_id:%d has set is_play to 1", device_id_int)
|
||||||
if isGetting != "true" {
|
go subscribeAndHandleMessagesV3(ws, device_id_int)
|
||||||
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)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func subscribeAndHandleMessagesV6(ws *websocket.Conn, deviceId int) {
|
func subscribeAndHandleMessagesV3(ws *websocket.Conn, device_id int) {
|
||||||
// 生成唯一连接 uuid
|
// 生成唯一连接 uuid
|
||||||
con_id := uuid.New().String()
|
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)
|
worker.SetRedisSetAddWithExpire(online_conn_key, con_id, time.Minute*5)
|
||||||
//图片计数器
|
//图片计数器
|
||||||
count := 0
|
count := 0
|
||||||
//定时器,发送计数器
|
//定时器,发送计数器
|
||||||
tCount := 0
|
t_count := 0
|
||||||
//计算帧率
|
|
||||||
for {
|
for {
|
||||||
//从service获取当前帧
|
//从service获取当前帧
|
||||||
c := service.DeviceFrameCount[deviceId]
|
var img gocv.Mat
|
||||||
|
c := service.GetDeviceCurrentFrameV2(&img, device_id)
|
||||||
if c != count {
|
if c != count {
|
||||||
count = c
|
//将img转[]byte
|
||||||
buf := service.Device1CurrentFrame
|
if img.Empty() {
|
||||||
err2 := ws.WriteMessage(websocket.BinaryMessage, buf)
|
log.Printf("device:%d img is empty!", device_id)
|
||||||
|
} 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)
|
||||||
|
goto end
|
||||||
|
}
|
||||||
|
buf1 := buf.GetBytes()
|
||||||
|
|
||||||
|
err2 := ws.WriteMessage(websocket.BinaryMessage, buf1)
|
||||||
if err2 != nil {
|
if err2 != nil {
|
||||||
log.Printf("send message to client err:%v", err2)
|
log.Printf("send message to client err:%v", err2)
|
||||||
worker.SetRedisSetRemove(online_conn_key, con_id)
|
worker.SetRedisSetRemove(online_conn_key, con_id)
|
||||||
break
|
goto end
|
||||||
}
|
}
|
||||||
c = count
|
c = count
|
||||||
|
err4 := img.Close()
|
||||||
|
if err4 != nil {
|
||||||
|
log.Printf("close img err:%v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
} else {
|
} else {
|
||||||
//每秒发送一次心跳检测
|
//每秒发送一次心跳检测
|
||||||
if tCount%10 == 0 {
|
if t_count%10 == 0 {
|
||||||
err := ws.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(time.Second))
|
err := ws.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(time.Second))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Printf("Connection check failed:%v", err)
|
log.Printf("Connection check failed:%v", err)
|
||||||
worker.SetRedisSetRemove(online_conn_key, con_id)
|
worker.SetRedisSetRemove(online_conn_key, con_id)
|
||||||
break
|
goto end
|
||||||
} else {
|
|
||||||
log.Printf("Connection check success")
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
time.Sleep(100 * time.Millisecond)
|
time.Sleep(100 * time.Millisecond)
|
||||||
tCount++
|
t_count++
|
||||||
}
|
}
|
||||||
|
|
||||||
|
end:
|
||||||
// 查看是否还有其他连接,没有则设置 is_play 为 0
|
// 查看是否还有其他连接,没有则设置 is_play 为 0
|
||||||
if worker.IsContainKey(online_conn_key) == false {
|
if worker.IsContainKey(online_conn_key) == false {
|
||||||
worker.SetRedisWithExpire(strconv.Itoa(deviceId)+"_is_play", "1", time.Minute*5)
|
worker.SetRedisWithExpire(strconv.Itoa(device_id)+"_is_play", "1", time.Minute*5)
|
||||||
log.Printf("device_id: %d has set is_play to 0", deviceId)
|
log.Printf("device_id: %d has set is_play to 0", device_id)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
47
main.go
47
main.go
|
|
@ -19,11 +19,10 @@ import (
|
||||||
)
|
)
|
||||||
|
|
||||||
var configPath string
|
var configPath string
|
||||||
var logCount int
|
|
||||||
|
|
||||||
func main() {
|
func main() {
|
||||||
fmt.Println("start server")
|
fmt.Println("start server")
|
||||||
gin.SetMode(gin.ReleaseMode) //设置为debug模式
|
gin.SetMode(gin.DebugMode) //设置为debug模式
|
||||||
r := gin.Default()
|
r := gin.Default()
|
||||||
//数据库初始
|
//数据库初始
|
||||||
err0 := dao.Init()
|
err0 := dao.Init()
|
||||||
|
|
@ -36,7 +35,7 @@ func main() {
|
||||||
if err0 != nil {
|
if err0 != nil {
|
||||||
panic("failed to connect redis:" + err0.Error())
|
panic("failed to connect redis:" + err0.Error())
|
||||||
}
|
}
|
||||||
r.Use(handler.CorsHandler())
|
r.Use(handler.CrosHandler())
|
||||||
r.Use(JWTAuthMiddleware()) // 使用 JWT 认证中间件
|
r.Use(JWTAuthMiddleware()) // 使用 JWT 认证中间件
|
||||||
handler.SetUpToolGroup(r) // Tool
|
handler.SetUpToolGroup(r) // Tool
|
||||||
err := worker.InitRedis()
|
err := worker.InitRedis()
|
||||||
|
|
@ -52,7 +51,6 @@ func main() {
|
||||||
}
|
}
|
||||||
c.Start()
|
c.Start()
|
||||||
fmt.Println("定时任务已启动")
|
fmt.Println("定时任务已启动")
|
||||||
initDeviceGettingStatus()
|
|
||||||
err3 := r.Run(":" + proto.Config.SERVER_PORT)
|
err3 := r.Run(":" + proto.Config.SERVER_PORT)
|
||||||
if err3 != nil {
|
if err3 != nil {
|
||||||
panic("failed to run server:" + err3.Error())
|
panic("failed to run server:" + err3.Error())
|
||||||
|
|
@ -68,20 +66,18 @@ func init() {
|
||||||
} else {
|
} else {
|
||||||
configPath = "/home/videoplayer/vp_stream.conf"
|
configPath = "/home/videoplayer/vp_stream.conf"
|
||||||
}
|
}
|
||||||
logCount = 0
|
|
||||||
//读取配置文件
|
//读取配置文件
|
||||||
err := proto.ReadConfig(configPath)
|
err := proto.ReadConfig(configPath)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
panic("failed to read config file:" + err.Error())
|
panic("failed to read config file:" + err.Error())
|
||||||
}
|
}
|
||||||
service.DeviceFrameCount = make(map[int]int)
|
|
||||||
for _, device := range proto.Config.DeviceInfo {
|
for _, device := range proto.Config.DeviceInfo {
|
||||||
//service.DeviceRWMap[device.ID] = sync.RWMutex{}
|
//service.DeviceRWMap[device.ID] = sync.RWMutex{}
|
||||||
//service.DeviceCurrentFrameMap[device.ID] = gocv.NewMat()
|
//service.DeviceCurrentFrameMap[device.ID] = gocv.NewMat()
|
||||||
//service.DeviceIsGettingFrame[device.ID] = false
|
//service.DeviceIsGettingFrame[device.ID] = false
|
||||||
service.DeviceRWMap.Store(device.ID, &sync.RWMutex{})
|
service.DeviceRWMap.Store(device.ID, &sync.RWMutex{})
|
||||||
service.DeviceCurrentFrameMap.Store(device.ID, gocv.NewMat())
|
service.DeviceCurrentFrameMap.Store(device.ID, gocv.NewMat())
|
||||||
service.DeviceFrameCount[device.ID] = 0
|
service.DeviceFrameCount.Store(device.ID, 0)
|
||||||
service.DeviceIsGettingFrame.Store(device.ID, false)
|
service.DeviceIsGettingFrame.Store(device.ID, false)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -90,20 +86,7 @@ func myTask() {
|
||||||
ReadConfigAndSetSystem()
|
ReadConfigAndSetSystem()
|
||||||
}
|
}
|
||||||
|
|
||||||
func initDeviceGettingStatus() {
|
|
||||||
for _, device := range proto.Config.DeviceInfo {
|
|
||||||
if device.NextStop == false {
|
|
||||||
worker.SetRedis(fmt.Sprintf("device_%d_is_getting", device.ID), "false")
|
|
||||||
}
|
|
||||||
if device.NextStop == true {
|
|
||||||
worker.DelRedis(fmt.Sprintf("device_%d_is_getting", device.ID))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
}
|
|
||||||
|
|
||||||
func ReadConfigAndSetSystem() {
|
func ReadConfigAndSetSystem() {
|
||||||
logCount++
|
|
||||||
//configPath := "/home/videoplayer/vp_stream.conf"
|
//configPath := "/home/videoplayer/vp_stream.conf"
|
||||||
//读取配置文件
|
//读取配置文件
|
||||||
err := proto.ReadConfig(configPath)
|
err := proto.ReadConfig(configPath)
|
||||||
|
|
@ -112,31 +95,20 @@ func ReadConfigAndSetSystem() {
|
||||||
}
|
}
|
||||||
//检测是否需要获取设备流,如果需要则开启
|
//检测是否需要获取设备流,如果需要则开启
|
||||||
for _, device := range proto.Config.DeviceInfo {
|
for _, device := range proto.Config.DeviceInfo {
|
||||||
//isGet, ok := service.DeviceIsGettingFrame.Load(device.ID)
|
is_get, ok := service.DeviceIsGettingFrame.Load(device.ID)
|
||||||
//isGet_ := isGet.(bool)
|
if !ok {
|
||||||
isGetting := worker.GetRedis(fmt.Sprintf("device_%d_is_getting", device.ID))
|
|
||||||
if isGetting == "" {
|
|
||||||
//说明没有这个设备,需初始化添加
|
//说明没有这个设备,需初始化添加
|
||||||
service.DeviceRWMap.Store(device.ID, &sync.RWMutex{})
|
service.DeviceRWMap.Store(device.ID, &sync.RWMutex{})
|
||||||
service.DeviceCurrentFrameMap.Store(device.ID, gocv.NewMat())
|
service.DeviceCurrentFrameMap.Store(device.ID, gocv.NewMat())
|
||||||
service.DeviceFrameCount[device.ID] = 0
|
service.DeviceFrameCount.Store(device.ID, 0)
|
||||||
service.DeviceIsGettingFrame.Store(device.ID, false)
|
service.DeviceIsGettingFrame.Store(device.ID, false)
|
||||||
worker.SetRedis(fmt.Sprintf("device_%d_is_getting", device.ID), "false")
|
|
||||||
}
|
|
||||||
if isGetting == "false" && device.NextStop == false { //如果设备流已经停止且不暂停,则开启
|
|
||||||
switch device.ID {
|
|
||||||
case 50:
|
|
||||||
service.Device50CurrentFrame = gocv.NewMat()
|
|
||||||
case 73:
|
|
||||||
service.Device73CurrentFrame = gocv.NewMat()
|
|
||||||
}
|
}
|
||||||
|
if is_get == false && device.NextStop == false { //如果设备流已经停止且不暂停,则开启
|
||||||
go service.GetVideoStream(device.ID)
|
go service.GetVideoStream(device.ID)
|
||||||
log.Printf("device:%d has started!\n", device.ID)
|
log.Printf("device:%d has started!\n", device.ID)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if logCount%3600 == 0 {
|
log.Println("每10秒执行一次,当前设备:", proto.Config.DeviceInfo)
|
||||||
log.Printf("每%d秒执行一次,当前设备:%v", logCount*10, proto.Config.DeviceInfo)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func JWTAuthMiddleware() gin.HandlerFunc {
|
func JWTAuthMiddleware() gin.HandlerFunc {
|
||||||
|
|
@ -180,8 +152,6 @@ func JWTAuthMiddleware() gin.HandlerFunc {
|
||||||
//查看token是否在超级token中
|
//查看token是否在超级token中
|
||||||
if worker.IsContainSet("super_permission_tokens", tokenString) {
|
if worker.IsContainSet("super_permission_tokens", tokenString) {
|
||||||
s_id := c.Request.Header.Get("super_id")
|
s_id := c.Request.Header.Get("super_id")
|
||||||
if s_id == "" {
|
|
||||||
s_id = c.Query("super_id")
|
|
||||||
if s_id == "" {
|
if s_id == "" {
|
||||||
c.AbortWithStatus(200)
|
c.AbortWithStatus(200)
|
||||||
c.JSON(200, gin.H{
|
c.JSON(200, gin.H{
|
||||||
|
|
@ -191,7 +161,6 @@ func JWTAuthMiddleware() gin.HandlerFunc {
|
||||||
})
|
})
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
}
|
|
||||||
id, _ := strconv.Atoi(s_id)
|
id, _ := strconv.Atoi(s_id)
|
||||||
//查看s_id类型
|
//查看s_id类型
|
||||||
c.Set("id", id)
|
c.Set("id", id)
|
||||||
|
|
|
||||||
159
service/tool.go
159
service/tool.go
|
|
@ -16,78 +16,120 @@ import (
|
||||||
|
|
||||||
var DeviceRWMap = &sync.Map{}
|
var DeviceRWMap = &sync.Map{}
|
||||||
var DeviceCurrentFrameMap = &sync.Map{}
|
var DeviceCurrentFrameMap = &sync.Map{}
|
||||||
var DeviceFrameCount map[int]int
|
var DeviceFrameCount = &sync.Map{}
|
||||||
var DeviceIsGettingFrame = &sync.Map{}
|
var DeviceIsGettingFrame = &sync.Map{}
|
||||||
var Device1CurrentFrame []byte
|
|
||||||
var Device50CurrentFrame gocv.Mat
|
|
||||||
var Device73CurrentFrame gocv.Mat
|
|
||||||
|
|
||||||
func SetDeviceCurrentFrame(frame gocv.Mat, deviceId int) error {
|
func SetDeviceCurrentFrame(frame gocv.Mat, device_id int) error {
|
||||||
//获取读写锁
|
//获取读写锁
|
||||||
mutex_, ok := DeviceRWMap.Load(deviceId)
|
mutex_, ok := DeviceRWMap.Load(device_id)
|
||||||
if !ok {
|
if !ok {
|
||||||
return fmt.Errorf("设备:%s 读写锁不存在", deviceId)
|
return fmt.Errorf("设备:%s 读写锁不存在", device_id)
|
||||||
}
|
}
|
||||||
mutex := mutex_.(*sync.RWMutex)
|
mutex := mutex_.(*sync.RWMutex)
|
||||||
mutex.Lock()
|
mutex.Lock()
|
||||||
defer mutex.Unlock()
|
defer mutex.Unlock()
|
||||||
//获取前一帧,将前一帧释放
|
//获取前一帧,将前一帧释放
|
||||||
framePrev, ok := DeviceCurrentFrameMap.Load(deviceId)
|
framePrev, ok := DeviceCurrentFrameMap.Load(device_id)
|
||||||
if ok {
|
if ok {
|
||||||
frame_, ok2 := framePrev.(gocv.Mat)
|
frame_, ok2 := framePrev.(gocv.Mat)
|
||||||
if ok2 {
|
if ok2 {
|
||||||
err2 := frame_.Close()
|
err2 := frame_.Close()
|
||||||
if err2 != nil {
|
if err2 != nil {
|
||||||
log.Printf("设备:%d, 错误: 无法关闭帧\n", deviceId)
|
log.Printf("设备:%d, 错误: 无法关闭帧\n", device_id)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
//设置当前帧
|
//设置当前帧
|
||||||
DeviceCurrentFrameMap.Store(deviceId, frame)
|
DeviceCurrentFrameMap.Store(device_id, frame)
|
||||||
frameCount, ok := DeviceFrameCount[deviceId]
|
frame_count, ok := DeviceFrameCount.Load(device_id)
|
||||||
if !ok {
|
if !ok {
|
||||||
return fmt.Errorf("设备:%s 当前帧计数不存在", deviceId)
|
return fmt.Errorf("设备:%s 当前帧计数不存在", device_id)
|
||||||
}
|
}
|
||||||
frameCount++
|
frame_count_ := frame_count.(int)
|
||||||
DeviceFrameCount[deviceId] = frameCount
|
frame_count_++
|
||||||
|
DeviceFrameCount.Store(device_id, frame_count_)
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func SetDeviceCurrentFrameV2(frame *gocv.Mat, deviceId int) error {
|
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)
|
mutex_, ok := DeviceRWMap.Load(deviceId)
|
||||||
if !ok {
|
if !ok {
|
||||||
return fmt.Errorf("设备:%s 读写锁不存在", deviceId)
|
log.Printf("DeviceRWMap 读写锁不存在,device_id: %d \n", deviceId)
|
||||||
|
return -1
|
||||||
}
|
}
|
||||||
mutex := mutex_.(*sync.RWMutex)
|
mutex, ok := mutex_.(*sync.RWMutex)
|
||||||
mutex.Lock()
|
|
||||||
defer mutex.Unlock()
|
|
||||||
//设置当前帧
|
|
||||||
switch deviceId {
|
|
||||||
case 1:
|
|
||||||
Device1CurrentFrame = (*frame).ToBytes()
|
|
||||||
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 {
|
if !ok {
|
||||||
return fmt.Errorf("设备:%s 当前帧计数不存在", deviceId)
|
log.Printf("DeviceRWMap 存储的不是 *sync.RWMutex 类型,device_id: %d \n", deviceId)
|
||||||
|
return -1
|
||||||
}
|
}
|
||||||
if frameCount%10 == 0 {
|
mutex.RLock()
|
||||||
log.Printf("设备:%d 当前帧: %d 已更新\n", deviceId, frameCount)
|
defer mutex.RUnlock()
|
||||||
|
var frame_ gocv.Mat
|
||||||
|
//获取当前帧
|
||||||
|
frameIface, ok := DeviceCurrentFrameMap.Load(deviceId)
|
||||||
|
if !ok {
|
||||||
|
return -1
|
||||||
}
|
}
|
||||||
frameCount++
|
frame_, ok = frameIface.(gocv.Mat)
|
||||||
DeviceFrameCount[deviceId] = frameCount
|
if !ok {
|
||||||
return nil
|
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 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)
|
||||||
|
}
|
||||||
|
frame_countIface, ok := DeviceFrameCount.Load(deviceId)
|
||||||
|
if !ok {
|
||||||
|
return gocv.NewMat(), -1
|
||||||
|
}
|
||||||
|
frame_count, ok := frame_countIface.(int)
|
||||||
|
if !ok {
|
||||||
|
log.Printf("DeviceFrameCount 存储的不是 int 类型,device_id: %d", deviceId)
|
||||||
|
}
|
||||||
|
return frame, frame_count
|
||||||
}
|
}
|
||||||
|
|
||||||
func getVideoFrame(device proto.DeviceInfo) {
|
func getVideoFrame(device proto.DeviceInfo) {
|
||||||
|
|
@ -115,7 +157,6 @@ func getVideoFrame(device proto.DeviceInfo) {
|
||||||
fontColor := color.RGBA{G: 255}
|
fontColor := color.RGBA{G: 255}
|
||||||
lineType := 2
|
lineType := 2
|
||||||
z := 0
|
z := 0
|
||||||
var frame gocv.Mat
|
|
||||||
for {
|
for {
|
||||||
if device.LogFrame > 0 && z%device.LogFrame == 0 {
|
if device.LogFrame > 0 && z%device.LogFrame == 0 {
|
||||||
log.Printf("设备:%d 当前帧: %d\n", device.ID, z)
|
log.Printf("设备:%d 当前帧: %d\n", device.ID, z)
|
||||||
|
|
@ -123,7 +164,7 @@ func getVideoFrame(device proto.DeviceInfo) {
|
||||||
if device.NextStop {
|
if device.NextStop {
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
frame = gocv.NewMat()
|
frame := gocv.NewMat()
|
||||||
ok := webcam.Read(&frame)
|
ok := webcam.Read(&frame)
|
||||||
if !ok {
|
if !ok {
|
||||||
log.Printf("设备:%v 错误: 无法从视频流中读取帧\n", device)
|
log.Printf("设备:%v 错误: 无法从视频流中读取帧\n", device)
|
||||||
|
|
@ -144,21 +185,11 @@ func getVideoFrame(device proto.DeviceInfo) {
|
||||||
currentTime := time.Now().Format("2006-01-02 15:04:05")
|
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{10, 20}, font, fontScale, fontColor, lineType)
|
||||||
//需要将帧付给全局变量
|
//需要将帧付给全局变量
|
||||||
err3 := SetDeviceCurrentFrameV2(&frame, device.ID)
|
err3 := SetDeviceCurrentFrame(frame, device.ID)
|
||||||
if err3 != nil {
|
if err3 != nil {
|
||||||
log.Printf("设备:%d 错误: 无法设置当前帧,err:%s \n", device.ID, err3.Error())
|
log.Printf("设备:%d 错误: 无法设置当前帧,err:%s \n", device.ID, err3.Error())
|
||||||
}
|
}
|
||||||
z++
|
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())
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -182,9 +213,13 @@ func GetVideoStream(id int) {
|
||||||
log.Printf("设备:%d 错误: %v\n", id, err)
|
log.Printf("设备:%d 错误: %v\n", id, err)
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
isGetting := worker.GetRedis(fmt.Sprintf("device_%d_is_getting", id))
|
is_get, ok := DeviceIsGettingFrame.Load(id)
|
||||||
if isGetting == "true" || isGetting == "" {
|
if !ok {
|
||||||
log.Printf("设备:%d 正在运行,isGetting:%s", id, isGetting)
|
log.Printf("device: %d not found", id)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if is_get == true {
|
||||||
|
log.Printf("device: %d is running!", id)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
for {
|
for {
|
||||||
|
|
@ -203,9 +238,9 @@ func GetVideoStream(id int) {
|
||||||
log.Printf("device: %d not found", id)
|
log.Printf("device: %d not found", id)
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
isGetting = worker.GetRedis(fmt.Sprintf("device_%d_is_getting", id))
|
is_get, ok = DeviceIsGettingFrame.Load(id)
|
||||||
if isGetting == "true" {
|
if is_get == true {
|
||||||
log.Printf("设备:%d 正在运行,isGetting:%s", id, isGetting)
|
log.Printf("for device:%d is running!", id)
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -216,11 +251,9 @@ func GetVideoStream(id int) {
|
||||||
//设置设备控制信息
|
//设置设备控制信息
|
||||||
status := Get(device.Control)
|
status := Get(device.Control)
|
||||||
DeviceIsGettingFrame.Store(id, true)
|
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)
|
log.Printf("device: %d set control info status: %d", device.ID, status)
|
||||||
getVideoFrame(device)
|
getVideoFrame(device)
|
||||||
DeviceIsGettingFrame.Store(id, false)
|
DeviceIsGettingFrame.Store(id, false)
|
||||||
worker.SetRedis(fmt.Sprintf("device_%d_is_getting", id), "false")
|
|
||||||
//等待1s
|
//等待1s
|
||||||
time.Sleep(1 * time.Second)
|
time.Sleep(1 * time.Second)
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue