feat: 实现GB28181API服务

1. 实现云台控制功能;
2. 支持关闭实时流;

README.md

@@ -14,7 +14,7 @@
     "keepalive_interval": 60,
     "name": "模拟1078设备4",
     "sim_number":"13800138000",
-    "manufacturer":"github/lkmio",
+    "manufacturer":"github.com/lkmio",
     "model":"gb-cms",
     "firmware":"dev"
 }
@@ -35,9 +35,9 @@ sim_number: 对应的部标设备sim卡号, 唯一键
     "root_id": "34020000001400000001",
     "device_id": "34020000001310000001",
     "name": "视频通道",
-    "manufacturer": "github/lkmio",
+    "manufacturer": "github.com/lkmio",
     "model": "gb-cms",
-    "owner": "github/lkmio",
+    "owner": "github.com/lkmio",
     "channel_number": 1
 }
 

api.go

@@ -55,6 +55,7 @@ type QueryRecordParams struct {
 	StartTime string `json:"starttime"`
 	EndTime   string `json:"endtime"`
 	//Type_     string `json:"type"`
+	Command string `json:"command"` // 云台控制命令 left/up/right/down/zoomin/zoomout
 }
 
 type DeviceChannelID struct {
@@ -166,7 +167,7 @@ func startApiServer(addr string) {
 	// 统一处理live/playback/download请求
 	apiServer.router.HandleFunc("/api/v1/{action}/start", withVerify(common.WithFormDataParams(apiServer.OnInvite, InviteParams{})))
 	// 关闭国标流. 如果是实时流, 等收流或空闲超时自行删除. 回放或下载流立即删除.
-	apiServer.router.HandleFunc("/api/v1/stream/close", common.WithJsonParams(apiServer.OnCloseStream, &StreamIDParams{}))
+	apiServer.router.HandleFunc("/api/v1/stream/stop", withVerify(common.WithFormDataParams(apiServer.OnCloseStream, InviteParams{})))
 
 	apiServer.router.HandleFunc("/api/v1/device/list", withVerify(common.WithQueryStringParams(apiServer.OnDeviceList, QueryDeviceChannel{})))           // 查询设备列表
 	apiServer.router.HandleFunc("/api/v1/device/channellist", withVerify(common.WithQueryStringParams(apiServer.OnChannelList, QueryDeviceChannel{})))   // 查询通道列表
@@ -174,9 +175,9 @@ func startApiServer(addr string) {
 	apiServer.router.HandleFunc("/api/v1/playback/recordlist", withVerify(common.WithQueryStringParams(apiServer.OnRecordList, QueryRecordParams{})))    // 查询录像列表
 	apiServer.router.HandleFunc("/api/v1/stream/info", withVerify(apiServer.OnStreamInfo))
 
-	apiServer.router.HandleFunc("/api/v1/position/sub", common.WithJsonResponse(apiServer.OnSubscribePosition, &DeviceChannelID{})) // 订阅移动位置
-	apiServer.router.HandleFunc("/api/v1/playback/seek", common.WithJsonResponse(apiServer.OnSeekPlayback, &SeekParams{}))          // 回放seek
-	apiServer.router.HandleFunc("/api/v1/control/ptz", apiServer.OnPTZControl)                                                      // 云台控制
+	apiServer.router.HandleFunc("/api/v1/position/sub", common.WithJsonResponse(apiServer.OnSubscribePosition, &DeviceChannelID{}))        // 订阅移动位置
+	apiServer.router.HandleFunc("/api/v1/playback/seek", common.WithJsonResponse(apiServer.OnSeekPlayback, &SeekParams{}))                 // 回放seek
+	apiServer.router.HandleFunc("/api/v1/control/ptz", withVerify(common.WithFormDataParams(apiServer.OnPTZControl, QueryRecordParams{}))) // 云台控制
 
 	apiServer.router.HandleFunc("/api/v1/platform/list", apiServer.OnPlatformList)                                                                 // 级联设备列表
 	apiServer.router.HandleFunc("/api/v1/platform/add", common.WithJsonResponse(apiServer.OnPlatformAdd, &dao.PlatformModel{}))                    // 添加级联设备
@@ -563,15 +564,17 @@ func (api *ApiServer) DoInvite(inviteType common.InviteType, params *InviteParam
 	return http.StatusOK, stream, nil
 }
 
-func (api *ApiServer) OnCloseStream(v *StreamIDParams, w http.ResponseWriter, _ *http.Request) {
-	//stream := StreamManager.Find(v.StreamID)
-	//
-	//// 等空闲或收流超时会自动关闭
-	//if stream != nil && stream.GetSinkCount() < 1 {
-	//	CloseStream(v.StreamID, true)
-	//}
+func (api *ApiServer) OnCloseStream(v *InviteParams, w http.ResponseWriter, _ *http.Request) (interface{}, error) {
+	streamID := common.GenerateStreamID(common.InviteTypePlay, v.DeviceID, v.ChannelID, "", "")
+	mode, err := dao.Stream.DeleteStream(streamID)
+	if err != nil {
+		log.Sugar.Errorf("删除流失败 err: %s", err.Error())
+		return nil, err
+	}
 
-	_ = common.HttpResponseOK(w, nil)
+	(&stack.Stream{mode}).Close(true, true)
+
+	return "OK", nil
 }
 
 // QueryDeviceChannel 查询设备和通道的参数
@@ -718,6 +721,11 @@ func (api *ApiServer) OnChannelList(q *QueryDeviceChannel, _ http.ResponseWriter
 		registerWay, _ := strconv.Atoi(channel.RegisterWay)
 		secrecy, _ := strconv.Atoi(channel.Secrecy)
 
+		streamID := common.GenerateStreamID(common.InviteTypePlay, channel.RootID, channel.DeviceID, "", "")
+		if stream, err := dao.Stream.QueryStream(streamID); err != nil || stream == nil {
+			streamID = ""
+		}
+
 		response.ChannelList = append(response.ChannelList, LiveGBSChannel{
 			Address:            channel.Address,
 			Altitude:           0,
@@ -776,7 +784,7 @@ func (api *ApiServer) OnChannelList(q *QueryDeviceChannel, _ http.ResponseWriter
 			SnapURL:            "",
 			Speed:              0,
 			Status:             channel.Status.String(),
-			StreamID:           "",
+			StreamID:           string(streamID), // 实时流ID
 			SubCount:           channel.SubCount,
 			UpdatedAt:          channel.UpdatedAt.Format("2006-01-02 15:04:05"),
 		})
@@ -909,8 +917,19 @@ func (api *ApiServer) OnSeekPlayback(v *SeekParams, _ http.ResponseWriter, _ *ht
 	return nil, nil
 }
 
-func (api *ApiServer) OnPTZControl(_ http.ResponseWriter, _ *http.Request) {
+func (api *ApiServer) OnPTZControl(v *QueryRecordParams, _ http.ResponseWriter, _ *http.Request) (interface{}, error) {
+	log.Sugar.Debugf("PTZ控制 %v", *v)
 
+	model, _ := dao.Device.QueryDevice(v.DeviceID)
+	if model == nil || !model.Online() {
+		log.Sugar.Errorf("PTZ控制失败, 设备离线 device: %s", v.DeviceID)
+		return nil, fmt.Errorf("设备离线")
+	}
+
+	device := &stack.Device{model}
+	device.ControlPTZ(v.Command, v.ChannelID)
+
+	return "OK", nil
 }
 
 func (api *ApiServer) OnHangup(v *BroadcastParams, _ http.ResponseWriter, _ *http.Request) (interface{}, error) {

stack/client.go

@@ -9,7 +9,7 @@ import (
 
 const (
 	DefaultDomainName   = "本域"
-	DefaultManufacturer = "github/lkmio"
+	DefaultManufacturer = "github.com/lkmio"
 	DefaultModel        = "gb-cms"
 	DefaultFirmware     = "dev"
 )

stack/client_benchmark_test_config.json

@@ -3,7 +3,7 @@
   "channel_id_prefix": "3402000000131",
   "server_id": "34020000002000000001",
   "?domain": "国标上级域的地址",
-  "domain": "160.202.253.143:15060",
+  "domain": "192.168.2.119:5060",
   "password": "12345678",
   "listenAddr": "192.168.2.119:15062",
   "count": 1,

stack/ptz_ctrl.go

@@ -0,0 +1,74 @@
+package stack
+
+import (
+	"fmt"
+	"gb-cms/common"
+)
+
+const (
+	DeviceControlFormat = "<?xml version=\"1.0\"?>\r\n" +
+		"<Control>\r\n" +
+		"<CmdType>DeviceControl</CmdType>\r\n" +
+		"<SN>%d</SN>\r\n" +
+		"<DeviceID>%s</DeviceID>\r\n" +
+		"<PTZCmd>%s</PTZCmd>\r\n" +
+		"</Control>\r\n"
+)
+
+// PTZCmd A.3.1 指令格式
+type PTZCmd struct {
+}
+
+func (c *PTZCmd) Unmarshal() {
+
+}
+
+func (c *PTZCmd) Marshal(cmd, horizontalSpeed, verticalSpeed, zoomSpeed byte) string {
+	checkCode := uint16(0xA5+0x0F+0x01+cmd+horizontalSpeed+verticalSpeed+(zoomSpeed&0xF0)) % 256
+	// 地址范围000H—FFFH（即0—4095），其中000H地址作为广播地址。
+	// 注： 前端设备控制中，不使用字节3和字节7的低4位地址码，使用前端设备控制消息体中的<DeviceID>统一编码标
+	// 识控制的前端设备。
+	// addr 12 bit
+	return fmt.Sprintf("A50F01%02X%02X%02X%02X%02X", cmd, horizontalSpeed, verticalSpeed, zoomSpeed, checkCode)
+}
+
+func (d *Device) ControlPTZ(command string, channelId string) {
+	var cmd byte
+	var horizontalSpeed, verticalSpeed, zoomSpeed byte = 0, 0, 0
+	switch command {
+	case "right":
+		cmd |= 1 << 0
+		horizontalSpeed = 30
+		break
+	case "left":
+		cmd |= 1 << 1
+		horizontalSpeed = 30
+		break
+	case "down":
+		cmd |= 1 << 2
+		verticalSpeed = 30
+		break
+	case "up":
+		cmd |= 1 << 3
+		verticalSpeed = 30
+		break
+	case "zoomin":
+		cmd |= 1 << 4
+		zoomSpeed = 30
+		break
+	case "zoomout":
+		cmd |= 1 << 5
+		zoomSpeed = 30
+		break
+	case "stop":
+		break
+	default:
+		return
+	}
+
+	ptzCmd := &PTZCmd{}
+	cmdHex := ptzCmd.Marshal(cmd, horizontalSpeed, verticalSpeed, zoomSpeed)
+	body := fmt.Sprintf(DeviceControlFormat, GetSN(), channelId, cmdHex)
+	request := d.BuildMessageRequest(channelId, body)
+	common.SipStack.SendRequest(request)
+}

stack/sip_server.go

@@ -389,7 +389,7 @@ func filterRequest(f func(wrapper *SipRequestSource)) gosip.RequestHandler {
 func StartSipServer(id, listenIP, publicIP string, listenPort int) (common.SipServer, error) {
 	ua := gosip.NewServer(gosip.ServerConfig{
 		Host:      publicIP,
-		UserAgent: "github/lkmio",
+		UserAgent: "github.com/lkmio",
 	}, nil, nil, common.Logger)
 
 	addr := net.JoinHostPort(listenIP, strconv.Itoa(listenPort))