视频分析2.0 多进程拆分仓库
zhangmeng
2019-05-06 788f46d4e67c08c7e4f26f62c90f8d85849bcf18
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
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
)
 
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")
}
 
// 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 (
    mapCI = make(map[string]CameraInfo)
    port  = 7001
)
 
func recvFromIPC(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)
    go recvFromIPC(ctx, "tcp://192.168.1.124:7000", ch)
 
    // demo.SendByIPC("rtsp://admin:a1234567@192.168.1.188:554/h264/ch1/main/av_stream", "cid0",
    //     "tcp://192.168.1.140:7000", false)
    for {
        select {
        case <-ctx.Done():
            return
        case c := <-ch:
            if _, ok := mapCI[c.ID]; !ok {
                mapCI[c.ID] = c
                ipc := "tcp://192.168.1.124:" + strconv.Itoa(port)
                port++
                fmt.Printf("create ipc %s for decode : %s, on camera id %s\n", ipc, c.URL, c.ID)
 
                url := strings.TrimSpace(c.URL)
                id := strings.TrimSpace(c.ID)
                i := strings.TrimSpace(ipc)
                go runSender(id, url, i)
            }
        }
    }
    cancel()
}
 
func runSender(cameraID, rtspURL, ipcLabel string) {
    d := srv.NewSender(cameraID, rtspURL, ipcLabel)
    d.RunAsClient()
}
func test() {
    fmt.Println("start test")
 
    fmt.Println(picFolder)
 
    runSender("cameraid", streamURL, ipcURL)
}