/* NAME revid.go AUTHORS Saxon A. Nelson-Milton Alan Noble Dan Kortschak Trek Hopton LICENSE revid is Copyright (C) 2017-2018 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 provides an API for reading, transcoding, and writing audio/video streams and files. package revid import ( "errors" "fmt" "io" "os/exec" "strconv" "strings" "sync" "time" "bitbucket.org/ausocean/av/codec/codecutil" "bitbucket.org/ausocean/av/codec/h264" "bitbucket.org/ausocean/av/codec/h265" "bitbucket.org/ausocean/av/codec/mjpeg" "bitbucket.org/ausocean/av/container/flv" "bitbucket.org/ausocean/av/container/mts" "bitbucket.org/ausocean/iot/pi/netsender" "bitbucket.org/ausocean/utils/ioext" "bitbucket.org/ausocean/utils/logger" "bitbucket.org/ausocean/utils/ring" ) // RTMP connection properties. const ( rtmpConnectionMaxTries = 5 rtmpConnectionTimeout = 10 ) const ( rtpPort = 60000 rtcpPort = 60001 defaultServerRTCPPort = 17301 ) const pkg = "revid: " type Logger interface { SetLevel(int8) Log(level int8, message string, params ...interface{}) } // Revid provides methods to control a revid session; providing methods // to start, stop and change the state of an instance using the Config struct. type Revid struct { // config holds the Revid configuration. // For historical reasons it also handles logging. // FIXME(kortschak): The relationship of concerns // in config/ns is weird. config Config // ns holds the netsender.Sender responsible for HTTP. ns *netsender.Sender // setupInput holds the current approach to setting up // the input stream. It returns a function used for cleaning up, and any errors. setupInput func() (func() error, error) // closeInput holds the cleanup function return from setupInput and is called // in Revid.Stop(). closeInput func() error // cmd is the exec'd process that may be used to produce // the input stream. // FIXME(kortschak): This should not exist. Replace this // with a context.Context cancellation. cmd *exec.Cmd // lexTo, encoder and packer handle transcoding the input stream. lexTo func(dest io.Writer, src io.Reader, delay time.Duration) error // encoders will hold the multiWriteCloser that writes to encoders from the lexer. encoders io.WriteCloser // running is used to keep track of revid's running state between methods. running bool // mu is used to protect isRunning during concurrent use. mu sync.Mutex // wg will be used to wait for any processing routines to finish. wg sync.WaitGroup // err will channel errors from revid routines to the handle errors routine. err chan error } // New returns a pointer to a new Revid with the desired configuration, and/or // an error if construction of the new instance was not successful. func New(c Config, ns *netsender.Sender) (*Revid, error) { r := Revid{ns: ns, err: make(chan error)} err := r.setConfig(c) if err != nil { return nil, fmt.Errorf("could not set config, failed with error: %v", err) } r.config.Logger.SetLevel(c.LogLevel) go r.handleErrors() return &r, nil } // Config returns a copy of revids current config. // // Config is not safe for concurrent use. func (r *Revid) Config() Config { return r.config } // TODO(Saxon): put more thought into error severity and how to handle these. func (r *Revid) handleErrors() { for { err := <-r.err if err != nil { r.config.Logger.Log(logger.Error, pkg+"async error", "error", err.Error()) } } } // Bitrate returns the result of the most recent bitrate check. // // TODO: get this working again. func (r *Revid) Bitrate() int { return -1 } // reset swaps the current config of a Revid with the passed // configuration; checking validity and returning errors if not valid. It then // sets up the data pipeline accordingly to this configuration. func (r *Revid) reset(config Config) error { err := r.setConfig(config) if err != nil { return err } r.config.Logger.SetLevel(config.LogLevel) err = r.setupPipeline( func(dst io.WriteCloser, fps float64) (io.WriteCloser, error) { var st int var encOptions []func(*mts.Encoder) error switch r.config.Input { case Raspivid: switch r.config.InputCodec { case codecutil.H264: st = mts.EncodeH264 case codecutil.MJPEG: st = mts.EncodeMJPEG encOptions = append(encOptions, mts.PacketBasedPSI(int(r.config.MinFrames))) default: panic("unknown input codec for raspivid input") } case File, V4L: st = mts.EncodeH264 case RTSP: switch r.config.InputCodec { case codecutil.H265: st = mts.EncodeH265 case codecutil.H264: st = mts.EncodeH264 case codecutil.MJPEG: st = mts.EncodeMJPEG encOptions = append(encOptions, mts.PacketBasedPSI(int(r.config.MinFrames))) default: panic("unknown input codec for RTSP input") } case Audio: st = mts.EncodeAudio default: panic("unknown input type") } return mts.NewEncoder(dst, float64(fps), st, encOptions...) }, func(dst io.WriteCloser, fps int) (io.WriteCloser, error) { return flv.NewEncoder(dst, true, true, fps) }, ioext.MultiWriteCloser, ) if err != nil { return err } return nil } // setConfig takes a config, checks it's validity and then replaces the current // revid config. func (r *Revid) setConfig(config Config) error { r.config.Logger = config.Logger err := config.Validate() if err != nil { return errors.New("Config struct is bad: " + err.Error()) } r.config = config return nil } // setupPipeline constructs the revid dataPipeline. Inputs, encoders and // senders are created and linked based on the current revid config. // // mtsEnc and flvEnc will be called to obtain an mts encoder and flv encoder // respectively. multiWriter will be used to create an ioext.multiWriteCloser // so that encoders can write to multiple senders. func (r *Revid) setupPipeline(mtsEnc func(dst io.WriteCloser, rate float64) (io.WriteCloser, error), flvEnc func(dst io.WriteCloser, rate int) (io.WriteCloser, error), multiWriter func(...io.WriteCloser) io.WriteCloser) error { // encoders will hold the encoders that are required for revid's current // configuration. var encoders []io.WriteCloser // mtsSenders will hold the senders the require MPEGTS encoding, and flvSenders // will hold senders that require FLV encoding. var mtsSenders, flvSenders []io.WriteCloser // We will go through our outputs and create the corresponding senders to add // to mtsSenders if the output requires MPEGTS encoding, or flvSenders if the // output requires FLV encoding. var w io.WriteCloser for _, out := range r.config.Outputs { switch out { case HTTP: w = newMtsSender( newHttpSender(r.ns, r.config.Logger.Log), r.config.Logger.Log, ring.NewBuffer(r.config.MTSRBSize, r.config.MTSRBElementSize, time.Duration(r.config.MTSRBWriteTimeout)*time.Second), r.config.ClipDuration, ) mtsSenders = append(mtsSenders, w) case RTP: w, err := newRtpSender(r.config.RTPAddress, r.config.Logger.Log, r.config.FrameRate) if err != nil { r.config.Logger.Log(logger.Warning, pkg+"rtp connect error", "error", err.Error()) } mtsSenders = append(mtsSenders, w) case File: w, err := newFileSender(r.config.OutputPath) if err != nil { return err } mtsSenders = append(mtsSenders, w) case RTMP: w, err := newRtmpSender( r.config.RTMPURL, rtmpConnectionTimeout, rtmpConnectionMaxTries, ring.NewBuffer(r.config.RTMPRBSize, r.config.RTMPRBElementSize, time.Duration(r.config.RTMPRBWriteTimeout)*time.Second), r.config.Logger.Log, ) if err != nil { r.config.Logger.Log(logger.Warning, pkg+"rtmp connect error", "error", err.Error()) } flvSenders = append(flvSenders, w) } } // If we have some senders that require MPEGTS encoding then add an MPEGTS // encoder to revid's encoder slice, and give this encoder the mtsSenders // as a destination. if len(mtsSenders) != 0 { mw := multiWriter(mtsSenders...) e, _ := mtsEnc(mw, r.config.WriteRate) encoders = append(encoders, e) } // If we have some senders that require FLV encoding then add an FLV // encoder to revid's encoder slice, and give this encoder the flvSenders // as a destination. if len(flvSenders) != 0 { mw := multiWriter(flvSenders...) e, err := flvEnc(mw, int(r.config.FrameRate)) if err != nil { return err } encoders = append(encoders, e) } r.encoders = multiWriter(encoders...) switch r.config.Input { case Raspivid: r.setupInput = r.startRaspivid switch r.config.InputCodec { case codecutil.H264: r.lexTo = h264.Lex case codecutil.MJPEG: r.lexTo = mjpeg.Lex } case V4L: r.setupInput = r.startV4L r.lexTo = h264.Lex case File: r.setupInput = r.setupInputForFile case RTSP: r.setupInput = r.startRTSPCamera switch r.config.InputCodec { case codecutil.H264: r.lexTo = h264.NewExtractor().Extract case codecutil.H265: r.lexTo = h265.NewLexer(false).Lex case codecutil.MJPEG: panic("not implemented") } case Audio: r.setupInput = r.startAudioDevice r.lexTo = codecutil.NewByteLexer(&r.config.ChunkSize).Lex } return nil } // Start invokes a Revid to start processing video from a defined input // and packetising (if theres packetization) to a defined output. // // Start is safe for concurrent use. func (r *Revid) Start() error { if r.IsRunning() { r.config.Logger.Log(logger.Warning, pkg+"start called, but revid already running") return nil } r.mu.Lock() defer r.mu.Unlock() r.config.Logger.Log(logger.Info, pkg+"starting Revid") err := r.reset(r.config) if err != nil { r.Stop() return err } r.closeInput, err = r.setupInput() if err != nil { return err } r.running = true return nil } // Stop closes down the pipeline. This closes encoders and sender output routines, // connections, and/or files. // // Stop is safe for concurrent use. func (r *Revid) Stop() { if !r.IsRunning() { r.config.Logger.Log(logger.Warning, pkg+"stop called but revid isn't running") return } r.mu.Lock() defer r.mu.Unlock() if r.closeInput != nil { err := r.closeInput() if err != nil { r.config.Logger.Log(logger.Error, pkg+"could not close input", "error", err.Error()) } } r.config.Logger.Log(logger.Info, pkg+"closing pipeline") err := r.encoders.Close() if err != nil { r.config.Logger.Log(logger.Error, pkg+"failed to close pipeline", "error", err.Error()) } r.config.Logger.Log(logger.Info, pkg+"closed pipeline") if r.cmd != nil && r.cmd.Process != nil { r.config.Logger.Log(logger.Info, pkg+"killing input proccess") r.cmd.Process.Kill() } r.config.Logger.Log(logger.Info, pkg+"waiting for routines to close") r.wg.Wait() r.config.Logger.Log(logger.Info, pkg+"revid stopped") r.running = false } func (r *Revid) IsRunning() bool { r.mu.Lock() defer r.mu.Unlock() return r.running } // Update takes a map of variables and their values and edits the current config // if the variables are recognised as valid parameters. // // Update is safe for concurrent use. func (r *Revid) Update(vars map[string]string) error { if r.IsRunning() { r.Stop() } r.mu.Lock() defer r.mu.Unlock() //look through the vars and update revid where needed for key, value := range vars { switch key { case "Input": v, ok := map[string]uint8{"raspivid": Raspivid, "rtsp": RTSP}[strings.ToLower(value)] if !ok { r.config.Logger.Log(logger.Warning, pkg+"invalid input var", "value", value) break } r.config.Input = v case "Saturation": s, err := strconv.ParseInt(value, 10, 0) if err != nil { r.config.Logger.Log(logger.Warning, pkg+"invalid saturation param", "value", value) break } r.config.Saturation = int(s) case "Brightness": b, err := strconv.ParseUint(value, 10, 0) if err != nil { r.config.Logger.Log(logger.Warning, pkg+"invalid brightness param", "value", value) break } r.config.Brightness = uint(b) case "Exposure": r.config.Exposure = value case "AutoWhiteBalance": r.config.AutoWhiteBalance = value case "InputCodec": switch value { case "H264": r.config.InputCodec = codecutil.H264 case "MJPEG": r.config.InputCodec = codecutil.MJPEG default: r.config.Logger.Log(logger.Warning, pkg+"invalid InputCodec variable value", "value", value) } case "Output": outputs := strings.Split(value, ",") r.config.Outputs = make([]uint8, len(outputs)) for i, output := range outputs { switch output { case "File": r.config.Outputs[i] = File case "Http": r.config.Outputs[i] = HTTP case "Rtmp": r.config.Outputs[i] = RTMP case "Rtp": r.config.Outputs[i] = RTP default: r.config.Logger.Log(logger.Warning, pkg+"invalid output param", "value", value) continue } } case "RtmpUrl": r.config.RTMPURL = value case "RtpAddress": r.config.RTPAddress = value case "Bitrate": v, err := strconv.ParseUint(value, 10, 0) if err != nil { r.config.Logger.Log(logger.Warning, pkg+"invalid framerate param", "value", value) break } r.config.Bitrate = uint(v) case "OutputPath": r.config.OutputPath = value case "InputPath": r.config.InputPath = value case "Height": h, err := strconv.ParseUint(value, 10, 0) if err != nil { r.config.Logger.Log(logger.Warning, pkg+"invalid height param", "value", value) break } r.config.Height = uint(h) case "Width": w, err := strconv.ParseUint(value, 10, 0) if err != nil { r.config.Logger.Log(logger.Warning, pkg+"invalid width param", "value", value) break } r.config.Width = uint(w) case "FrameRate": v, err := strconv.ParseUint(value, 10, 0) if err != nil { r.config.Logger.Log(logger.Warning, pkg+"invalid framerate param", "value", value) break } r.config.FrameRate = uint(v) case "Rotation": v, err := strconv.ParseUint(value, 10, 0) if err != nil || v > 359 { r.config.Logger.Log(logger.Warning, pkg+"invalid rotation param", "value", value) break } r.config.Rotation = uint(v) case "HttpAddress": r.config.HTTPAddress = value case "Quantization": v, err := strconv.Atoi(value) if err != nil { r.config.Logger.Log(logger.Warning, pkg+"invalid quantization param", "value", v) break } r.config.Quantization = uint(v) case "MinFrames": v, err := strconv.Atoi(value) if err != nil { r.config.Logger.Log(logger.Warning, pkg+"invalid MinFrames param", "value", value) break } r.config.MinFrames = uint(v) case "ClipDuration": v, err := strconv.Atoi(value) if err != nil { r.config.Logger.Log(logger.Warning, pkg+"invalid ClipDuration param", "value", value) break } r.config.ClipDuration = time.Duration(v) * time.Second case "HorizontalFlip": switch strings.ToLower(value) { case "true": r.config.FlipHorizontal = true case "false": r.config.FlipHorizontal = false default: r.config.Logger.Log(logger.Warning, pkg+"invalid HorizontalFlip param", "value", value) } case "VerticalFlip": switch strings.ToLower(value) { case "true": r.config.FlipVertical = true case "false": r.config.FlipVertical = false default: r.config.Logger.Log(logger.Warning, pkg+"invalid VerticalFlip param", "value", value) } case "BurstPeriod": v, err := strconv.ParseUint(value, 10, 0) if err != nil { r.config.Logger.Log(logger.Warning, pkg+"invalid BurstPeriod param", "value", value) break } r.config.BurstPeriod = uint(v) case "Logging": switch value { case "Debug": r.config.LogLevel = logger.Debug case "Info": r.config.LogLevel = logger.Info case "Warning": r.config.LogLevel = logger.Warning case "Error": r.config.LogLevel = logger.Error case "Fatal": r.config.LogLevel = logger.Fatal default: r.config.Logger.Log(logger.Warning, pkg+"invalid Logging param", "value", value) } case "RTMPRBSize": v, err := strconv.Atoi(value) if err != nil || v < 0 { r.config.Logger.Log(logger.Warning, pkg+"invalid RTMPRBSize var", "value", value) break } r.config.RTMPRBSize = v case "RTMPRBElementSize": v, err := strconv.Atoi(value) if err != nil || v < 0 { r.config.Logger.Log(logger.Warning, pkg+"invalid RTMPRBElementSize var", "value", value) break } r.config.RTMPRBElementSize = v case "RTMPRBWriteTimeout": v, err := strconv.Atoi(value) if err != nil || v <= 0 { r.config.Logger.Log(logger.Warning, pkg+"invalid RTMPRBWriteTimeout var", "value", value) break } r.config.RTMPRBWriteTimeout = v case "MTSRBSize": v, err := strconv.Atoi(value) if err != nil || v < 0 { r.config.Logger.Log(logger.Warning, pkg+"invalid MTSRBSize var", "value", value) break } r.config.MTSRBSize = v case "MTSRBElementSize": v, err := strconv.Atoi(value) if err != nil || v < 0 { r.config.Logger.Log(logger.Warning, pkg+"invalid MTSRBElementSize var", "value", value) break } r.config.MTSRBElementSize = v case "MTSRBWriteTimeout": v, err := strconv.Atoi(value) if err != nil || v <= 0 { r.config.Logger.Log(logger.Warning, pkg+"invalid MTSRBWriteTimeout var", "value", value) break } r.config.MTSRBWriteTimeout = v case "VBR": v, ok := map[string]bool{"true": true, "false": false}[strings.ToLower(value)] if !ok { r.config.Logger.Log(logger.Warning, pkg+"invalid VBR var", "value", value) break } r.config.VBR = v case "VBRQuality": v, ok := map[string]quality{"standard": qualityStandard, "fair": qualityFair, "good": qualityGood, "great": qualityGreat, "excellent": qualityExcellent}[strings.ToLower(value)] if !ok { r.config.Logger.Log(logger.Warning, pkg+"invalid VBRQuality var", "value", value) break } r.config.VBRQuality = v case "VBRBitrate": v, err := strconv.Atoi(value) if err != nil || v <= 0 { r.config.Logger.Log(logger.Warning, pkg+"invalid VBRBitrate var", "value", value) break } r.config.VBRBitrate = v case "CameraChan": v, err := strconv.Atoi(value) if err != nil || (v != 1 && v != 2) { r.config.Logger.Log(logger.Warning, pkg+"invalid CameraChan var", "value", value) break } r.config.CameraChan = v } } r.config.Logger.Log(logger.Info, pkg+"revid config changed", "config", fmt.Sprintf("%+v", r.config)) return nil }