diff --git a/revid/inputs.go b/revid/inputs.go new file mode 100644 index 00000000..4df3abb3 --- /dev/null +++ b/revid/inputs.go @@ -0,0 +1,281 @@ +/* +DESCRIPTION + inputs.go contains code for interfacing with various revid inputs to obtain + input data streams. + +AUTHORS + Saxon A. Nelson-Milton + Alan Noble + Dan Kortschak + Trek Hopton + +LICENSE + revid is Copyright (C) 2017-2019 the Australian Ocean Lab (AusOcean) + + It is free software: you can redistribute it and/or modify them + under the terms of the GNU General Public License as published by the + Free Software Foundation, either version 3 of the License, or (at your + option) any later version. + + It is distributed in the hope that it will be useful, but WITHOUT + ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or + FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License + for more details. + + You should have received a copy of the GNU General Public License + in gpl.txt. If not, see http://www.gnu.org/licenses. +*/ + +package revid + +import ( + "errors" + "fmt" + "io" + "net" + "os" + "os/exec" + "strconv" + "strings" + "time" + + "bitbucket.org/ausocean/av/codec/codecutil" + "bitbucket.org/ausocean/av/protocol/rtcp" + "bitbucket.org/ausocean/av/protocol/rtp" + "bitbucket.org/ausocean/av/protocol/rtsp" + "bitbucket.org/ausocean/utils/logger" +) + +// startRaspivid sets up things for input from raspivid i.e. starts +// a raspivid process and pipes it's data output. +func (r *Revid) startRaspivid() (func() error, error) { + r.config.Logger.Log(logger.Info, pkg+"starting raspivid") + + const disabled = "0" + args := []string{ + "--output", "-", + "--nopreview", + "--timeout", disabled, + "--width", fmt.Sprint(r.config.Width), + "--height", fmt.Sprint(r.config.Height), + "--bitrate", fmt.Sprint(r.config.Bitrate), + "--framerate", fmt.Sprint(r.config.FrameRate), + "--rotation", fmt.Sprint(r.config.Rotation), + "--brightness", fmt.Sprint(r.config.Brightness), + "--saturation", fmt.Sprint(r.config.Saturation), + "--exposure", fmt.Sprint(r.config.Exposure), + "--awb", fmt.Sprint(r.config.AutoWhiteBalance), + } + + if r.config.FlipHorizontal { + args = append(args, "--hflip") + } + + if r.config.FlipVertical { + args = append(args, "--vflip") + } + if r.config.FlipHorizontal { + args = append(args, "--hflip") + } + + switch r.config.InputCodec { + default: + return nil, fmt.Errorf("revid: invalid input codec: %v", r.config.InputCodec) + case codecutil.H264: + args = append(args, + "--codec", "H264", + "--inline", + "--intra", fmt.Sprint(r.config.MinFrames), + ) + if r.config.Quantization != 0 { + args = append(args, "-qp", fmt.Sprint(r.config.Quantization)) + } + case codecutil.MJPEG: + args = append(args, "--codec", "MJPEG") + } + r.config.Logger.Log(logger.Info, pkg+"raspivid args", "raspividArgs", strings.Join(args, " ")) + r.cmd = exec.Command("raspivid", args...) + + stdout, err := r.cmd.StdoutPipe() + if err != nil { + return nil, err + } + err = r.cmd.Start() + if err != nil { + r.config.Logger.Log(logger.Fatal, pkg+"cannot start raspivid", "error", err.Error()) + } + + r.wg.Add(1) + go r.processFrom(stdout, 0) + return nil, nil +} + +// startV4l sets up webcam input and starts the revid.processFrom routine. +func (r *Revid) startV4L() (func() error, error) { + const defaultVideo = "/dev/video0" + + r.config.Logger.Log(logger.Info, pkg+"starting webcam") + if r.config.InputPath == "" { + r.config.Logger.Log(logger.Info, pkg+"using default video device", "device", defaultVideo) + r.config.InputPath = defaultVideo + } + + args := []string{ + "-i", r.config.InputPath, + "-f", "h264", + "-r", fmt.Sprint(r.config.FrameRate), + } + + args = append(args, + "-b:v", fmt.Sprint(r.config.Bitrate), + "-maxrate", fmt.Sprint(r.config.Bitrate), + "-bufsize", fmt.Sprint(r.config.Bitrate/2), + "-s", fmt.Sprintf("%dx%d", r.config.Width, r.config.Height), + "-", + ) + + r.config.Logger.Log(logger.Info, pkg+"ffmpeg args", "args", strings.Join(args, " ")) + r.cmd = exec.Command("ffmpeg", args...) + + stdout, err := r.cmd.StdoutPipe() + if err != nil { + return nil, nil + } + + err = r.cmd.Start() + if err != nil { + r.config.Logger.Log(logger.Fatal, pkg+"cannot start webcam", "error", err.Error()) + return nil, nil + } + + r.wg.Add(1) + go r.processFrom(stdout, time.Duration(0)) + return nil, nil +} + +// setupInputForFile sets up input from file and starts the revid.processFrom +// routine. +func (r *Revid) setupInputForFile() (func() error, error) { + f, err := os.Open(r.config.InputPath) + if err != nil { + r.config.Logger.Log(logger.Error, err.Error()) + r.Stop() + return nil, err + } + + // TODO(kortschak): Maybe we want a context.Context-aware parser that we can stop. + r.wg.Add(1) + go r.processFrom(f, 0) + return func() error { return f.Close() }, nil +} + +// startRTSPCamera uses RTSP to request an RTP stream from an IP camera. An RTP +// client is created from which RTP packets containing either h264/h265 can be +// read by the selected lexer. +func (r *Revid) startRTSPCamera() (func() error, error) { + rtspClt, local, remote, err := rtsp.NewClient(r.config.RTSPURL) + if err != nil { + return nil, err + } + + resp, err := rtspClt.Options() + if err != nil { + return nil, err + } + r.config.Logger.Log(logger.Info, pkg+"RTSP OPTIONS response", "response", resp.String()) + + resp, err = rtspClt.Describe() + if err != nil { + return nil, err + } + r.config.Logger.Log(logger.Info, pkg+"RTSP DESCRIBE response", "response", resp.String()) + + resp, err = rtspClt.Setup("track1", fmt.Sprintf("RTP/AVP;unicast;client_port=%d-%d", rtpPort, rtcpPort)) + if err != nil { + return nil, err + } + r.config.Logger.Log(logger.Info, pkg+"RTSP SETUP response", "response", resp.String()) + rtpCltAddr, rtcpCltAddr, rtcpSvrAddr, err := formAddrs(local, remote, *resp) + if err != nil { + return nil, err + } + + resp, err = rtspClt.Play() + if err != nil { + return nil, err + } + r.config.Logger.Log(logger.Info, pkg+"RTSP server PLAY response", "response", resp.String()) + + rtpClt, err := rtp.NewClient(rtpCltAddr) + if err != nil { + return nil, err + } + + rtcpClt, err := rtcp.NewClient(rtcpCltAddr, rtcpSvrAddr, rtpClt, r.config.Logger.Log) + if err != nil { + return nil, err + } + + // Check errors from RTCP client until it has stopped running. + go func() { + for { + err, ok := <-rtcpClt.Err() + if ok { + r.config.Logger.Log(logger.Warning, pkg+"RTCP error", "error", err.Error()) + } else { + return + } + } + }() + + // Start the RTCP client. + rtcpClt.Start() + + // Start reading data from the RTP client. + r.wg.Add(1) + go r.processFrom(rtpClt, time.Second/time.Duration(r.config.FrameRate)) + + return func() error { + rtspClt.Close() + rtcpClt.Stop() + return nil + }, nil +} + +// formAddrs is a helper function to form the addresses for the RTP client, +// RTCP client, and the RTSP server's RTCP addr using the local, remote addresses +// of the RTSP conn, and the SETUP method response. +func formAddrs(local, remote *net.TCPAddr, setupResp rtsp.Response) (rtpCltAddr, rtcpCltAddr, rtcpSvrAddr string, err error) { + svrRTCPPort, err := parseSvrRTCPPort(setupResp) + if err != nil { + return "", "", "", err + } + rtpCltAddr = strings.Split(local.String(), ":")[0] + ":" + strconv.Itoa(rtpPort) + rtcpCltAddr = strings.Split(local.String(), ":")[0] + ":" + strconv.Itoa(rtcpPort) + rtcpSvrAddr = strings.Split(remote.String(), ":")[0] + ":" + strconv.Itoa(svrRTCPPort) + return +} + +// parseServerRTCPPort is a helper function to get the RTSP server's RTCP port. +func parseSvrRTCPPort(resp rtsp.Response) (int, error) { + transport := resp.Header.Get("Transport") + for _, p := range strings.Split(transport, ";") { + if strings.Contains(p, "server_port") { + port, err := strconv.Atoi(strings.Split(p, "-")[1]) + if err != nil { + return 0, err + } + return port, nil + } + } + return 0, errors.New("SETUP response did not provide RTCP port") +} + +// processFrom is run as a routine to read from a input data source, lex and +// then send individual access units to revid's encoders. +func (r *Revid) processFrom(read io.Reader, delay time.Duration) { + r.config.Logger.Log(logger.Info, pkg+"reading input data") + r.err <- r.lexTo(r.encoders, read, delay) + r.config.Logger.Log(logger.Info, pkg+"finished reading input data") + r.wg.Done() +} diff --git a/revid/revid.go b/revid/revid.go index 085fcdd4..1c5caac3 100644 --- a/revid/revid.go +++ b/revid/revid.go @@ -32,8 +32,6 @@ import ( "errors" "fmt" "io" - "net" - "os" "os/exec" "strconv" "strings" @@ -46,9 +44,6 @@ import ( "bitbucket.org/ausocean/av/codec/mjpeg" "bitbucket.org/ausocean/av/container/flv" "bitbucket.org/ausocean/av/container/mts" - "bitbucket.org/ausocean/av/protocol/rtcp" - "bitbucket.org/ausocean/av/protocol/rtp" - "bitbucket.org/ausocean/av/protocol/rtsp" "bitbucket.org/ausocean/iot/pi/netsender" "bitbucket.org/ausocean/utils/ioext" "bitbucket.org/ausocean/utils/logger" @@ -576,233 +571,3 @@ func (r *Revid) Update(vars map[string]string) error { r.config.Logger.Log(logger.Info, pkg+"revid config changed", "config", fmt.Sprintf("%+v", r.config)) return nil } - -// startRaspivid sets up things for input from raspivid i.e. starts -// a raspivid process and pipes it's data output. -func (r *Revid) startRaspivid() (func() error, error) { - r.config.Logger.Log(logger.Info, pkg+"starting raspivid") - - const disabled = "0" - args := []string{ - "--output", "-", - "--nopreview", - "--timeout", disabled, - "--width", fmt.Sprint(r.config.Width), - "--height", fmt.Sprint(r.config.Height), - "--bitrate", fmt.Sprint(r.config.Bitrate), - "--framerate", fmt.Sprint(r.config.FrameRate), - "--rotation", fmt.Sprint(r.config.Rotation), - "--brightness", fmt.Sprint(r.config.Brightness), - "--saturation", fmt.Sprint(r.config.Saturation), - "--exposure", fmt.Sprint(r.config.Exposure), - "--awb", fmt.Sprint(r.config.AutoWhiteBalance), - } - - if r.config.FlipHorizontal { - args = append(args, "--hflip") - } - - if r.config.FlipVertical { - args = append(args, "--vflip") - } - if r.config.FlipHorizontal { - args = append(args, "--hflip") - } - - switch r.config.InputCodec { - default: - return nil, fmt.Errorf("revid: invalid input codec: %v", r.config.InputCodec) - case codecutil.H264: - args = append(args, - "--codec", "H264", - "--inline", - "--intra", fmt.Sprint(r.config.MinFrames), - ) - if r.config.Quantization != 0 { - args = append(args, "-qp", fmt.Sprint(r.config.Quantization)) - } - case codecutil.MJPEG: - args = append(args, "--codec", "MJPEG") - } - r.config.Logger.Log(logger.Info, pkg+"raspivid args", "raspividArgs", strings.Join(args, " ")) - r.cmd = exec.Command("raspivid", args...) - - stdout, err := r.cmd.StdoutPipe() - if err != nil { - return nil, err - } - err = r.cmd.Start() - if err != nil { - r.config.Logger.Log(logger.Fatal, pkg+"cannot start raspivid", "error", err.Error()) - } - - r.wg.Add(1) - go r.processFrom(stdout, 0) - return nil, nil -} - -func (r *Revid) startV4L() (func() error, error) { - const defaultVideo = "/dev/video0" - - r.config.Logger.Log(logger.Info, pkg+"starting webcam") - if r.config.InputPath == "" { - r.config.Logger.Log(logger.Info, pkg+"using default video device", "device", defaultVideo) - r.config.InputPath = defaultVideo - } - - args := []string{ - "-i", r.config.InputPath, - "-f", "h264", - "-r", fmt.Sprint(r.config.FrameRate), - } - - args = append(args, - "-b:v", fmt.Sprint(r.config.Bitrate), - "-maxrate", fmt.Sprint(r.config.Bitrate), - "-bufsize", fmt.Sprint(r.config.Bitrate/2), - "-s", fmt.Sprintf("%dx%d", r.config.Width, r.config.Height), - "-", - ) - - r.config.Logger.Log(logger.Info, pkg+"ffmpeg args", "args", strings.Join(args, " ")) - r.cmd = exec.Command("ffmpeg", args...) - - stdout, err := r.cmd.StdoutPipe() - if err != nil { - return nil, nil - } - - err = r.cmd.Start() - if err != nil { - r.config.Logger.Log(logger.Fatal, pkg+"cannot start webcam", "error", err.Error()) - return nil, nil - } - - r.wg.Add(1) - go r.processFrom(stdout, time.Duration(0)) - return nil, nil -} - -// setupInputForFile sets things up for getting input from a file -func (r *Revid) setupInputForFile() (func() error, error) { - f, err := os.Open(r.config.InputPath) - if err != nil { - r.config.Logger.Log(logger.Error, err.Error()) - r.Stop() - return nil, err - } - - // TODO(kortschak): Maybe we want a context.Context-aware parser that we can stop. - r.wg.Add(1) - go r.processFrom(f, 0) - return func() error { return f.Close() }, nil -} - -// startRTSPCamera uses RTSP to request an RTP stream from an IP camera. An RTP -// client is created from which RTP packets containing either h264/h265 can read -// by the selected lexer. -func (r *Revid) startRTSPCamera() (func() error, error) { - rtspClt, local, remote, err := rtsp.NewClient(r.config.RTSPURL) - if err != nil { - return nil, err - } - - resp, err := rtspClt.Options() - if err != nil { - return nil, err - } - r.config.Logger.Log(logger.Info, pkg+"RTSP OPTIONS response", "response", resp.String()) - - resp, err = rtspClt.Describe() - if err != nil { - return nil, err - } - r.config.Logger.Log(logger.Info, pkg+"RTSP DESCRIBE response", "response", resp.String()) - - resp, err = rtspClt.Setup("track1", fmt.Sprintf("RTP/AVP;unicast;client_port=%d-%d", rtpPort, rtcpPort)) - if err != nil { - return nil, err - } - r.config.Logger.Log(logger.Info, pkg+"RTSP SETUP response", "response", resp.String()) - rtpCltAddr, rtcpCltAddr, rtcpSvrAddr, err := formAddrs(local, remote, *resp) - if err != nil { - return nil, err - } - - resp, err = rtspClt.Play() - if err != nil { - return nil, err - } - r.config.Logger.Log(logger.Info, pkg+"RTSP server PLAY response", "response", resp.String()) - - rtpClt, err := rtp.NewClient(rtpCltAddr) - if err != nil { - return nil, err - } - - rtcpClt, err := rtcp.NewClient(rtcpCltAddr, rtcpSvrAddr, rtpClt, r.config.Logger.Log) - if err != nil { - return nil, err - } - - // Check errors from RTCP client until it has stopped running. - go func() { - for { - err, ok := <-rtcpClt.Err() - if ok { - r.config.Logger.Log(logger.Warning, pkg+"RTCP error", "error", err.Error()) - } else { - return - } - } - }() - - // Start the RTCP client. - rtcpClt.Start() - - // Start reading data from the RTP client. - r.wg.Add(1) - go r.processFrom(rtpClt, time.Second/time.Duration(r.config.FrameRate)) - - return func() error { - rtspClt.Close() - rtcpClt.Stop() - return nil - }, nil -} - -// formAddrs is a helper function to form the addresses for the RTP client, -// RTCP client, and the RTSP server's RTCP addr using the local, remote addresses -// of the RTSP conn, and the SETUP method response. -func formAddrs(local, remote *net.TCPAddr, setupResp rtsp.Response) (rtpCltAddr, rtcpCltAddr, rtcpSvrAddr string, err error) { - svrRTCPPort, err := parseSvrRTCPPort(setupResp) - if err != nil { - return "", "", "", err - } - rtpCltAddr = strings.Split(local.String(), ":")[0] + ":" + strconv.Itoa(rtpPort) - rtcpCltAddr = strings.Split(local.String(), ":")[0] + ":" + strconv.Itoa(rtcpPort) - rtcpSvrAddr = strings.Split(remote.String(), ":")[0] + ":" + strconv.Itoa(svrRTCPPort) - return -} - -// parseServerRTCPPort is a helper function to get the RTSP server's RTCP port. -func parseSvrRTCPPort(resp rtsp.Response) (int, error) { - transport := resp.Header.Get("Transport") - for _, p := range strings.Split(transport, ";") { - if strings.Contains(p, "server_port") { - port, err := strconv.Atoi(strings.Split(p, "-")[1]) - if err != nil { - return 0, err - } - return port, nil - } - } - return 0, errors.New("SETUP response did not provide RTCP port") -} - -func (r *Revid) processFrom(read io.Reader, delay time.Duration) { - r.config.Logger.Log(logger.Info, pkg+"reading input data") - r.err <- r.lexTo(r.encoders, read, delay) - r.config.Logger.Log(logger.Info, pkg+"finished reading input data") - r.wg.Done() -}