mirror of https://bitbucket.org/ausocean/av.git
Merged in move-inputs (pull request #258)
revid: moved code relating to inputs to separate file called inputs.go Approved-by: kortschak <dan@kortschak.io>
This commit is contained in:
Saxon Milton
commit
fad98c0dd2
|
|
@ -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 <saxon@ausocean.org>
|
||||
Alan Noble <alan@ausocean.org>
|
||||
Dan Kortschak <dan@ausocean.org>
|
||||
Trek Hopton <trek@ausocean.org>
|
||||
|
||||
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()
|
||||
}
|
||||
235
revid/revid.go
235
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()
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue