/* 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/input/gvctrl" "bitbucket.org/ausocean/av/protocol/rtcp" "bitbucket.org/ausocean/av/protocol/rtp" "bitbucket.org/ausocean/av/protocol/rtsp" "bitbucket.org/ausocean/utils/logger" ) // TODO: remove this when gvctrl has configurable user and pass. const ( ipCamUser = "admin" ipCamPass = "admin" ) // AVDevice describes a configurable audio or video device from which media data // can be obtained. AVDevice implements io.Reader. type AVDevice interface { io.Reader // Set allows for configuration of the AVDevice using a Config struct. All, // some or none of the fields of the Config struct may be used for configuration // by an implementation. An implementation should specify what fields are // considered. Set(c Config) error // Start will start the AVDevice capturing media data; after which the Read // method may be called to obtain the data. The format of the data may differ // and should be specified by the implementation. Start() error // Stop will stop the AVDevice from capturing media data. From this point // Reads will no longer be successful. Stop() error } // multiError implements the built in error interface. multiError is used here // to collect multi errors during validation of configruation parameters for o // AVDevices. type multiError []error func (me multiError) Error() string { return fmt.Sprintf("%v", me) } // 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 * 1000), // Convert from kbps to bps. "--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.VBR { 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 { return nil, fmt.Errorf("could not start raspivid command: %w", err) } 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), } br := r.config.Bitrate * 1000 args = append(args, "-b:v", fmt.Sprint(br), "-maxrate", fmt.Sprint(br), "-bufsize", fmt.Sprint(br/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. // // TODO(saxon): this function should really be startGeoVision. It's much too // specific to be called startRTSPCamera. func (r *Revid) startRTSPCamera() (func() error, error) { r.config.Logger.Log(logger.Info, pkg+"starting geovision...") err := gvctrl.Set( r.config.CameraIP, gvctrl.Channel(r.config.CameraChan), gvctrl.CodecOut( map[uint8]gvctrl.Codec{ codecutil.H264: gvctrl.CodecH264, codecutil.H265: gvctrl.CodecH265, codecutil.MJPEG: gvctrl.CodecMJPEG, }[r.config.InputCodec], ), gvctrl.Height(int(r.config.Height)), gvctrl.FrameRate(int(r.config.FrameRate)), gvctrl.VariableBitrate(r.config.VBR), gvctrl.VBRQuality( map[quality]gvctrl.Quality{ qualityStandard: gvctrl.QualityStandard, qualityFair: gvctrl.QualityFair, qualityGood: gvctrl.QualityGood, qualityGreat: gvctrl.QualityGreat, qualityExcellent: gvctrl.QualityExcellent, }[r.config.VBRQuality], ), gvctrl.VBRBitrate(r.config.VBRBitrate), gvctrl.CBRBitrate(int(r.config.Bitrate)), gvctrl.Refresh(float64(r.config.MinFrames)/float64(r.config.FrameRate)), ) if err != nil { return nil, fmt.Errorf("could not set IPCamera settings: %w", err) } r.config.Logger.Log(logger.Info, pkg+"completed geovision configuration") time.Sleep(5 * time.Second) rtspClt, local, remote, err := rtsp.NewClient("rtsp://" + ipCamUser + ":" + ipCamPass + "@" + r.config.CameraIP + ":8554/" + "CH002.sdp") if err != nil { return nil, err } r.config.Logger.Log(logger.Info, pkg+"created RTSP client") resp, err := rtspClt.Options() if err != nil { return nil, err } r.config.Logger.Log(logger.Debug, pkg+"RTSP OPTIONS response", "response", resp.String()) resp, err = rtspClt.Describe() if err != nil { return nil, err } r.config.Logger.Log(logger.Debug, 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.Debug, pkg+"RTSP SETUP response", "response", resp.String()) rtpCltAddr, rtcpCltAddr, rtcpSvrAddr, err := formAddrs(local, remote, *resp) if err != nil { return nil, err } r.config.Logger.Log(logger.Info, pkg+"RTSP session setup complete") 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 } r.config.Logger.Log(logger.Info, pkg+"RTCP and RTP clients created") // 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() r.config.Logger.Log(logger.Info, pkg+"RTCP client started") // Start reading data from the RTP client. r.wg.Add(1) go r.processFrom(rtpClt, time.Second/time.Duration(r.config.FrameRate)) r.config.Logger.Log(logger.Info, pkg+"started input processor") resp, err = rtspClt.Play() if err != nil { return nil, err } r.config.Logger.Log(logger.Debug, pkg+"RTSP server PLAY response", "response", resp.String()) r.config.Logger.Log(logger.Info, pkg+"play requested, now receiving stream") return func() error { err := rtpClt.Close() if err != nil { return fmt.Errorf("could not close RTP client: %w", err) } err = rtspClt.Close() if err != nil { return fmt.Errorf("could not close RTSP client: %w", err) } rtcpClt.Stop() r.config.Logger.Log(logger.Info, pkg+"RTP, RTSP and RTCP clients stopped and closed") 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.err <- r.lexTo(r.encoders, read, delay) r.config.Logger.Log(logger.Info, pkg+"finished lexing") r.wg.Done() }