From 8337d87a25d3e437e4c9f4a946079608a19d1094 Mon Sep 17 00:00:00 2001 From: Saxon Date: Mon, 13 May 2019 15:42:16 +0930 Subject: [PATCH 01/27] revid/config.go: cleaned up enums for inputs, outputs and codecs. --- revid/config.go | 165 +++++++++++++++++++++++------------------------- 1 file changed, 79 insertions(+), 86 deletions(-) diff --git a/revid/config.go b/revid/config.go index 010af3fd..d26a72d5 100644 --- a/revid/config.go +++ b/revid/config.go @@ -33,8 +33,85 @@ import ( "bitbucket.org/ausocean/utils/logger" ) +// Possible modes for raspivid --exposure parameter. +var ExposureModes = [...]string{ + "auto", + "night", + "nightpreview", + "backlight", + "spotlight", + "sports", + "snow", + "beach", + "verylong", + "fixedfps", + "antishake", + "fireworks", +} + +// Possible modes for raspivid --awb parameter. +var AutoWhiteBalanceModes = [...]string{ + "off", + "auto", + "sun", + "cloud", + "shade", + "tungsten", + "fluorescent", + "incandescent", + "flash", + "horizon", +} + +// Enums to define inputs, outputs and codecs. +const ( + // Indicates no option has been set. + NothingDefined = iota + + // Input/Output. + File + + // Inputs. + Raspivid + V4L + + // Outputs. + Rtmp + Rtp + Http + Mpegts + + // Codecs. + H264 + Mjpeg +) + +// Default config settings +const ( + defaultInput = Raspivid + defaultOutput = Http + defaultFrameRate = 25 + defaultWidth = 1280 + defaultHeight = 720 + defaultIntraRefreshPeriod = 100 + defaultTimeout = 0 + defaultQuantization = 40 + defaultBitrate = 400000 + defaultFramesPerClip = 1 + httpFramesPerClip = 560 + defaultInputCodec = H264 + defaultVerbosity = logger.Error + defaultRtpAddr = "localhost:6970" + defaultBurstPeriod = 10 // Seconds + defaultRotation = 0 // Degrees + defaultBrightness = 50 + defaultExposure = "auto" + defaultAutoWhiteBalance = "auto" +) + // Config provides parameters relevant to a revid instance. A new config must -// be passed to the constructor. +// be passed to the constructor. Default values for these fields are defined +// as consts at the beginning of this file. type Config struct { LogLevel int8 @@ -76,87 +153,6 @@ type Config struct { AutoWhiteBalance string } -// Possible modes for raspivid --exposure parameter. -var ExposureModes = [...]string{ - "auto", - "night", - "nightpreview", - "backlight", - "spotlight", - "sports", - "snow", - "beach", - "verylong", - "fixedfps", - "antishake", - "fireworks", -} - -// Possible modes for raspivid --awb parameter. -var AutoWhiteBalanceModes = [...]string{ - "off", - "auto", - "sun", - "cloud", - "shade", - "tungsten", - "fluorescent", - "incandescent", - "flash", - "horizon", -} - -// Enums for config struct -const ( - NothingDefined = iota - Raspivid - V4L - H264Codec - File - Http - H264 - Mjpeg - None - Mpegts - Ffmpeg - Flv - LibRtmp - QuantizationOn - QuantizationOff - Yes - No - Rtmp - FfmpegRtmp - Udp - MpegtsRtp - Rtp -) - -// Default config settings -const ( - defaultInput = Raspivid - defaultOutput = Http - defaultPacketization = Flv - defaultFrameRate = 25 - defaultWidth = 1280 - defaultHeight = 720 - defaultIntraRefreshPeriod = 100 - defaultTimeout = 0 - defaultQuantization = 40 - defaultBitrate = 400000 - defaultQuantizationMode = QuantizationOff - defaultFramesPerClip = 1 - httpFramesPerClip = 560 - defaultInputCodec = H264 - defaultVerbosity = logger.Error - defaultRtpAddr = "localhost:6970" - defaultBurstPeriod = 10 // Seconds - defaultRotation = 0 // Degrees - defaultBrightness = 50 - defaultExposure = "auto" - defaultAutoWhiteBalance = "auto" -) - // Validate checks for any errors in the config fields and defaults settings // if particular parameters have not been defined. func (c *Config) Validate(r *Revid) error { @@ -209,13 +205,11 @@ func (c *Config) Validate(r *Revid) error { if c.Outputs == nil { c.Logger.Log(logger.Info, pkg+"no output defined, defaulting", "output", defaultOutput) c.Outputs = append(c.Outputs, defaultOutput) - c.Packetization = defaultPacketization } else { for i, o := range c.Outputs { switch o { case File: - case Udp: - case Rtmp, FfmpegRtmp: + case Rtmp: if c.RtmpUrl == "" { c.Logger.Log(logger.Info, pkg+"no RTMP URL: falling back to HTTP") c.Outputs[i] = Http @@ -225,7 +219,6 @@ func (c *Config) Validate(r *Revid) error { } c.Logger.Log(logger.Info, pkg+"defaulting frames per clip for rtmp out", "framesPerClip", defaultFramesPerClip) c.FramesPerClip = defaultFramesPerClip - c.Packetization = Flv c.SendRetry = true case Http, Rtp: c.Logger.Log(logger.Info, pkg+"defaulting frames per clip for http out", "framesPerClip", httpFramesPerClip) From 001c8696aaeb26d02cce4930cadda762a6b9a5da Mon Sep 17 00:00:00 2001 From: Saxon Date: Mon, 13 May 2019 16:12:08 +0930 Subject: [PATCH 02/27] revid/config.go: started commenting config fields --- revid/config.go | 55 ++++++++++++++++++++++++++++++++++++++----------- 1 file changed, 43 insertions(+), 12 deletions(-) diff --git a/revid/config.go b/revid/config.go index d26a72d5..ff6f198e 100644 --- a/revid/config.go +++ b/revid/config.go @@ -111,27 +111,59 @@ const ( // Config provides parameters relevant to a revid instance. A new config must // be passed to the constructor. Default values for these fields are defined -// as consts at the beginning of this file. +// as consts above. type Config struct { + // LogLevel is the revid logging verbosity level. + // Valid values are defined by enums from the logger package: logger.Debug, + // logger.Info, logger.Warning logger.Error, logger.Fatal. LogLevel int8 - Input uint8 - InputCodec uint8 - Outputs []uint8 - RtmpMethod uint8 - Packetization uint8 + // Input defines the input data source. + // + // Valid values are defined by enums: + // Raspivid: + // Read data from a Raspberry Pi Camera. + // V4l: + // Read from webcam. + // File: + // Location must be specified in InputPath field. + Input uint8 - // Quantize specifies whether the input to - // revid will have constant or variable + // InputCodec defines the input codec we wish to use, and therefore define the + // lexer for use in the pipeline. In most cases this defaults according to a + // particular input. Both Raspivid and V4l use H264, but File input may use + // H264 or MJPEG. + InputCodec uint8 + + // Outputs define the outputs we wish to output data too. + // + // Valid outputs are defined by enums: + // File: + // Location must be defined by the OutputPath field. MPEG-TS packetization + // is used. + // Http: + // Destination is defined by the sh field located in /etc/netsender.conf. + // MPEGT-TS packetization is used. + // Rtmp: + // Destination URL must be defined in the RtmpUrl field. FLV packetization + // is used. + // Rtp: + // Destination is defined by RtpAddr field, otherwise it will default to + // localhost:6970. MPEGT-TS packetization is used. + Outputs []uint8 + + // Quantize specifies whether the input to revid will have constant or variable // bitrate. Quantize bool - // FlipHorizonatla and FlipVertical specify - // whether video frames should be flipped. + // FlipHorizonatla and FlipVertical specify whether video frames should be flipped. FlipHorizontal bool FlipVertical bool - FramesPerClip uint + FramesPerClip uint + + // RtmpUrl specifies the Rtmp output destination URL. This must be defined if + // RTMP is to be used as an output. RtmpUrl string Bitrate uint OutputPath string @@ -223,7 +255,6 @@ func (c *Config) Validate(r *Revid) error { case Http, Rtp: c.Logger.Log(logger.Info, pkg+"defaulting frames per clip for http out", "framesPerClip", httpFramesPerClip) c.FramesPerClip = httpFramesPerClip - c.Packetization = Mpegts default: return errors.New("bad output type defined in config") } From 51fcb18505799b580abca48d112cd74710033528 Mon Sep 17 00:00:00 2001 From: Saxon Date: Mon, 13 May 2019 16:18:41 +0930 Subject: [PATCH 03/27] revid: capitalize exported enums that are acronyms like Rtmp->RTMP, Mpegts->MPEGTS etc. --- cmd/revid-cli/main.go | 17 +++-------------- revid/config.go | 20 ++++++++++---------- revid/revid.go | 16 ++++++++-------- revid/revid_test.go | 12 ++++++------ 4 files changed, 27 insertions(+), 38 deletions(-) diff --git a/cmd/revid-cli/main.go b/cmd/revid-cli/main.go index fc90dbcc..51b3d31f 100644 --- a/cmd/revid-cli/main.go +++ b/cmd/revid-cli/main.go @@ -107,7 +107,6 @@ func handleFlags() revid.Config { inputPtr = flag.String("Input", "", "The input type: Raspivid, File, Webcam") inputCodecPtr = flag.String("InputCodec", "", "The codec of the input: H264, Mjpeg") - rtmpMethodPtr = flag.String("RtmpMethod", "", "The method used to send over rtmp: Ffmpeg, Librtmp") quantizePtr = flag.Bool("Quantize", false, "Quantize input (non-variable bitrate)") verbosityPtr = flag.String("Verbosity", "Info", "Verbosity: Debug, Info, Warning, Error, Fatal") rtpAddrPtr = flag.String("RtpAddr", "", "Rtp destination address: : (port is generally 6970-6999)") @@ -198,27 +197,17 @@ func handleFlags() revid.Config { case "File": cfg.Outputs = append(cfg.Outputs, revid.File) case "Http": - cfg.Outputs = append(cfg.Outputs, revid.Http) + cfg.Outputs = append(cfg.Outputs, revid.HTTP) case "Rtmp": - cfg.Outputs = append(cfg.Outputs, revid.Rtmp) + cfg.Outputs = append(cfg.Outputs, revid.RTMP) case "Rtp": - cfg.Outputs = append(cfg.Outputs, revid.Rtp) + cfg.Outputs = append(cfg.Outputs, revid.RTP) case "": default: log.Log(logger.Error, pkg+"bad output argument", "arg", o) } } - switch *rtmpMethodPtr { - case "Ffmpeg": - cfg.RtmpMethod = revid.Ffmpeg - case "LibRtmp": - cfg.RtmpMethod = revid.LibRtmp - case "": - default: - log.Log(logger.Error, pkg+"bad rtmp method argument") - } - if *configFilePtr != "" { netsender.ConfigFile = *configFilePtr } diff --git a/revid/config.go b/revid/config.go index ff6f198e..b42059f4 100644 --- a/revid/config.go +++ b/revid/config.go @@ -76,20 +76,20 @@ const ( V4L // Outputs. - Rtmp - Rtp - Http - Mpegts + RTMP + RTP + HTTP + MPEGTS // Codecs. H264 - Mjpeg + MJPEG ) // Default config settings const ( defaultInput = Raspivid - defaultOutput = Http + defaultOutput = HTTP defaultFrameRate = 25 defaultWidth = 1280 defaultHeight = 720 @@ -220,7 +220,7 @@ func (c *Config) Validate(r *Revid) error { return errors.New("bad bitrate and quantization combination for H264 input") } - case Mjpeg: + case MJPEG: if c.Quantization > 0 || c.Bitrate == 0 { return errors.New("bad bitrate or quantization for mjpeg input") } @@ -241,10 +241,10 @@ func (c *Config) Validate(r *Revid) error { for i, o := range c.Outputs { switch o { case File: - case Rtmp: + case RTMP: if c.RtmpUrl == "" { c.Logger.Log(logger.Info, pkg+"no RTMP URL: falling back to HTTP") - c.Outputs[i] = Http + c.Outputs[i] = HTTP // FIXME(kortschak): Does this want the same line as below? // c.FramesPerClip = httpFramesPerClip break @@ -252,7 +252,7 @@ func (c *Config) Validate(r *Revid) error { c.Logger.Log(logger.Info, pkg+"defaulting frames per clip for rtmp out", "framesPerClip", defaultFramesPerClip) c.FramesPerClip = defaultFramesPerClip c.SendRetry = true - case Http, Rtp: + case HTTP, RTP: c.Logger.Log(logger.Info, pkg+"defaulting frames per clip for http out", "framesPerClip", httpFramesPerClip) c.FramesPerClip = httpFramesPerClip default: diff --git a/revid/revid.go b/revid/revid.go index 43c4f982..92f8b35f 100644 --- a/revid/revid.go +++ b/revid/revid.go @@ -212,10 +212,10 @@ func (r *Revid) setupPipeline(mtsEnc, flvEnc func(dst io.WriteCloser, rate int) var w io.WriteCloser for _, out := range r.config.Outputs { switch out { - case Http: + case HTTP: w = newMtsSender(newHttpSender(r.ns, r.config.Logger.Log), r.config.Logger.Log, rbSize, rbElementSize, 0) mtsSenders = append(mtsSenders, w) - case Rtp: + 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()) @@ -227,7 +227,7 @@ func (r *Revid) setupPipeline(mtsEnc, flvEnc func(dst io.WriteCloser, rate int) return err } mtsSenders = append(mtsSenders, w) - case Rtmp: + case RTMP: w, err := newRtmpSender(r.config.RtmpUrl, rtmpConnectionTimeout, rtmpConnectionMaxTries, r.config.Logger.Log) if err != nil { r.config.Logger.Log(logger.Warning, pkg+"rtmp connect error", "error", err.Error()) @@ -272,7 +272,7 @@ func (r *Revid) setupPipeline(mtsEnc, flvEnc func(dst io.WriteCloser, rate int) case H264: r.config.Logger.Log(logger.Info, pkg+"using H264 lexer") r.lexTo = lex.H264 - case Mjpeg: + case MJPEG: r.config.Logger.Log(logger.Info, pkg+"using MJPEG lexer") r.lexTo = lex.MJPEG } @@ -366,11 +366,11 @@ func (r *Revid) Update(vars map[string]string) error { case "File": r.config.Outputs[i] = File case "Http": - r.config.Outputs[i] = Http + r.config.Outputs[i] = HTTP case "Rtmp": - r.config.Outputs[i] = Rtmp + r.config.Outputs[i] = RTMP case "Rtp": - r.config.Outputs[i] = Rtp + r.config.Outputs[i] = RTP default: r.config.Logger.Log(logger.Warning, pkg+"invalid output param", "value", value) continue @@ -527,7 +527,7 @@ func (r *Revid) startRaspivid() (func() error, error) { if r.config.Quantize { args = append(args, "-qp", fmt.Sprint(r.config.Quantization)) } - case Mjpeg: + case MJPEG: args = append(args, "--codec", "MJPEG") } r.config.Logger.Log(logger.Info, pkg+"raspivid args", "raspividArgs", strings.Join(args, " ")) diff --git a/revid/revid_test.go b/revid/revid_test.go index 18086912..e17740df 100644 --- a/revid/revid_test.go +++ b/revid/revid_test.go @@ -148,7 +148,7 @@ func TestResetEncoderSenderSetup(t *testing.T) { encoders []encoder }{ { - outputs: []uint8{Http}, + outputs: []uint8{HTTP}, encoders: []encoder{ { encoderType: mtsEncoderStr, @@ -157,7 +157,7 @@ func TestResetEncoderSenderSetup(t *testing.T) { }, }, { - outputs: []uint8{Rtmp}, + outputs: []uint8{RTMP}, encoders: []encoder{ { encoderType: flvEncoderStr, @@ -166,7 +166,7 @@ func TestResetEncoderSenderSetup(t *testing.T) { }, }, { - outputs: []uint8{Rtp}, + outputs: []uint8{RTP}, encoders: []encoder{ { encoderType: mtsEncoderStr, @@ -175,7 +175,7 @@ func TestResetEncoderSenderSetup(t *testing.T) { }, }, { - outputs: []uint8{Http, Rtmp}, + outputs: []uint8{HTTP, RTMP}, encoders: []encoder{ { encoderType: mtsEncoderStr, @@ -188,7 +188,7 @@ func TestResetEncoderSenderSetup(t *testing.T) { }, }, { - outputs: []uint8{Http, Rtp, Rtmp}, + outputs: []uint8{HTTP, RTP, RTMP}, encoders: []encoder{ { encoderType: mtsEncoderStr, @@ -201,7 +201,7 @@ func TestResetEncoderSenderSetup(t *testing.T) { }, }, { - outputs: []uint8{Rtp, Rtmp}, + outputs: []uint8{RTP, RTMP}, encoders: []encoder{ { encoderType: mtsEncoderStr, From 835f97203a030acf153be501d8565f8f9a806342 Mon Sep 17 00:00:00 2001 From: Saxon Date: Mon, 13 May 2019 16:23:38 +0930 Subject: [PATCH 04/27] revid: config fields that are exported and acronyms now capitalized. --- cmd/revid-cli/main.go | 6 +++--- revid/config.go | 18 +++++++++--------- revid/revid.go | 10 +++++----- revid/revid_test.go | 2 +- 4 files changed, 18 insertions(+), 18 deletions(-) diff --git a/cmd/revid-cli/main.go b/cmd/revid-cli/main.go index 51b3d31f..c4729b2a 100644 --- a/cmd/revid-cli/main.go +++ b/cmd/revid-cli/main.go @@ -217,17 +217,17 @@ func handleFlags() revid.Config { cfg.FlipHorizontal = *horizontalFlipPtr cfg.FlipVertical = *verticalFlipPtr cfg.FramesPerClip = *framesPerClipPtr - cfg.RtmpUrl = *rtmpUrlPtr + cfg.RTMPURL = *rtmpUrlPtr cfg.Bitrate = *bitratePtr cfg.OutputPath = *outputPathPtr cfg.InputPath = *inputFilePtr cfg.Height = *heightPtr cfg.Width = *widthPtr cfg.FrameRate = *frameRatePtr - cfg.HttpAddress = *httpAddressPtr + cfg.HTTPAddress = *httpAddressPtr cfg.Quantization = *quantizationPtr cfg.IntraRefreshPeriod = *intraRefreshPeriodPtr - cfg.RtpAddress = *rtpAddrPtr + cfg.RTPAddress = *rtpAddrPtr cfg.SendRetry = *sendRetryPtr cfg.Brightness = *brightnessPtr cfg.Saturation = *saturationPtr diff --git a/revid/config.go b/revid/config.go index b42059f4..b71256a1 100644 --- a/revid/config.go +++ b/revid/config.go @@ -141,13 +141,13 @@ type Config struct { // File: // Location must be defined by the OutputPath field. MPEG-TS packetization // is used. - // Http: + // HTTP: // Destination is defined by the sh field located in /etc/netsender.conf. // MPEGT-TS packetization is used. - // Rtmp: + // RTMP: // Destination URL must be defined in the RtmpUrl field. FLV packetization // is used. - // Rtp: + // RTP: // Destination is defined by RtpAddr field, otherwise it will default to // localhost:6970. MPEGT-TS packetization is used. Outputs []uint8 @@ -164,17 +164,17 @@ type Config struct { // RtmpUrl specifies the Rtmp output destination URL. This must be defined if // RTMP is to be used as an output. - RtmpUrl string + RTMPURL string Bitrate uint OutputPath string InputPath string Height uint Width uint FrameRate uint - HttpAddress string + HTTPAddress string Quantization uint IntraRefreshPeriod uint - RtpAddress string + RTPAddress string Logger Logger SendRetry bool BurstPeriod uint @@ -242,7 +242,7 @@ func (c *Config) Validate(r *Revid) error { switch o { case File: case RTMP: - if c.RtmpUrl == "" { + if c.RTMPURL == "" { c.Logger.Log(logger.Info, pkg+"no RTMP URL: falling back to HTTP") c.Outputs[i] = HTTP // FIXME(kortschak): Does this want the same line as below? @@ -308,8 +308,8 @@ func (c *Config) Validate(r *Revid) error { return errors.New("quantisation is over threshold") } - if c.RtpAddress == "" { - c.RtpAddress = defaultRtpAddr + if c.RTPAddress == "" { + c.RTPAddress = defaultRtpAddr } switch { diff --git a/revid/revid.go b/revid/revid.go index 92f8b35f..88657a2d 100644 --- a/revid/revid.go +++ b/revid/revid.go @@ -216,7 +216,7 @@ func (r *Revid) setupPipeline(mtsEnc, flvEnc func(dst io.WriteCloser, rate int) w = newMtsSender(newHttpSender(r.ns, r.config.Logger.Log), r.config.Logger.Log, rbSize, rbElementSize, 0) mtsSenders = append(mtsSenders, w) case RTP: - w, err := newRtpSender(r.config.RtpAddress, r.config.Logger.Log, r.config.FrameRate) + 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()) } @@ -228,7 +228,7 @@ func (r *Revid) setupPipeline(mtsEnc, flvEnc func(dst io.WriteCloser, rate int) } mtsSenders = append(mtsSenders, w) case RTMP: - w, err := newRtmpSender(r.config.RtmpUrl, rtmpConnectionTimeout, rtmpConnectionMaxTries, r.config.Logger.Log) + w, err := newRtmpSender(r.config.RTMPURL, rtmpConnectionTimeout, rtmpConnectionMaxTries, r.config.Logger.Log) if err != nil { r.config.Logger.Log(logger.Warning, pkg+"rtmp connect error", "error", err.Error()) } @@ -378,9 +378,9 @@ func (r *Revid) Update(vars map[string]string) error { } case "RtmpUrl": - r.config.RtmpUrl = value + r.config.RTMPURL = value case "RtpAddress": - r.config.RtpAddress = value + r.config.RTPAddress = value case "Bitrate": v, err := strconv.ParseUint(value, 10, 0) if err != nil { @@ -421,7 +421,7 @@ func (r *Revid) Update(vars map[string]string) error { } r.config.Rotation = uint(v) case "HttpAddress": - r.config.HttpAddress = value + r.config.HTTPAddress = value case "Quantization": q, err := strconv.ParseUint(value, 10, 0) if err != nil { diff --git a/revid/revid_test.go b/revid/revid_test.go index e17740df..36cc913d 100644 --- a/revid/revid_test.go +++ b/revid/revid_test.go @@ -224,7 +224,7 @@ func TestResetEncoderSenderSetup(t *testing.T) { for testNum, test := range tests { // Create a new config and reset revid with it. const dummyURL = "rtmp://dummy" - c := Config{Logger: &testLogger{}, Outputs: test.outputs, RtmpUrl: dummyURL} + c := Config{Logger: &testLogger{}, Outputs: test.outputs, RTMPURL: dummyURL} err := rv.setConfig(c) if err != nil { t.Fatalf("unexpected error: %v for test %v", err, testNum) From 1762adf3383b332d16e573359df3415459304fcf Mon Sep 17 00:00:00 2001 From: Saxon Date: Mon, 13 May 2019 16:55:20 +0930 Subject: [PATCH 05/27] revid/config.go: finished commenting config fields, and removed unused options. --- cmd/revid-cli/main.go | 2 -- revid/config.go | 82 ++++++++++++++++++++++++++++++------------- 2 files changed, 57 insertions(+), 27 deletions(-) diff --git a/cmd/revid-cli/main.go b/cmd/revid-cli/main.go index c4729b2a..98804a5b 100644 --- a/cmd/revid-cli/main.go +++ b/cmd/revid-cli/main.go @@ -116,7 +116,6 @@ func handleFlags() revid.Config { outputPathPtr = flag.String("OutputPath", "", "The directory of the output file") inputFilePtr = flag.String("InputPath", "", "The directory of the input file") httpAddressPtr = flag.String("HttpAddress", "", "Destination address of http posts") - sendRetryPtr = flag.Bool("retry", false, "Specify whether a failed send should be retried.") verticalFlipPtr = flag.Bool("VerticalFlip", false, "Flip video vertically: Yes, No") horizontalFlipPtr = flag.Bool("HorizontalFlip", false, "Flip video horizontally: Yes, No") framesPerClipPtr = flag.Uint("FramesPerClip", 0, "Number of frames per clip sent") @@ -228,7 +227,6 @@ func handleFlags() revid.Config { cfg.Quantization = *quantizationPtr cfg.IntraRefreshPeriod = *intraRefreshPeriodPtr cfg.RTPAddress = *rtpAddrPtr - cfg.SendRetry = *sendRetryPtr cfg.Brightness = *brightnessPtr cfg.Saturation = *saturationPtr cfg.Exposure = *exposurePtr diff --git a/revid/config.go b/revid/config.go index b71256a1..b33eb7ad 100644 --- a/revid/config.go +++ b/revid/config.go @@ -153,36 +153,69 @@ type Config struct { Outputs []uint8 // Quantize specifies whether the input to revid will have constant or variable - // bitrate. + // bitrate, if configurable with the chosen input. Raspivid supports quantization. Quantize bool - // FlipHorizonatla and FlipVertical specify whether video frames should be flipped. - FlipHorizontal bool - FlipVertical bool - + // FramesPerClip defines the number of packetization units to pack into a clip + // per HTTP send. FramesPerClip uint - // RtmpUrl specifies the Rtmp output destination URL. This must be defined if + // RTMPURL specifies the Rtmp output destination URL. This must be defined if // RTMP is to be used as an output. - RTMPURL string - Bitrate uint - OutputPath string - InputPath string - Height uint - Width uint - FrameRate uint - HTTPAddress string - Quantization uint + RTMPURL string + + // OutputPath defines the output destination for File output. This must be + // defined if File output is to be used. + OutputPath string + + // InputPath defines the input file location for File Input. This must be + // defined if File input is to be used. + InputPath string + + // FrameRate defines the input frame rate if configurable by the chosen input. + // Raspivid input supports custom framerate. + FrameRate uint + + // HTTPAddress defines a custom HTTP destination if we do not wish to use that + // defined in /etc/netsender.conf. + HTTPAddress string + + // Quantization defines the quantization level, which may be a value between + // 0-40. This will only take effect if the Quantize field is true and if we + // are using Raspivid input. + Quantization uint + + // IntraRefreshPeriod defines the frequency of video parameter NAL units for + // Raspivid input. IntraRefreshPeriod uint - RTPAddress string - Logger Logger - SendRetry bool - BurstPeriod uint - Rotation uint - Brightness uint - Saturation int - Exposure string - AutoWhiteBalance string + + // Logger holds an implementation of the Logger interface as defined in revid.go. + // This must be set for revid to work correctly. + Logger Logger + + // Brightness and saturation define the brightness and saturation levels for + // Raspivid input. + Brightness uint + Saturation int + + // Exposure defines the exposure mode used by the Raspivid input. Valid modes + // are defined in the exported []string ExposureModes defined at the start + // of the file. + Exposure string + + // AutoWhiteBalance defines the auto white balance mode used by Raspivid input. + // Valid modes are defined in the exported []string AutoWhiteBalanceModes + // defined at the start of the file. + AutoWhiteBalance string + + RTPAddress string // RTPAddress defines the RTP output destination. + BurstPeriod uint // BurstPeriod defines the revid burst period in seconds. + Rotation uint // Rotation defines the video rotation angle in degrees Raspivid input. + Height uint // Height defines the input video height Raspivid input. + Width uint // Width defines the input video width Raspivid input. + Bitrate uint // Bitrate specifies the input bitrate for Raspivid input. + FlipHorizontal bool // FlipHorizontal flips video horizontally for Raspivid input. + FlipVertical bool // FlipVertial flips video vertically for Raspivid input. } // Validate checks for any errors in the config fields and defaults settings @@ -251,7 +284,6 @@ func (c *Config) Validate(r *Revid) error { } c.Logger.Log(logger.Info, pkg+"defaulting frames per clip for rtmp out", "framesPerClip", defaultFramesPerClip) c.FramesPerClip = defaultFramesPerClip - c.SendRetry = true case HTTP, RTP: c.Logger.Log(logger.Info, pkg+"defaulting frames per clip for http out", "framesPerClip", httpFramesPerClip) c.FramesPerClip = httpFramesPerClip From 819b4a122f86464acd3592e5ba6e103d2db74dcf Mon Sep 17 00:00:00 2001 From: Saxon Date: Sun, 19 May 2019 21:44:23 +0930 Subject: [PATCH 06/27] codec/h264: renamed Lex to LexFromBytestream --- codec/h264/lex.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/codec/h264/lex.go b/codec/h264/lex.go index 3976cef1..d7e26876 100644 --- a/codec/h264/lex.go +++ b/codec/h264/lex.go @@ -42,11 +42,11 @@ func init() { var h264Prefix = [...]byte{0x00, 0x00, 0x01, 0x09, 0xf0} -// Lex lexes H.264 NAL units read from src into separate writes to dst with -// successive writes being performed not earlier than the specified delay. -// NAL units are split after type 1 (Coded slice of a non-IDR picture), 5 +// LexFromBytestream lexes H.264 NAL units read from src into separate writes +// to dst with successive writes being performed not earlier than the specified +// delay. NAL units are split after type 1 (Coded slice of a non-IDR picture), 5 // (Coded slice of a IDR picture) and 8 (Picture parameter set). -func Lex(dst io.Writer, src io.Reader, delay time.Duration) error { +func LexFromBytestream(dst io.Writer, src io.Reader, delay time.Duration) error { var tick <-chan time.Time if delay == 0 { tick = noDelay From ceb15e53c30165072508ee0611b800d2bae1d682 Mon Sep 17 00:00:00 2001 From: Saxon Date: Wed, 22 May 2019 12:23:29 +0930 Subject: [PATCH 07/27] codec/h264/lex.go: wrote lexer for lexing h264 access units from RTP stream --- codec/h264/lex.go | 167 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 167 insertions(+) diff --git a/codec/h264/lex.go b/codec/h264/lex.go index e2e20b05..14e0f63a 100644 --- a/codec/h264/lex.go +++ b/codec/h264/lex.go @@ -30,10 +30,14 @@ LICENSE package h264 import ( + "bytes" + "encoding/binary" + "fmt" "io" "time" "bitbucket.org/ausocean/av/codec/codecutil" + "bitbucket.org/ausocean/av/protocol/rtp" ) var noDelay = make(chan time.Time) @@ -133,3 +137,166 @@ outer: _, err := dst.Write(buf) return err } + +const ( + // Single nal units bounds. + typeSingleNALULowBound = 1 + typeSingleNALUHighBound = 23 + + // Single-time aggregation packets. + typeSTAPA = 24 + typeSTAPB = 25 + + // Multi-time aggregation packets. + typeMTAP16 = 26 + typeMTAP24 = 27 + + // Fragmentation packets. + typeFUA = 28 + typeFUB = 29 +) + +// Buffer sizes. +const ( + maxAUSize = 100000 + maxRTPSize = 4096 +) + +// RTPLexer is a lexer for lexing H265 from RTP packets. +type RTPLexer struct { + buf *bytes.Buffer // Holds the current access unit. + frag bool // Indicates if we're currently dealing with a fragmentation packet. +} + +// NewRTPLexer returns a new RTPLexer. +func NewRTPLexer() *RTPLexer { + return &RTPLexer{ + buf: bytes.NewBuffer(make([]byte, 0, maxAUSize))} +} + +// LexFromRTP extracts H264 access units from an RTP stream. This function +// expects that each read from src will provide a single RTP packet. +func (l *RTPLexer) LexFromRTP(dst io.Writer, src io.Reader, delay time.Duration) error { + buf := make([]byte, maxRTPSize) + for { + n, err := src.Read(buf) + switch err { + case nil: // Do nothing. + case io.EOF: + return nil + default: + return fmt.Errorf("source read error: %v\n", err) + } + + // Get payload from RTP packet. + payload, err := rtp.Payload(buf[:n]) + if err != nil { + return fmt.Errorf("could not get RTP payload, failed with err: %v\n", err) + } + + nalType := payload[0] & 0x1f + + // If not currently fragmented then we ignore current write. + if l.frag && nalType != typeFUA { + l.buf.Reset() + l.frag = false + continue + } + + if nalType >= typeSingleNALULowBound && nalType <= typeSingleNALUHighBound { + l.writeWithPrefix(payload) + } else { + switch nalType { + case typeSTAPA: + l.handleSTAPA(payload) + case typeFUA: + l.handleFUA(payload) + case typeSTAPB: + panic("STAP-B type unsupported") + case typeMTAP16: + panic("MTAP16 type unsupported") + case typeMTAP24: + panic("MTAP24 type unsupported") + case typeFUB: + panic("FU-B type unsupported") + default: + panic("unsupported type") + } + } + + markerIsSet, err := rtp.Marker(buf[:n]) + if err != nil { + return fmt.Errorf("could not get marker bit, failed with err: %v\n", err) + } + + if markerIsSet { + l.buf.WriteTo(dst) + l.buf.Reset() + } + } + return nil +} + +// handleSTAP parses NAL units from an aggregation packet and writes +// them to the Lexers buffer buf. +func (l *RTPLexer) handleSTAPA(d []byte) { + for i := 1; i < len(d); { + size := int(binary.BigEndian.Uint16(d[i:])) + + // Skip over NAL unit size. + const sizeOfFieldLen = 2 + i += sizeOfFieldLen + + // Get the NALU. + nalu := d[i : i+size] + i += size + l.writeWithPrefix(nalu) + } +} + +// handleFragmentation parses NAL units from fragmentation packets and writes +// them to the Lexer's buf. +func (l *RTPLexer) handleFUA(d []byte) { + // Get start and end indiciators from FU header. + const FUHeadIdx = 1 + start := d[FUHeadIdx]&0x80 != 0 + end := d[FUHeadIdx]&0x40 != 0 + + // If start, form new header, skip FU indicator only and set first byte to + // new header. Otherwise, skip over both FU indicator and FU header. + if start { + const FUIndicatorIdx = 0 + newHead := (d[FUIndicatorIdx] & 0xe0) | (d[FUHeadIdx] & 0x1f) + d = d[1:] + d[0] = newHead + } else { + d = d[2:] + } + + switch { + case start && !end: + l.frag = true + l.writeWithPrefix(d) + case !start && end: + l.frag = false + fallthrough + case !start && !end: + l.writeNoPrefix(d) + default: + panic("bad fragmentation packet") + } +} + +// write writes a NAL unit to the Lexer's buf in byte stream format using the +// start code. +func (l *RTPLexer) writeWithPrefix(d []byte) { + const prefix = "\x00\x00\x00\x01" + l.buf.Write([]byte(prefix)) + l.buf.Write(d) +} + +// writeNoPrefix writes data to the Lexer's buf. This is used for non start +// fragmentations of a NALU. +func (l *RTPLexer) writeNoPrefix(d []byte) { + l.buf.Write(d) +} From 722ddc6ccb04b82c181ae24931969ce30943049d Mon Sep 17 00:00:00 2001 From: Saxon Date: Wed, 22 May 2019 14:49:25 +0930 Subject: [PATCH 08/27] codec/h264/lex_test.go: added test for RTPLexer.Lex(...) and also fixed some build errors --- codec/h264/lex.go | 39 +++++----- codec/h264/lex_test.go | 143 +++++++++++++++++++++++++++++++++++++ protocol/rtmp/rtmp_test.go | 4 +- revid/revid.go | 2 +- 4 files changed, 166 insertions(+), 22 deletions(-) diff --git a/codec/h264/lex.go b/codec/h264/lex.go index 14e0f63a..5f100c82 100644 --- a/codec/h264/lex.go +++ b/codec/h264/lex.go @@ -40,6 +40,25 @@ import ( "bitbucket.org/ausocean/av/protocol/rtp" ) +// NAL types. +const ( + // Single nal units bounds. + typeSingleNALULowBound = 1 + typeSingleNALUHighBound = 23 + + // Single-time aggregation packets. + typeSTAPA = 24 + typeSTAPB = 25 + + // Multi-time aggregation packets. + typeMTAP16 = 26 + typeMTAP24 = 27 + + // Fragmentation packets. + typeFUA = 28 + typeFUB = 29 +) + var noDelay = make(chan time.Time) func init() { @@ -138,24 +157,6 @@ outer: return err } -const ( - // Single nal units bounds. - typeSingleNALULowBound = 1 - typeSingleNALUHighBound = 23 - - // Single-time aggregation packets. - typeSTAPA = 24 - typeSTAPB = 25 - - // Multi-time aggregation packets. - typeMTAP16 = 26 - typeMTAP24 = 27 - - // Fragmentation packets. - typeFUA = 28 - typeFUB = 29 -) - // Buffer sizes. const ( maxAUSize = 100000 @@ -176,7 +177,7 @@ func NewRTPLexer() *RTPLexer { // LexFromRTP extracts H264 access units from an RTP stream. This function // expects that each read from src will provide a single RTP packet. -func (l *RTPLexer) LexFromRTP(dst io.Writer, src io.Reader, delay time.Duration) error { +func (l *RTPLexer) Lex(dst io.Writer, src io.Reader, delay time.Duration) error { buf := make([]byte, maxRTPSize) for { n, err := src.Read(buf) diff --git a/codec/h264/lex_test.go b/codec/h264/lex_test.go index d2eeae2a..3632268b 100644 --- a/codec/h264/lex_test.go +++ b/codec/h264/lex_test.go @@ -7,6 +7,7 @@ DESCRIPTION AUTHOR Dan Kortschak + Saxon A. Nelson-Milton LICENSE lex_test.go is Copyright (C) 2017 the Australian Ocean Lab (AusOcean) @@ -30,6 +31,8 @@ LICENSE package h264 import ( + "io" + "testing" "time" ) @@ -220,3 +223,143 @@ func TestH264(t *testing.T) { } } */ + +// rtpReader provides the RTP stream. +type rtpReader struct { + packets [][]byte + idx int +} + +// Read implements io.Reader. +func (r *rtpReader) Read(p []byte) (int, error) { + if r.idx == len(r.packets) { + return 0, io.EOF + } + b := r.packets[r.idx] + n := copy(p, b) + if n < len(r.packets[r.idx]) { + r.packets[r.idx] = r.packets[r.idx][n:] + } else { + r.idx++ + } + return n, nil +} + +// destination holds the access units extracted during the lexing process. +type destination [][]byte + +// Write implements io.Writer. +func (d *destination) Write(p []byte) (int, error) { + t := make([]byte, len(p)) + copy(t, p) + *d = append([][]byte(*d), t) + return len(p), nil +} + +// TestLex checks that the Lexer can correctly extract H264 access units from +// h264 RTP stream in RTP payload format. +func TestLex(t *testing.T) { + const rtpVer = 2 + + tests := []struct { + packets [][]byte + expect [][]byte + }{ + { + packets: [][]byte{ + { // Single NAL unit. + 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // RTP header. + typeSingleNALULowBound, // NAL header. + 0x01, 0x02, 0x03, 0x04, // NAL Data. + }, + { // Fragmentation (start packet). + 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // RTP header. + typeFUA, // FU indicator. + 0x80 | typeSingleNALULowBound, // FU header. + 0x01, 0x02, 0x03, // FU payload. + }, + { // Fragmentation (middle packet) + 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // RTP header. + typeFUA, // NAL indicator. + typeSingleNALULowBound, // FU header. + 0x04, 0x05, 0x06, // FU payload. + }, + { // Fragmentation (end packet) + 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // RTP header. + typeFUA, // NAL indicator. + 0x40 | typeSingleNALULowBound, // FU header. + 0x07, 0x08, 0x09, // FU payload + }, + + { // Aggregation. Make last packet of access unit => marker bit true. + 0x80, 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // RTP header. + typeSTAPA, // NAL header. + 0x00, 0x04, // NAL 1 size. + 0x01, 0x02, 0x03, 0x04, // NAL 1 data. + 0x00, 0x04, // NAL 2 size. + 0x01, 0x02, 0x03, 0x04, // NAL 2 data. + }, + // Second access unit. + { // Single NAL unit. + 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // RTP header. + typeSingleNALULowBound, // NAL header. + 0x01, 0x02, 0x03, 0x04, // NAL Data. + }, + { // Single NAL. Make last packet of access unit => marker bit true. + 0x80, 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // RTP header. + typeSingleNALULowBound, // NAL header. + 0x01, 0x02, 0x03, 0x04, // NAL data. + }, + }, + expect: [][]byte{ + // First access unit. + { + // NAL 1 + 0x00, 0x00, 0x00, 0x01, // Start code. + typeSingleNALULowBound, // NAL header. + 0x01, 0x02, 0x03, 0x04, // NAL data. + // NAL 2 + 0x00, 0x00, 0x00, 0x01, // Start code. + typeSingleNALULowBound, + 0x01, 0x02, 0x03, // FU payload. + 0x04, 0x05, 0x06, // FU payload. + 0x07, 0x08, 0x09, // FU payload. + // NAL 3 + 0x00, 0x00, 0x00, 0x01, // Start code. + 0x01, 0x02, 0x03, 0x04, // NAL data. + // NAL 4 + 0x00, 0x00, 0x00, 0x01, // Start code. + 0x01, 0x02, 0x03, 0x04, // NAL 2 data + }, + // Second access unit. + { + // NAL 1 + 0x00, 0x00, 0x00, 0x01, // Start code. + typeSingleNALULowBound, // NAL header. + 0x01, 0x02, 0x03, 0x04, // Data. + // NAL 2 + 0x00, 0x00, 0x00, 0x01, // Start code. + typeSingleNALULowBound, // NAL header. + 0x01, 0x02, 0x03, 0x04, // Data. + }, + }, + }, + } + + for testNum, test := range tests { + r := &rtpReader{packets: test.packets} + d := &destination{} + err := NewRTPLexer().Lex(d, r, 0) + if err != nil { + t.Fatalf("error lexing: %v\n", err) + } + + for i, accessUnit := range test.expect { + for j, part := range accessUnit { + if part != [][]byte(*d)[i][j] { + t.Fatalf("did not get expected data for test: %v.\nGot: %v\nWant: %v\n", testNum, d, test.expect) + } + } + } + } +} diff --git a/protocol/rtmp/rtmp_test.go b/protocol/rtmp/rtmp_test.go index e1e79796..c7c38bd4 100644 --- a/protocol/rtmp/rtmp_test.go +++ b/protocol/rtmp/rtmp_test.go @@ -199,7 +199,7 @@ func TestFromFrame(t *testing.T) { if err != nil { t.Errorf("Failed to create flv encoder with error: %v", err) } - err = h264.Lex(flvEncoder, bytes.NewReader(videoData), time.Second/time.Duration(frameRate)) + err = h264.LexFromBytestream(flvEncoder, bytes.NewReader(videoData), time.Second/time.Duration(frameRate)) if err != nil { t.Errorf("Lexing failed with error: %v", err) } @@ -251,7 +251,7 @@ func TestFromFile(t *testing.T) { if err != nil { t.Fatalf("failed to create encoder: %v", err) } - err = h264.Lex(flvEncoder, f, time.Second/time.Duration(25)) + err = h264.LexFromBytestream(flvEncoder, f, time.Second/time.Duration(25)) if err != nil { t.Errorf("Lexing and encoding failed with error: %v", err) } diff --git a/revid/revid.go b/revid/revid.go index 911d095f..c5d92c5a 100644 --- a/revid/revid.go +++ b/revid/revid.go @@ -272,7 +272,7 @@ func (r *Revid) setupPipeline(mtsEnc, flvEnc func(dst io.WriteCloser, rate int) switch r.config.InputCodec { case H264: r.config.Logger.Log(logger.Info, pkg+"using H264 lexer") - r.lexTo = h264.Lex + r.lexTo = h264.LexFromBytestream case Mjpeg: r.config.Logger.Log(logger.Info, pkg+"using MJPEG lexer") r.lexTo = mjpeg.Lex From 0e7504374e2ab3c432b9a6f56a64dedb22cf6268 Mon Sep 17 00:00:00 2001 From: Saxon Date: Wed, 22 May 2019 15:09:54 +0930 Subject: [PATCH 09/27] codec/h264: fixed some minor naming and comment spelling --- codec/h264/lex.go | 4 ++-- codec/h264/lex_test.go | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/codec/h264/lex.go b/codec/h264/lex.go index 5f100c82..0883c160 100644 --- a/codec/h264/lex.go +++ b/codec/h264/lex.go @@ -163,13 +163,13 @@ const ( maxRTPSize = 4096 ) -// RTPLexer is a lexer for lexing H265 from RTP packets. +// RTPLexer is a lexer for lexing H264 from RTP packets. type RTPLexer struct { buf *bytes.Buffer // Holds the current access unit. frag bool // Indicates if we're currently dealing with a fragmentation packet. } -// NewRTPLexer returns a new RTPLexer. +// NewRTPLexer returns a new RTPLexer. func NewRTPLexer() *RTPLexer { return &RTPLexer{ buf: bytes.NewBuffer(make([]byte, 0, maxAUSize))} diff --git a/codec/h264/lex_test.go b/codec/h264/lex_test.go index 3632268b..63150968 100644 --- a/codec/h264/lex_test.go +++ b/codec/h264/lex_test.go @@ -258,7 +258,7 @@ func (d *destination) Write(p []byte) (int, error) { // TestLex checks that the Lexer can correctly extract H264 access units from // h264 RTP stream in RTP payload format. -func TestLex(t *testing.T) { +func TestRTPLex(t *testing.T) { const rtpVer = 2 tests := []struct { From 35069bd4f3071d76785a7f2184c780e64148ca19 Mon Sep 17 00:00:00 2001 From: Saxon Date: Thu, 23 May 2019 13:55:25 +0930 Subject: [PATCH 10/27] codec/h264/lex.go: using if and else statements rather than switch for start and end indicator logic in handleFUA --- codec/h264/lex.go | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/codec/h264/lex.go b/codec/h264/lex.go index 0883c160..a5b94bb8 100644 --- a/codec/h264/lex.go +++ b/codec/h264/lex.go @@ -274,17 +274,17 @@ func (l *RTPLexer) handleFUA(d []byte) { d = d[2:] } - switch { - case start && !end: + if start { + if end { + panic("bad fragmentation packet") + } l.frag = true l.writeWithPrefix(d) - case !start && end: - l.frag = false - fallthrough - case !start && !end: + } else { + if end { + l.frag = false + } l.writeNoPrefix(d) - default: - panic("bad fragmentation packet") } } From b49e65d928a192d859364f250a7e1e918e332578 Mon Sep 17 00:00:00 2001 From: Saxon Date: Fri, 24 May 2019 10:33:34 +0930 Subject: [PATCH 11/27] codec/h264/lex.go: commented buffer size consts --- codec/h264/lex.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/codec/h264/lex.go b/codec/h264/lex.go index a5b94bb8..cde1e2e0 100644 --- a/codec/h264/lex.go +++ b/codec/h264/lex.go @@ -159,8 +159,8 @@ outer: // Buffer sizes. const ( - maxAUSize = 100000 - maxRTPSize = 4096 + maxAUSize = 100000 // Max access unit size in bytes. + maxRTPSize = 1500 // Max ethernet transmission unit in bytes. ) // RTPLexer is a lexer for lexing H264 from RTP packets. From e1fd6837ef29b252805e08f5b5c16f807c411445 Mon Sep 17 00:00:00 2001 From: Saxon Date: Fri, 24 May 2019 10:34:52 +0930 Subject: [PATCH 12/27] codec/h264/lex.go: updated comment for RTPLexer.Lex() --- codec/h264/lex.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/codec/h264/lex.go b/codec/h264/lex.go index cde1e2e0..ce3ae14c 100644 --- a/codec/h264/lex.go +++ b/codec/h264/lex.go @@ -175,7 +175,7 @@ func NewRTPLexer() *RTPLexer { buf: bytes.NewBuffer(make([]byte, 0, maxAUSize))} } -// LexFromRTP extracts H264 access units from an RTP stream. This function +// Lex extracts H264 access units from an RTP stream. This function // expects that each read from src will provide a single RTP packet. func (l *RTPLexer) Lex(dst io.Writer, src io.Reader, delay time.Duration) error { buf := make([]byte, maxRTPSize) From 930c35a154fb6c2cb15a35cf6e99d18e706cefe1 Mon Sep 17 00:00:00 2001 From: Saxon Date: Fri, 24 May 2019 10:36:00 +0930 Subject: [PATCH 13/27] codec/h264/lex.go: udpated comment for RTPLexer.handleSTAPA --- codec/h264/lex.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/codec/h264/lex.go b/codec/h264/lex.go index ce3ae14c..ef1de277 100644 --- a/codec/h264/lex.go +++ b/codec/h264/lex.go @@ -238,7 +238,7 @@ func (l *RTPLexer) Lex(dst io.Writer, src io.Reader, delay time.Duration) error return nil } -// handleSTAP parses NAL units from an aggregation packet and writes +// handleSTAPA parses NAL units from an aggregation packet and writes // them to the Lexers buffer buf. func (l *RTPLexer) handleSTAPA(d []byte) { for i := 1; i < len(d); { From 04cd2e66ff8f13ba72455968ee166794e242098d Mon Sep 17 00:00:00 2001 From: Saxon Date: Fri, 24 May 2019 10:36:30 +0930 Subject: [PATCH 14/27] codec/h264/lex.go: updated comment for RTPLexer.handleFUA() --- codec/h264/lex.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/codec/h264/lex.go b/codec/h264/lex.go index ef1de277..7838a04d 100644 --- a/codec/h264/lex.go +++ b/codec/h264/lex.go @@ -255,7 +255,7 @@ func (l *RTPLexer) handleSTAPA(d []byte) { } } -// handleFragmentation parses NAL units from fragmentation packets and writes +// handleFUA parses NAL units from fragmentation packets and writes // them to the Lexer's buf. func (l *RTPLexer) handleFUA(d []byte) { // Get start and end indiciators from FU header. From 5dfff618879195a55f57e66ab2e5a326644b9c7a Mon Sep 17 00:00:00 2001 From: Saxon Date: Fri, 24 May 2019 10:38:06 +0930 Subject: [PATCH 15/27] codec/h264/lex.go: merge logic for start end packet writing into first if --- codec/h264/lex.go | 12 ++++-------- 1 file changed, 4 insertions(+), 8 deletions(-) diff --git a/codec/h264/lex.go b/codec/h264/lex.go index 7838a04d..e9cbe67c 100644 --- a/codec/h264/lex.go +++ b/codec/h264/lex.go @@ -266,24 +266,20 @@ func (l *RTPLexer) handleFUA(d []byte) { // If start, form new header, skip FU indicator only and set first byte to // new header. Otherwise, skip over both FU indicator and FU header. if start { + if end { + panic("bad fragmentation packet") + } const FUIndicatorIdx = 0 newHead := (d[FUIndicatorIdx] & 0xe0) | (d[FUHeadIdx] & 0x1f) d = d[1:] d[0] = newHead - } else { - d = d[2:] - } - - if start { - if end { - panic("bad fragmentation packet") - } l.frag = true l.writeWithPrefix(d) } else { if end { l.frag = false } + d = d[2:] l.writeNoPrefix(d) } } From e1581011700f4eb2181d37197bbba00f94a1dfe4 Mon Sep 17 00:00:00 2001 From: Saxon Date: Fri, 24 May 2019 10:38:54 +0930 Subject: [PATCH 16/27] codec/h264/lex_test.go: fixed Saxon Nelson-Milton author indentation in file header --- codec/h264/lex_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/codec/h264/lex_test.go b/codec/h264/lex_test.go index 63150968..4f5a7af3 100644 --- a/codec/h264/lex_test.go +++ b/codec/h264/lex_test.go @@ -7,7 +7,7 @@ DESCRIPTION AUTHOR Dan Kortschak - Saxon A. Nelson-Milton + Saxon A. Nelson-Milton LICENSE lex_test.go is Copyright (C) 2017 the Australian Ocean Lab (AusOcean) From 41f442437ddbf0d2c1526754c8e6c0de17f08d40 Mon Sep 17 00:00:00 2001 From: Saxon Date: Fri, 24 May 2019 10:40:17 +0930 Subject: [PATCH 17/27] codec/h264/lex_test.go: improve comment for rtpReader struct --- codec/h264/lex_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/codec/h264/lex_test.go b/codec/h264/lex_test.go index 4f5a7af3..722379e4 100644 --- a/codec/h264/lex_test.go +++ b/codec/h264/lex_test.go @@ -224,7 +224,7 @@ func TestH264(t *testing.T) { } */ -// rtpReader provides the RTP stream. +// rtpReader provides an io.Reader for reading the test RTP stream. type rtpReader struct { packets [][]byte idx int From 5b970823efa7874654d28fde5435622b1348310e Mon Sep 17 00:00:00 2001 From: Saxon Date: Fri, 24 May 2019 10:41:18 +0930 Subject: [PATCH 18/27] codec/h264/lex_test.go: renamed t to tmp in destination.Write() --- codec/h264/lex_test.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/codec/h264/lex_test.go b/codec/h264/lex_test.go index 722379e4..7e3422ea 100644 --- a/codec/h264/lex_test.go +++ b/codec/h264/lex_test.go @@ -250,9 +250,9 @@ type destination [][]byte // Write implements io.Writer. func (d *destination) Write(p []byte) (int, error) { - t := make([]byte, len(p)) - copy(t, p) - *d = append([][]byte(*d), t) + tmp := make([]byte, len(p)) + copy(tmp, p) + *d = append([][]byte(*d), tmp) return len(p), nil } From a39cd31a5735096a0fa99fc71f166d9099f14c68 Mon Sep 17 00:00:00 2001 From: Saxon Date: Mon, 27 May 2019 14:07:04 +0930 Subject: [PATCH 19/27] codec/h264/lex.go: added spec link for nal type consts --- codec/h264/lex.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/codec/h264/lex.go b/codec/h264/lex.go index e9cbe67c..81eca759 100644 --- a/codec/h264/lex.go +++ b/codec/h264/lex.go @@ -40,7 +40,7 @@ import ( "bitbucket.org/ausocean/av/protocol/rtp" ) -// NAL types. +// NAL types (from https://tools.ietf.org/html/rfc6184#page-13) const ( // Single nal units bounds. typeSingleNALULowBound = 1 From 85a5f043c88b5b95cde444e91697d2eec3bd43fc Mon Sep 17 00:00:00 2001 From: Saxon Date: Mon, 27 May 2019 14:08:44 +0930 Subject: [PATCH 20/27] codec/h264/lex.go: removed FUIndicatorIdx const in handleFUA --- codec/h264/lex.go | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/codec/h264/lex.go b/codec/h264/lex.go index 81eca759..b9a58ad2 100644 --- a/codec/h264/lex.go +++ b/codec/h264/lex.go @@ -269,8 +269,7 @@ func (l *RTPLexer) handleFUA(d []byte) { if end { panic("bad fragmentation packet") } - const FUIndicatorIdx = 0 - newHead := (d[FUIndicatorIdx] & 0xe0) | (d[FUHeadIdx] & 0x1f) + newHead := (d[0] & 0xe0) | (d[0] & 0x1f) d = d[1:] d[0] = newHead l.frag = true From 0ee6e86638bc92af8209127de46a9c3d0c81c46a Mon Sep 17 00:00:00 2001 From: Saxon Date: Mon, 27 May 2019 14:23:33 +0930 Subject: [PATCH 21/27] codec/h264/lex.go: fixed bug in handleFUA --- codec/h264/lex.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/codec/h264/lex.go b/codec/h264/lex.go index b9a58ad2..07c9acc1 100644 --- a/codec/h264/lex.go +++ b/codec/h264/lex.go @@ -266,19 +266,19 @@ func (l *RTPLexer) handleFUA(d []byte) { // If start, form new header, skip FU indicator only and set first byte to // new header. Otherwise, skip over both FU indicator and FU header. if start { + newHead := (d[0] & 0xe0) | (d[1] & 0x1f) + d = d[1:] + d[0] = newHead if end { panic("bad fragmentation packet") } - newHead := (d[0] & 0xe0) | (d[0] & 0x1f) - d = d[1:] - d[0] = newHead l.frag = true l.writeWithPrefix(d) } else { + d = d[2:] if end { l.frag = false } - d = d[2:] l.writeNoPrefix(d) } } From 061b01529705afa593665306ef7b83ecf88b2a65 Mon Sep 17 00:00:00 2001 From: Saxon Date: Tue, 28 May 2019 12:37:50 +0930 Subject: [PATCH 22/27] revid/config.go: updated commend for Config's InputCodec field. --- revid/config.go | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/revid/config.go b/revid/config.go index f2d04193..4046f320 100644 --- a/revid/config.go +++ b/revid/config.go @@ -133,10 +133,9 @@ type Config struct { // RTSPURL must also be defined. Input uint8 - // InputCodec defines the input codec we wish to use, and therefore define the - // lexer for use in the pipeline. In most cases this defaults according to a - // particular input. Both Raspivid and V4l use H264, but File input may use - // H264 or MJPEG. + // InputCodec defines the input codec we wish to use, and therefore defines the + // lexer for use in the pipeline. This defaults to H264, but H265 is also a + // valid option if we expect this from the input. InputCodec uint8 // Outputs define the outputs we wish to output data too. From 8b7c62602fe00d6e46e1b68e6bf160ab788be447 Mon Sep 17 00:00:00 2001 From: Saxon Date: Thu, 30 May 2019 18:49:27 +0930 Subject: [PATCH 23/27] codec/h264/lex.go: made nalType if statement in RTPLexer.Lex more go like --- codec/h264/lex.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/codec/h264/lex.go b/codec/h264/lex.go index 07c9acc1..1e8e7ecb 100644 --- a/codec/h264/lex.go +++ b/codec/h264/lex.go @@ -204,7 +204,7 @@ func (l *RTPLexer) Lex(dst io.Writer, src io.Reader, delay time.Duration) error continue } - if nalType >= typeSingleNALULowBound && nalType <= typeSingleNALUHighBound { + if typeSingleNALULowBound <= nalType && nalType <= typeSingleNALUHighBound { l.writeWithPrefix(payload) } else { switch nalType { From 20d5f9605f6b555c20ea5c01fab1c44c31a8251e Mon Sep 17 00:00:00 2001 From: Saxon Date: Thu, 30 May 2019 19:05:08 +0930 Subject: [PATCH 24/27] codec/h264/lex.go: checking NAL packet lengths --- codec/h264/lex.go | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/codec/h264/lex.go b/codec/h264/lex.go index 1e8e7ecb..9a9c48ab 100644 --- a/codec/h264/lex.go +++ b/codec/h264/lex.go @@ -59,6 +59,13 @@ const ( typeFUB = 29 ) +// Min NAL lengths. +const ( + minSingleNALLen = 1 + minSTAPALen = 4 + minFUALen = 2 +) + var noDelay = make(chan time.Time) func init() { @@ -205,6 +212,10 @@ func (l *RTPLexer) Lex(dst io.Writer, src io.Reader, delay time.Duration) error } if typeSingleNALULowBound <= nalType && nalType <= typeSingleNALUHighBound { + // If len too small, ignore. + if len(payload) < minSingleNALLen { + continue + } l.writeWithPrefix(payload) } else { switch nalType { @@ -241,6 +252,11 @@ func (l *RTPLexer) Lex(dst io.Writer, src io.Reader, delay time.Duration) error // handleSTAPA parses NAL units from an aggregation packet and writes // them to the Lexers buffer buf. func (l *RTPLexer) handleSTAPA(d []byte) { + // If the length is too small, ignore. + if len(d) < minSTAPALen { + return + } + for i := 1; i < len(d); { size := int(binary.BigEndian.Uint16(d[i:])) @@ -258,6 +274,11 @@ func (l *RTPLexer) handleSTAPA(d []byte) { // handleFUA parses NAL units from fragmentation packets and writes // them to the Lexer's buf. func (l *RTPLexer) handleFUA(d []byte) { + // If length is too small, ignore. + if len(d) < minFUALen { + return + } + // Get start and end indiciators from FU header. const FUHeadIdx = 1 start := d[FUHeadIdx]&0x80 != 0 From aad0c5403905a5363f254cf8fb773333d78e9ad2 Mon Sep 17 00:00:00 2001 From: Saxon Date: Thu, 30 May 2019 19:06:50 +0930 Subject: [PATCH 25/27] codec/h264/lex_test.go: removed unnecessary conversion --- codec/h264/lex_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/codec/h264/lex_test.go b/codec/h264/lex_test.go index 7e3422ea..46191fbd 100644 --- a/codec/h264/lex_test.go +++ b/codec/h264/lex_test.go @@ -252,7 +252,7 @@ type destination [][]byte func (d *destination) Write(p []byte) (int, error) { tmp := make([]byte, len(p)) copy(tmp, p) - *d = append([][]byte(*d), tmp) + *d = append(*d, tmp) return len(p), nil } From fa128d18094801c1fccb393a8ff17f26059bb673 Mon Sep 17 00:00:00 2001 From: Saxon Date: Fri, 7 Jun 2019 01:27:31 +0930 Subject: [PATCH 26/27] codec/h264: renamed RTPLexer to Extracter Renamed the RTPLexer to Extracter, renamed NewRTPLexer to NewExtracter and renamed Lex to Extract. Put Extracter and accompanying methods in file extract.go. Put tests relating to Extracter in extract_test.go. --- codec/h264/extract.go | 217 +++++++++++++++++++++++++++++++++++++ codec/h264/extract_test.go | 173 +++++++++++++++++++++++++++++ 2 files changed, 390 insertions(+) create mode 100644 codec/h264/extract.go create mode 100644 codec/h264/extract_test.go diff --git a/codec/h264/extract.go b/codec/h264/extract.go new file mode 100644 index 00000000..d4fdcb0f --- /dev/null +++ b/codec/h264/extract.go @@ -0,0 +1,217 @@ +/* +NAME + extract.go + +DESCRIPTION + extract.go provides an extracter to get access units from an RTP stream. + +AUTHOR + Saxon Nelson-Milton + +LICENSE + Copyright (C) 2017 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 + along with revid in gpl.txt. If not, see http://www.gnu.org/licenses. +*/ +package h264 + +import ( + "bytes" + "encoding/binary" + "fmt" + "io" + "time" + + "bitbucket.org/ausocean/av/protocol/rtp" +) + +// NAL types (from https://tools.ietf.org/html/rfc6184#page-13) +const ( + // Single nal units bounds. + typeSingleNALULowBound = 1 + typeSingleNALUHighBound = 23 + + // Single-time aggregation packets. + typeSTAPA = 24 + typeSTAPB = 25 + + // Multi-time aggregation packets. + typeMTAP16 = 26 + typeMTAP24 = 27 + + // Fragmentation packets. + typeFUA = 28 + typeFUB = 29 +) + +// Min NAL lengths. +const ( + minSingleNALLen = 1 + minSTAPALen = 4 + minFUALen = 2 +) + +// Buffer sizes. +const ( + maxAUSize = 100000 // Max access unit size in bytes. + maxRTPSize = 1500 // Max ethernet transmission unit in bytes. +) + +// Extracter is an extracter for extracting H264 access units from RTP stream. +type Extracter struct { + buf *bytes.Buffer // Holds the current access unit. + frag bool // Indicates if we're currently dealing with a fragmentation packet. +} + +// NewExtracter returns a new Extracter. +func NewExtracter() *Extracter { + return &Extracter{ + buf: bytes.NewBuffer(make([]byte, 0, maxAUSize))} +} + +// Extract extracts H264 access units from an RTP stream. This function +// expects that each read from src will provide a single RTP packet. +func (e *Extracter) Extract(dst io.Writer, src io.Reader, delay time.Duration) error { + buf := make([]byte, maxRTPSize) + for { + n, err := src.Read(buf) + switch err { + case nil: // Do nothing. + case io.EOF: + return nil + default: + return fmt.Errorf("source read error: %v\n", err) + } + + // Get payload from RTP packet. + payload, err := rtp.Payload(buf[:n]) + if err != nil { + return fmt.Errorf("could not get RTP payload, failed with err: %v\n", err) + } + + nalType := payload[0] & 0x1f + + // If not currently fragmented then we ignore current write. + if e.frag && nalType != typeFUA { + e.buf.Reset() + e.frag = false + continue + } + + if typeSingleNALULowBound <= nalType && nalType <= typeSingleNALUHighBound { + // If len too small, ignore. + if len(payload) < minSingleNALLen { + continue + } + e.writeWithPrefix(payload) + } else { + switch nalType { + case typeSTAPA: + e.handleSTAPA(payload) + case typeFUA: + e.handleFUA(payload) + case typeSTAPB: + panic("STAP-B type unsupported") + case typeMTAP16: + panic("MTAP16 type unsupported") + case typeMTAP24: + panic("MTAP24 type unsupported") + case typeFUB: + panic("FU-B type unsupported") + default: + panic("unsupported type") + } + } + + markerIsSet, err := rtp.Marker(buf[:n]) + if err != nil { + return fmt.Errorf("could not get marker bit, failed with err: %v\n", err) + } + + if markerIsSet { + e.buf.WriteTo(dst) + e.buf.Reset() + } + } + return nil +} + +// handleSTAPA parses NAL units from an aggregation packet and writes +// them to the Extracter's buffer buf. +func (e *Extracter) handleSTAPA(d []byte) { + // If the length is too small, ignore. + if len(d) < minSTAPALen { + return + } + + for i := 1; i < len(d); { + size := int(binary.BigEndian.Uint16(d[i:])) + + // Skip over NAL unit size. + const sizeOfFieldLen = 2 + i += sizeOfFieldLen + + // Get the NALU. + nalu := d[i : i+size] + i += size + e.writeWithPrefix(nalu) + } +} + +// handleFUA parses NAL units from fragmentation packets and writes +// them to the Extracter's buf. +func (e *Extracter) handleFUA(d []byte) { + // If length is too small, ignore. + if len(d) < minFUALen { + return + } + + // Get start and end indiciators from FU header. + const FUHeadIdx = 1 + start := d[FUHeadIdx]&0x80 != 0 + end := d[FUHeadIdx]&0x40 != 0 + + // If start, form new header, skip FU indicator only and set first byte to + // new header. Otherwise, skip over both FU indicator and FU header. + if start { + newHead := (d[0] & 0xe0) | (d[1] & 0x1f) + d = d[1:] + d[0] = newHead + if end { + panic("bad fragmentation packet") + } + e.frag = true + e.writeWithPrefix(d) + } else { + d = d[2:] + if end { + e.frag = false + } + e.writeNoPrefix(d) + } +} + +// write writes a NAL unit to the Extracter's buf in byte stream format using the +// start code. +func (e *Extracter) writeWithPrefix(d []byte) { + const prefix = "\x00\x00\x00\x01" + e.buf.Write([]byte(prefix)) + e.buf.Write(d) +} + +// writeNoPrefix writes data to the Extracter's buf. This is used for non start +// fragmentations of a NALU. +func (e *Extracter) writeNoPrefix(d []byte) { + e.buf.Write(d) +} diff --git a/codec/h264/extract_test.go b/codec/h264/extract_test.go new file mode 100644 index 00000000..cf315606 --- /dev/null +++ b/codec/h264/extract_test.go @@ -0,0 +1,173 @@ +/* +NAME + extract_test.go + +DESCRIPTION + extract_test.go provides tests for the extracter in extract.go + +AUTHOR + Saxon A. Nelson-Milton + +LICENSE + Copyright (C) 2017 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 + along with revid in gpl.txt. If not, see http://www.gnu.org/licenses. +*/ + +package h264 + +import ( + "io" + "testing" +) + +// rtpReader provides an io.Reader for reading the test RTP stream. +type rtpReader struct { + packets [][]byte + idx int +} + +// Read implements io.Reader. +func (r *rtpReader) Read(p []byte) (int, error) { + if r.idx == len(r.packets) { + return 0, io.EOF + } + b := r.packets[r.idx] + n := copy(p, b) + if n < len(r.packets[r.idx]) { + r.packets[r.idx] = r.packets[r.idx][n:] + } else { + r.idx++ + } + return n, nil +} + +// destination holds the access units extracted during the lexing process. +type destination [][]byte + +// Write implements io.Writer. +func (d *destination) Write(p []byte) (int, error) { + tmp := make([]byte, len(p)) + copy(tmp, p) + *d = append(*d, tmp) + return len(p), nil +} + +// TestLex checks that the Lexer can correctly extract H264 access units from +// h264 RTP stream in RTP payload format. +func TestRTPLex(t *testing.T) { + const rtpVer = 2 + + tests := []struct { + packets [][]byte + expect [][]byte + }{ + { + packets: [][]byte{ + { // Single NAL unit. + 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // RTP header. + typeSingleNALULowBound, // NAL header. + 0x01, 0x02, 0x03, 0x04, // NAL Data. + }, + { // Fragmentation (start packet). + 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // RTP header. + typeFUA, // FU indicator. + 0x80 | typeSingleNALULowBound, // FU header. + 0x01, 0x02, 0x03, // FU payload. + }, + { // Fragmentation (middle packet) + 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // RTP header. + typeFUA, // NAL indicator. + typeSingleNALULowBound, // FU header. + 0x04, 0x05, 0x06, // FU payload. + }, + { // Fragmentation (end packet) + 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // RTP header. + typeFUA, // NAL indicator. + 0x40 | typeSingleNALULowBound, // FU header. + 0x07, 0x08, 0x09, // FU payload + }, + + { // Aggregation. Make last packet of access unit => marker bit true. + 0x80, 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // RTP header. + typeSTAPA, // NAL header. + 0x00, 0x04, // NAL 1 size. + 0x01, 0x02, 0x03, 0x04, // NAL 1 data. + 0x00, 0x04, // NAL 2 size. + 0x01, 0x02, 0x03, 0x04, // NAL 2 data. + }, + // Second access unit. + { // Single NAL unit. + 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // RTP header. + typeSingleNALULowBound, // NAL header. + 0x01, 0x02, 0x03, 0x04, // NAL Data. + }, + { // Single NAL. Make last packet of access unit => marker bit true. + 0x80, 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // RTP header. + typeSingleNALULowBound, // NAL header. + 0x01, 0x02, 0x03, 0x04, // NAL data. + }, + }, + expect: [][]byte{ + // First access unit. + { + // NAL 1 + 0x00, 0x00, 0x00, 0x01, // Start code. + typeSingleNALULowBound, // NAL header. + 0x01, 0x02, 0x03, 0x04, // NAL data. + // NAL 2 + 0x00, 0x00, 0x00, 0x01, // Start code. + typeSingleNALULowBound, + 0x01, 0x02, 0x03, // FU payload. + 0x04, 0x05, 0x06, // FU payload. + 0x07, 0x08, 0x09, // FU payload. + // NAL 3 + 0x00, 0x00, 0x00, 0x01, // Start code. + 0x01, 0x02, 0x03, 0x04, // NAL data. + // NAL 4 + 0x00, 0x00, 0x00, 0x01, // Start code. + 0x01, 0x02, 0x03, 0x04, // NAL 2 data + }, + // Second access unit. + { + // NAL 1 + 0x00, 0x00, 0x00, 0x01, // Start code. + typeSingleNALULowBound, // NAL header. + 0x01, 0x02, 0x03, 0x04, // Data. + // NAL 2 + 0x00, 0x00, 0x00, 0x01, // Start code. + typeSingleNALULowBound, // NAL header. + 0x01, 0x02, 0x03, 0x04, // Data. + }, + }, + }, + } + + for testNum, test := range tests { + r := &rtpReader{packets: test.packets} + d := &destination{} + err := NewExtracter().Extract(d, r, 0) + if err != nil { + t.Fatalf("error lexing: %v\n", err) + } + + for i, accessUnit := range test.expect { + for j, part := range accessUnit { + if part != [][]byte(*d)[i][j] { + t.Fatalf("did not get expected data for test: %v.\nGot: %v\nWant: %v\n", testNum, d, test.expect) + } + } + } + } +} From ce457e1d958e5dddc3525f494cf557d7cd0e84af Mon Sep 17 00:00:00 2001 From: Saxon Date: Fri, 7 Jun 2019 02:00:31 +0930 Subject: [PATCH 27/27] should have been in last commit --- codec/h264/lex.go | 193 +------------------------------------ codec/h264/lex_test.go | 142 --------------------------- protocol/rtmp/rtmp_test.go | 4 +- revid/revid.go | 4 +- 4 files changed, 9 insertions(+), 334 deletions(-) diff --git a/codec/h264/lex.go b/codec/h264/lex.go index 9a9c48ab..176c8b3b 100644 --- a/codec/h264/lex.go +++ b/codec/h264/lex.go @@ -7,6 +7,7 @@ DESCRIPTION AUTHOR Dan Kortschak + Saxon Nelson-Milton LICENSE lex.go is Copyright (C) 2017 the Australian Ocean Lab (AusOcean) @@ -25,45 +26,15 @@ LICENSE along with revid in gpl.txt. If not, see http://www.gnu.org/licenses. */ -// lex.go provides a lexer to lex h264 bytestream into access units. - +// Package h264 provides a h264 bytestream lexer and RTP H264 access unit +// extracter. package h264 import ( - "bytes" - "encoding/binary" - "fmt" "io" "time" "bitbucket.org/ausocean/av/codec/codecutil" - "bitbucket.org/ausocean/av/protocol/rtp" -) - -// NAL types (from https://tools.ietf.org/html/rfc6184#page-13) -const ( - // Single nal units bounds. - typeSingleNALULowBound = 1 - typeSingleNALUHighBound = 23 - - // Single-time aggregation packets. - typeSTAPA = 24 - typeSTAPB = 25 - - // Multi-time aggregation packets. - typeMTAP16 = 26 - typeMTAP24 = 27 - - // Fragmentation packets. - typeFUA = 28 - typeFUB = 29 -) - -// Min NAL lengths. -const ( - minSingleNALLen = 1 - minSTAPALen = 4 - minFUALen = 2 ) var noDelay = make(chan time.Time) @@ -74,11 +45,11 @@ func init() { var h264Prefix = [...]byte{0x00, 0x00, 0x01, 0x09, 0xf0} -// LexFromBytestream lexes H.264 NAL units read from src into separate writes +// Lex lexes H.264 NAL units read from src into separate writes // to dst with successive writes being performed not earlier than the specified // delay. NAL units are split after type 1 (Coded slice of a non-IDR picture), 5 // (Coded slice of a IDR picture) and 8 (Picture parameter set). -func LexFromBytestream(dst io.Writer, src io.Reader, delay time.Duration) error { +func Lex(dst io.Writer, src io.Reader, delay time.Duration) error { var tick <-chan time.Time if delay == 0 { tick = noDelay @@ -163,157 +134,3 @@ outer: _, err := dst.Write(buf) return err } - -// Buffer sizes. -const ( - maxAUSize = 100000 // Max access unit size in bytes. - maxRTPSize = 1500 // Max ethernet transmission unit in bytes. -) - -// RTPLexer is a lexer for lexing H264 from RTP packets. -type RTPLexer struct { - buf *bytes.Buffer // Holds the current access unit. - frag bool // Indicates if we're currently dealing with a fragmentation packet. -} - -// NewRTPLexer returns a new RTPLexer. -func NewRTPLexer() *RTPLexer { - return &RTPLexer{ - buf: bytes.NewBuffer(make([]byte, 0, maxAUSize))} -} - -// Lex extracts H264 access units from an RTP stream. This function -// expects that each read from src will provide a single RTP packet. -func (l *RTPLexer) Lex(dst io.Writer, src io.Reader, delay time.Duration) error { - buf := make([]byte, maxRTPSize) - for { - n, err := src.Read(buf) - switch err { - case nil: // Do nothing. - case io.EOF: - return nil - default: - return fmt.Errorf("source read error: %v\n", err) - } - - // Get payload from RTP packet. - payload, err := rtp.Payload(buf[:n]) - if err != nil { - return fmt.Errorf("could not get RTP payload, failed with err: %v\n", err) - } - - nalType := payload[0] & 0x1f - - // If not currently fragmented then we ignore current write. - if l.frag && nalType != typeFUA { - l.buf.Reset() - l.frag = false - continue - } - - if typeSingleNALULowBound <= nalType && nalType <= typeSingleNALUHighBound { - // If len too small, ignore. - if len(payload) < minSingleNALLen { - continue - } - l.writeWithPrefix(payload) - } else { - switch nalType { - case typeSTAPA: - l.handleSTAPA(payload) - case typeFUA: - l.handleFUA(payload) - case typeSTAPB: - panic("STAP-B type unsupported") - case typeMTAP16: - panic("MTAP16 type unsupported") - case typeMTAP24: - panic("MTAP24 type unsupported") - case typeFUB: - panic("FU-B type unsupported") - default: - panic("unsupported type") - } - } - - markerIsSet, err := rtp.Marker(buf[:n]) - if err != nil { - return fmt.Errorf("could not get marker bit, failed with err: %v\n", err) - } - - if markerIsSet { - l.buf.WriteTo(dst) - l.buf.Reset() - } - } - return nil -} - -// handleSTAPA parses NAL units from an aggregation packet and writes -// them to the Lexers buffer buf. -func (l *RTPLexer) handleSTAPA(d []byte) { - // If the length is too small, ignore. - if len(d) < minSTAPALen { - return - } - - for i := 1; i < len(d); { - size := int(binary.BigEndian.Uint16(d[i:])) - - // Skip over NAL unit size. - const sizeOfFieldLen = 2 - i += sizeOfFieldLen - - // Get the NALU. - nalu := d[i : i+size] - i += size - l.writeWithPrefix(nalu) - } -} - -// handleFUA parses NAL units from fragmentation packets and writes -// them to the Lexer's buf. -func (l *RTPLexer) handleFUA(d []byte) { - // If length is too small, ignore. - if len(d) < minFUALen { - return - } - - // Get start and end indiciators from FU header. - const FUHeadIdx = 1 - start := d[FUHeadIdx]&0x80 != 0 - end := d[FUHeadIdx]&0x40 != 0 - - // If start, form new header, skip FU indicator only and set first byte to - // new header. Otherwise, skip over both FU indicator and FU header. - if start { - newHead := (d[0] & 0xe0) | (d[1] & 0x1f) - d = d[1:] - d[0] = newHead - if end { - panic("bad fragmentation packet") - } - l.frag = true - l.writeWithPrefix(d) - } else { - d = d[2:] - if end { - l.frag = false - } - l.writeNoPrefix(d) - } -} - -// write writes a NAL unit to the Lexer's buf in byte stream format using the -// start code. -func (l *RTPLexer) writeWithPrefix(d []byte) { - const prefix = "\x00\x00\x00\x01" - l.buf.Write([]byte(prefix)) - l.buf.Write(d) -} - -// writeNoPrefix writes data to the Lexer's buf. This is used for non start -// fragmentations of a NALU. -func (l *RTPLexer) writeNoPrefix(d []byte) { - l.buf.Write(d) -} diff --git a/codec/h264/lex_test.go b/codec/h264/lex_test.go index 46191fbd..87e6b7d2 100644 --- a/codec/h264/lex_test.go +++ b/codec/h264/lex_test.go @@ -31,8 +31,6 @@ LICENSE package h264 import ( - "io" - "testing" "time" ) @@ -223,143 +221,3 @@ func TestH264(t *testing.T) { } } */ - -// rtpReader provides an io.Reader for reading the test RTP stream. -type rtpReader struct { - packets [][]byte - idx int -} - -// Read implements io.Reader. -func (r *rtpReader) Read(p []byte) (int, error) { - if r.idx == len(r.packets) { - return 0, io.EOF - } - b := r.packets[r.idx] - n := copy(p, b) - if n < len(r.packets[r.idx]) { - r.packets[r.idx] = r.packets[r.idx][n:] - } else { - r.idx++ - } - return n, nil -} - -// destination holds the access units extracted during the lexing process. -type destination [][]byte - -// Write implements io.Writer. -func (d *destination) Write(p []byte) (int, error) { - tmp := make([]byte, len(p)) - copy(tmp, p) - *d = append(*d, tmp) - return len(p), nil -} - -// TestLex checks that the Lexer can correctly extract H264 access units from -// h264 RTP stream in RTP payload format. -func TestRTPLex(t *testing.T) { - const rtpVer = 2 - - tests := []struct { - packets [][]byte - expect [][]byte - }{ - { - packets: [][]byte{ - { // Single NAL unit. - 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // RTP header. - typeSingleNALULowBound, // NAL header. - 0x01, 0x02, 0x03, 0x04, // NAL Data. - }, - { // Fragmentation (start packet). - 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // RTP header. - typeFUA, // FU indicator. - 0x80 | typeSingleNALULowBound, // FU header. - 0x01, 0x02, 0x03, // FU payload. - }, - { // Fragmentation (middle packet) - 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // RTP header. - typeFUA, // NAL indicator. - typeSingleNALULowBound, // FU header. - 0x04, 0x05, 0x06, // FU payload. - }, - { // Fragmentation (end packet) - 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // RTP header. - typeFUA, // NAL indicator. - 0x40 | typeSingleNALULowBound, // FU header. - 0x07, 0x08, 0x09, // FU payload - }, - - { // Aggregation. Make last packet of access unit => marker bit true. - 0x80, 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // RTP header. - typeSTAPA, // NAL header. - 0x00, 0x04, // NAL 1 size. - 0x01, 0x02, 0x03, 0x04, // NAL 1 data. - 0x00, 0x04, // NAL 2 size. - 0x01, 0x02, 0x03, 0x04, // NAL 2 data. - }, - // Second access unit. - { // Single NAL unit. - 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // RTP header. - typeSingleNALULowBound, // NAL header. - 0x01, 0x02, 0x03, 0x04, // NAL Data. - }, - { // Single NAL. Make last packet of access unit => marker bit true. - 0x80, 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // RTP header. - typeSingleNALULowBound, // NAL header. - 0x01, 0x02, 0x03, 0x04, // NAL data. - }, - }, - expect: [][]byte{ - // First access unit. - { - // NAL 1 - 0x00, 0x00, 0x00, 0x01, // Start code. - typeSingleNALULowBound, // NAL header. - 0x01, 0x02, 0x03, 0x04, // NAL data. - // NAL 2 - 0x00, 0x00, 0x00, 0x01, // Start code. - typeSingleNALULowBound, - 0x01, 0x02, 0x03, // FU payload. - 0x04, 0x05, 0x06, // FU payload. - 0x07, 0x08, 0x09, // FU payload. - // NAL 3 - 0x00, 0x00, 0x00, 0x01, // Start code. - 0x01, 0x02, 0x03, 0x04, // NAL data. - // NAL 4 - 0x00, 0x00, 0x00, 0x01, // Start code. - 0x01, 0x02, 0x03, 0x04, // NAL 2 data - }, - // Second access unit. - { - // NAL 1 - 0x00, 0x00, 0x00, 0x01, // Start code. - typeSingleNALULowBound, // NAL header. - 0x01, 0x02, 0x03, 0x04, // Data. - // NAL 2 - 0x00, 0x00, 0x00, 0x01, // Start code. - typeSingleNALULowBound, // NAL header. - 0x01, 0x02, 0x03, 0x04, // Data. - }, - }, - }, - } - - for testNum, test := range tests { - r := &rtpReader{packets: test.packets} - d := &destination{} - err := NewRTPLexer().Lex(d, r, 0) - if err != nil { - t.Fatalf("error lexing: %v\n", err) - } - - for i, accessUnit := range test.expect { - for j, part := range accessUnit { - if part != [][]byte(*d)[i][j] { - t.Fatalf("did not get expected data for test: %v.\nGot: %v\nWant: %v\n", testNum, d, test.expect) - } - } - } - } -} diff --git a/protocol/rtmp/rtmp_test.go b/protocol/rtmp/rtmp_test.go index c7c38bd4..e1e79796 100644 --- a/protocol/rtmp/rtmp_test.go +++ b/protocol/rtmp/rtmp_test.go @@ -199,7 +199,7 @@ func TestFromFrame(t *testing.T) { if err != nil { t.Errorf("Failed to create flv encoder with error: %v", err) } - err = h264.LexFromBytestream(flvEncoder, bytes.NewReader(videoData), time.Second/time.Duration(frameRate)) + err = h264.Lex(flvEncoder, bytes.NewReader(videoData), time.Second/time.Duration(frameRate)) if err != nil { t.Errorf("Lexing failed with error: %v", err) } @@ -251,7 +251,7 @@ func TestFromFile(t *testing.T) { if err != nil { t.Fatalf("failed to create encoder: %v", err) } - err = h264.LexFromBytestream(flvEncoder, f, time.Second/time.Duration(25)) + err = h264.Lex(flvEncoder, f, time.Second/time.Duration(25)) if err != nil { t.Errorf("Lexing and encoding failed with error: %v", err) } diff --git a/revid/revid.go b/revid/revid.go index 7b88aada..2f0f768e 100644 --- a/revid/revid.go +++ b/revid/revid.go @@ -280,10 +280,10 @@ func (r *Revid) setupPipeline(mtsEnc, flvEnc func(dst io.WriteCloser, rate int) switch r.config.Input { case Raspivid: r.setupInput = r.startRaspivid - r.lexTo = h264.LexFromBytestream + r.lexTo = h264.Lex case V4L: r.setupInput = r.startV4L - r.lexTo = h264.LexFromBytestream + r.lexTo = h264.Lex case File: r.setupInput = r.setupInputForFile case RTSP: