视频分析2.0 多进程拆分仓库
zhangmeng
2019-05-06 e67660d0bf785eb2feaceee1ae96d3a66b23ba49
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
package main
 
import (
    "context"
    "decoder/valib/ipc"
    srv "decoder/work/service"
    "encoding/json"
    "flag"
    "fmt"
    "strconv"
    "strings"
)
 
var (
    streamURL string
    picFolder string
 
    ipcURL string
    proc   string
 
    testIt bool
 
    asServer bool
)
 
func init() {
    flag.StringVar(&streamURL, "i", "rtsp://192.168.1.203:8554/16.mkv", "input url")
    flag.StringVar(&picFolder, "f", ".", "test pic folder")
 
    flag.StringVar(&ipcURL, "ipc", "ipc:///tmp/pic.ipc", "ipc label")
 
    flag.BoolVar(&testIt, "test", false, "use test")
 
    flag.BoolVar(&asServer, "server", false, "run ipc as server")
}
 
// CameraInfo camera info
type CameraInfo struct {
    ID  string `json:"Cameraid"`
    URL string `json:"Rtsp"`
}
 
// MsgIPC msg for ipc
type MsgIPC struct {
    CMD  string `json:"Command"`
    Port int    `jsong:"PortIpc"`
}
 
var (
    mapCameraInfo = make(map[string]CameraInfo)
 
    tcp  = `tcp://192.168.1.124`
    port = 7000
)
 
func recvCameraInfoFromIPC(ctx context.Context, url string, ch chan<- CameraInfo) {
    ipc := ipc.NewClient(ctx, url)
 
    for {
        msg := ipc.Recv()
        if msg != nil {
            fmt.Println(string(msg))
            var c CameraInfo
            if err := json.Unmarshal(msg, &c); err == nil {
                ch <- c
                fmt.Printf("recv camere info %+v\n", c)
            } else {
                fmt.Println(err)
            }
            msgIpc := MsgIPC{"new decoder", port}
            if b, err := json.Marshal(msgIpc); err == nil {
                ipc.Send(b)
            }
        }
    }
}
 
func main() {
    flag.Parse()
 
    if testIt {
        test()
    }
 
    ctx, cancel := context.WithCancel(context.Background())
 
    ch := make(chan CameraInfo)
    tcpURL := tcp + strconv.Itoa(port)
    port++
    go recvCameraInfoFromIPC(ctx, tcpURL, ch)
 
    for {
        select {
        case <-ctx.Done():
            return
        case c := <-ch:
            if _, ok := mapCameraInfo[c.ID]; !ok {
                mapCameraInfo[c.ID] = c
                ipcAddr := tcp + strconv.Itoa(port)
                port++
                fmt.Printf("create ipc %s for decode : %s, on camera id %s\n", ipcAddr, c.URL, c.ID)
 
                url := strings.TrimSpace(c.URL)
                id := strings.TrimSpace(c.ID)
                addr := strings.TrimSpace(ipcAddr)
                go runSender(id, url, addr)
            }
        }
    }
    cancel()
}
 
func runSender(cameraID, rtspURL, ipcLabel string) {
    d := srv.NewSender(cameraID, rtspURL, ipcLabel)
    if asServer {
        d.RunAsServer()
    }
    d.RunAsClient()
}
func test() {
    fmt.Println("start test")
 
    fmt.Println(picFolder)
 
    runSender("cameraid", streamURL, ipcURL)
}