mirror
/
av
mirror of https://bitbucket.org/ausocean/av.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into m3u-update

This commit is contained in:
Trek H 2020-01-28 13:14:24 +10:30
parent 640f50543d cae7e55723
commit c5e965c14c
8 changed files with 246 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
cmd/revid-cli/main.go
View File

@ -128,6 +128,7 @@ func handleFlags() config.Config {
httpAddressPtr = flag.String("HttpAddress", "", "Destination address of http posts")
verticalFlipPtr = flag.Bool("VerticalFlip", false, "Flip video vertically: Yes, No")
horizontalFlipPtr = flag.Bool("HorizontalFlip", false, "Flip video horizontally: Yes, No")
loopPtr = flag.Bool("Loop", false, "Loop input source on completion (true/false)")
bitratePtr = flag.Uint("Bitrate", 0, "Bitrate of recorded video")
heightPtr = flag.Uint("Height", 0, "Height in pixels")
widthPtr = flag.Uint("Width", 0, "Width in pixels")
@ -138,6 +139,7 @@ func handleFlags() config.Config {
saturationPtr = flag.Int("Saturation", 0, "Set Saturation. (100-100)")
exposurePtr = flag.String("Exposure", "auto", "Set exposure mode. ("+strings.Join(raspivid.ExposureModes[:], ",")+")")
autoWhiteBalancePtr = flag.String("Awb", "auto", "Set automatic white balance mode. ("+strings.Join(raspivid.AutoWhiteBalanceModes[:], ",")+")")
fileFPSPtr = flag.Int("FileFPS", 0, "File source frame processing FPS")
// Audio specific flags.
sampleRatePtr = flag.Int("SampleRate", 48000, "Sample rate of recorded audio")
@ -254,6 +256,8 @@ func handleFlags() config.Config {
netsender.ConfigFile = *configFilePtr
}
cfg.FileFPS = *fileFPSPtr
cfg.Loop = *loopPtr
cfg.CameraIP = *cameraIPPtr
cfg.Rotation = *rotationPtr
cfg.HorizontalFlip = *horizontalFlipPtr

129
filter/difference.go Normal file
View File

@ -0,0 +1,129 @@
// +build !circleci
/*
DESCRIPTION
A filter that detects motion and discards frames without motion. The
algorithm calculates the absolute difference for each pixel between
two frames, then finds the mean. If the mean is above a given threshold,
then it is considered motion.
AUTHORS
Scott Barnard <scott@ausocean.org>
LICENSE
difference.go is Copyright (C) 2020 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 filter
import (
"fmt"
"image"
"image/color"
"io"
"gocv.io/x/gocv"
)
// Difference is a filter that provides basic motion detection. Difference calculates
// the absolute difference for each pixel between two frames, then finds the mean. If
// the mean is above a given threshold, then it is considered motion.
type Difference struct {
dst io.WriteCloser
thresh float64
prev gocv.Mat
debug bool
windows []*gocv.Window
}
// NewDifference returns a pointer to a new Difference struct.
func NewDifference(dst io.WriteCloser, debug bool, threshold float64) *Difference {
var windows []*gocv.Window
if debug {
windows = []*gocv.Window{gocv.NewWindow("Diff: Bounding boxes"), gocv.NewWindow("Diff: Motion")}
}
return &Difference{dst, threshold, gocv.NewMat(), debug, windows}
}
// Implements io.Closer.
// Close frees resources used by gocv, because it has to be done manually, due to
// it using c-go.
func (d *Difference) Close() error {
d.prev.Close()
for _, window := range d.windows {
window.Close()
}
return nil
}
// Implements io.Writer.
// Write applies the motion filter to the video stream. Only frames with motion
// are written to the destination encoder, frames without are discarded.
func (d *Difference) Write(f []byte) (int, error) {
if d.prev.Empty() {
var err error
d.prev, err = gocv.IMDecode(f, gocv.IMReadColor)
if err != nil {
return 0, err
}
return len(f), nil
}
img, err := gocv.IMDecode(f, gocv.IMReadColor)
defer img.Close()
if err != nil {
return 0, err
}
imgDelta := gocv.NewMat()
defer imgDelta.Close()
// Seperate foreground and background.
gocv.AbsDiff(img, d.prev, &imgDelta)
gocv.CvtColor(imgDelta, &imgDelta, gocv.ColorBGRToGray)
mean := imgDelta.Mean().Val1
// Update History.
d.prev = img.Clone()
// Draw debug information.
if d.debug {
if mean >= d.thresh {
gocv.PutText(
&img,
fmt.Sprintf("motion - mean:%f", mean),
image.Pt(32, 32),
gocv.FontHersheyPlain,
2.0,
color.RGBA{255, 0, 0, 0},
2,
)
}
d.windows[0].IMShow(img)
d.windows[1].IMShow(imgDelta)
d.windows[0].WaitKey(1)
}
// Don't write to destination if there is no motion.
if mean < d.thresh {
return len(f), nil
}
// Write to destination.
return d.dst.Write(f)
}

6
filter/filters_circleci.go
View File

@ -36,6 +36,12 @@ func NewMOGFilter(dst io.WriteCloser, area, threshold float64, history int, debu
return &NoOp{dst: dst}
}
// NewKNNFilter returns a pointer to a new NoOp struct for testing purposes only.
func NewKNNFilter(dst io.WriteCloser, area, threshold float64, history, kernelSize int, debug bool, hf int) *NoOp {
return &NoOp{dst: dst}
}
// NewDiffference returns a pointer to a new NoOp struct for testing purposes only.
func NewDifference(dst io.WriteCloser, debug bool, threshold float64) *NoOp {
return &NoOp{dst: dst}
}

2
go.mod
View File

@ -4,7 +4,7 @@ go 1.13
require (
bitbucket.org/ausocean/iot v1.2.13
bitbucket.org/ausocean/utils v1.2.12
bitbucket.org/ausocean/utils v1.2.13
github.com/Comcast/gots v0.0.0-20190305015453-8d56e473f0f7
github.com/go-audio/audio v0.0.0-20181013203223-7b2a6ca21480
github.com/go-audio/wav v0.0.0-20181013172942-de841e69b884

2
go.sum
View File

@ -12,6 +12,8 @@ bitbucket.org/ausocean/utils v1.2.11 h1:zA0FOaPjN960ryp8PKCkV5y50uWBYrIxCVnXjwbv
bitbucket.org/ausocean/utils v1.2.11/go.mod h1:uXzX9z3PLemyURTMWRhVI8uLhPX4uuvaaO85v2hcob8=
bitbucket.org/ausocean/utils v1.2.12 h1:VnskjWTDM475TnQRhBQE0cNp9D6Y6OELrd4UkD2VVIQ=
bitbucket.org/ausocean/utils v1.2.12/go.mod h1:uXzX9z3PLemyURTMWRhVI8uLhPX4uuvaaO85v2hcob8=
bitbucket.org/ausocean/utils v1.2.13 h1:tUaIywtoMc1+zl1GCVQokX4mL5X7LNHX5O51AgAPrWA=
bitbucket.org/ausocean/utils v1.2.13/go.mod h1:uXzX9z3PLemyURTMWRhVI8uLhPX4uuvaaO85v2hcob8=
github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=
github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
github.com/Comcast/gots v0.0.0-20190305015453-8d56e473f0f7 h1:LdOc9B9Bj6LEsKiXShkLA3/kpxXb6LJpH+ekU2krbzw=

28
revid/config/config.go
View File

@ -86,6 +86,7 @@ const (
defaultAudioInputCodec = codecutil.ADPCM
defaultPSITime = 2
defaultMotionInterval = 5
defaultFileFPS = 0
// Ring buffer defaults.
defaultRBMaxElements = 10000
@ -126,6 +127,7 @@ const (
FilterMOG
FilterVariableFPS
FilterKNN
FilterDifference
)
// OS names
@ -297,6 +299,14 @@ type Config struct {
MOGMinArea float64 // Used to ignore small areas of motion detection.
MOGThreshold float64 // Intensity value from the KNN motion detection algorithm that is considered motion.
MOGHistory uint // Length of MOG filter's history
// If true will restart reading of input after an io.EOF.
Loop bool
// Defines the rate at which frames from a file source are processed.
FileFPS int
// Difference filter parameters.
DiffThreshold float64 // Intensity value from the Difference motion detection algorithm that is considered motion.
}
// TypeData contains information about all of the variables that
@ -310,8 +320,10 @@ var TypeData = map[string]string{
"CameraIP": "string",
"CBR": "bool",
"ClipDuration": "uint",
"DiffThreshold": "float",
"Exposure": "enum:auto,night,nightpreview,backlight,spotlight,sports,snow,beach,verylong,fixedfps,antishake,fireworks",
"Filters": "enums:NoOp,MOG,VariableFPS,KNN",
"FileFPS": "int",
"Filters": "enums:NoOp,MOG,VariableFPS,KNN,Difference",
"FrameRate": "uint",
"Height": "uint",
"HorizontalFlip": "bool",
@ -324,20 +336,21 @@ var TypeData = map[string]string{
"KNNMinArea": "float",
"KNNThreshold": "float",
"logging": "enum:Debug,Info,Warning,Error,Fatal",
"Loop": "bool",
"MinFPS": "float",
"MinFrames": "uint",
"mode": "enum:Normal,Paused,Burst",
"mode": "enum:Normal,Paused,Burst,Loop",
"MOGHistory": "uint",
"MOGMinArea": "float",
"MOGThreshold": "float",
"MotionInterval": "int",
"RBCapacity": "uint",
"RBMaxElements": "uint",
"RBWriteTimeout": "uint",
"Output": "enum:File,Http,Rtmp,Rtp",
"OutputPath": "string",
"Outputs": "enums:File,Http,Rtmp,Rtp",
"Quantization": "uint",
"RBCapacity": "uint",
"RBMaxElements": "uint",
"RBWriteTimeout": "uint",
"Rotation": "uint",
"RTMPURL": "string",
"RTPAddress": "string",
@ -521,6 +534,11 @@ func (c *Config) Validate() error {
}
}
if c.FileFPS <= 0 || (c.FileFPS > 0 && c.Input != InputFile) {
c.Logger.Log(logger.Info, pkg+"FileFPS bad or unset, defaulting", "FileFPS", defaultFileFPS)
c.FileFPS = defaultFileFPS
}
return nil
}

84
revid/revid.go
View File

@ -7,6 +7,7 @@ AUTHORS
Alan Noble <alan@ausocean.org>
Dan Kortschak <dan@ausocean.org>
Trek Hopton <trek@ausocean.org>
Scott Barnard <scott@ausocean.org>
LICENSE
revid is Copyright (C) 2017-2020 the Australian Ocean Lab (AusOcean)
@ -52,6 +53,7 @@ import (
"bitbucket.org/ausocean/av/filter"
"bitbucket.org/ausocean/av/revid/config"
"bitbucket.org/ausocean/iot/pi/netsender"
"bitbucket.org/ausocean/utils/bitrate"
"bitbucket.org/ausocean/utils/ioext"
"bitbucket.org/ausocean/utils/logger"
"bitbucket.org/ausocean/utils/vring"
@ -115,6 +117,9 @@ type Revid struct {
// err will channel errors from revid routines to the handle errors routine.
err chan error
// bitrate is used for bitrate calculations.
bitrate bitrate.Calculator
}
// New returns a pointer to a new Revid with the desired configuration, and/or
@ -148,10 +153,8 @@ func (r *Revid) handleErrors() {
}
// Bitrate returns the result of the most recent bitrate check.
//
// TODO: get this working again.
func (r *Revid) Bitrate() int {
return -1
return r.bitrate.Bitrate()
}
// reset swaps the current config of a Revid with the passed
@ -266,14 +269,14 @@ func (r *Revid) setupPipeline(mtsEnc func(dst io.WriteCloser, rate float64) (io.
return fmt.Errorf("could not initialise MTS ring buffer: %w", err)
}
w = newMtsSender(
newHttpSender(r.ns, r.cfg.Logger.Log),
newHttpSender(r.ns, r.cfg.Logger.Log, r.bitrate.Report),
r.cfg.Logger.Log,
rb,
r.cfg.ClipDuration,
)
mtsSenders = append(mtsSenders, w)
case config.OutputRTP:
w, err := newRtpSender(r.cfg.RTPAddress, r.cfg.Logger.Log, r.cfg.FrameRate)
w, err := newRtpSender(r.cfg.RTPAddress, r.cfg.Logger.Log, r.cfg.FrameRate, r.bitrate.Report)
if err != nil {
r.cfg.Logger.Log(logger.Warning, pkg+"rtp connect error", "error", err.Error())
}
@ -295,6 +298,7 @@ func (r *Revid) setupPipeline(mtsEnc func(dst io.WriteCloser, rate float64) (io.
rtmpConnectionMaxTries,
rb,
r.cfg.Logger.Log,
r.bitrate.Report,
)
if err != nil {
r.cfg.Logger.Log(logger.Warning, pkg+"rtmp connect error", "error", err.Error())
@ -342,6 +346,8 @@ func (r *Revid) setupPipeline(mtsEnc func(dst io.WriteCloser, rate float64) (io.
r.filters[i] = filter.NewVariableFPSFilter(dst, r.cfg.MinFPS, filter.NewMOGFilter(dst, r.cfg.MOGMinArea, r.cfg.MOGThreshold, int(r.cfg.MOGHistory), r.cfg.ShowWindows, r.cfg.MotionInterval))
case config.FilterKNN:
r.filters[i] = filter.NewKNNFilter(dst, r.cfg.KNNMinArea, r.cfg.KNNThreshold, int(r.cfg.KNNHistory), int(r.cfg.KNNKernel), r.cfg.ShowWindows, r.cfg.MotionInterval)
case config.FilterDifference:
r.filters[i] = filter.NewDifference(dst, r.cfg.ShowWindows, r.cfg.DiffThreshold)
default:
panic("Undefined Filter")
}
@ -428,13 +434,14 @@ func (r *Revid) Start() error {
return err
}
err = r.input.Start()
if err != nil {
return fmt.Errorf("could not start input device: %w", err)
// Calculate delay between frames based on FileFPS.
d := time.Duration(0)
if r.cfg.FileFPS != 0 {
d = time.Duration(1000/r.cfg.FileFPS) * time.Millisecond
}
r.wg.Add(1)
go r.processFrom(r.input, 0)
go r.processFrom(r.input, d)
r.running = true
return nil
@ -662,7 +669,7 @@ func (r *Revid) Update(vars map[string]string) error {
}
case "Filters":
filters := strings.Split(value, ",")
m := map[string]int{"NoOp": config.FilterNoOp, "MOG": config.FilterMOG, "VariableFPS": config.FilterVariableFPS, "KNN": config.FilterKNN}
m := map[string]int{"NoOp": config.FilterNoOp, "MOG": config.FilterMOG, "VariableFPS": config.FilterVariableFPS, "KNN": config.FilterKNN, "Difference": config.FilterDifference}
r.cfg.Filters = make([]int, len(filters))
for i, filter := range filters {
v, ok := m[filter]
@ -810,6 +817,13 @@ func (r *Revid) Update(vars map[string]string) error {
break
}
r.cfg.KNNThreshold = v
case "DiffThreshold":
v, err := strconv.ParseFloat(value, 64)
if err != nil {
r.cfg.Logger.Log(logger.Warning, pkg+"invalid DiffThreshold var", "value", value)
break
}
r.cfg.DiffThreshold = v
case "KNNKernel":
v, err := strconv.Atoi(value)
if err != nil {
@ -845,6 +859,18 @@ func (r *Revid) Update(vars map[string]string) error {
break
}
r.cfg.MOGHistory = uint(v)
case "FileFPS":
v, err := strconv.Atoi(value)
if err != nil {
r.cfg.Logger.Log(logger.Warning, pkg+"invalid FileFPS var", "value", value)
break
}
r.cfg.FileFPS = v
case "mode":
r.cfg.Loop = false
if value == "Loop" {
r.cfg.Loop = true
}
}
}
r.cfg.Logger.Log(logger.Info, pkg+"revid config changed", "config", fmt.Sprintf("%+v", r.cfg))
@ -853,14 +879,34 @@ func (r *Revid) Update(vars map[string]string) error {
// processFrom is run as a routine to read from a input data source, lex and
// then send individual access units to revid's encoders.
func (r *Revid) processFrom(read io.Reader, delay time.Duration) {
err := r.lexTo(r.filters[0], read, delay)
r.cfg.Logger.Log(logger.Debug, pkg+"finished lexing")
switch err {
case nil: // Do nothing.
case io.EOF: // TODO: handle this depending on loop mode.
default:
r.err <- err
func (r *Revid) processFrom(in device.AVDevice, delay time.Duration) {
defer r.wg.Done()
for l := true; l; l = r.cfg.Loop {
err := in.Start()
if err != nil {
r.err <- fmt.Errorf("could not start input device: %w", err)
return
}
// Lex data from input device, in, until finished or an error is encountered.
// For a continuous source e.g. a camera or microphone, we should remain
// in this call indefinitely unless in.Stop() is called and an io.EOF is forced.
r.cfg.Logger.Log(logger.Info, pkg+"lexing")
err = r.lexTo(r.filters[0], in, delay)
switch err {
case nil, io.EOF:
case io.ErrUnexpectedEOF:
r.cfg.Logger.Log(logger.Info, pkg+"unexpected EOF from input")
default:
r.err <- err
}
err = in.Stop()
if err != nil {
r.err <- fmt.Errorf("could not stop input source: %w", err)
}
}
r.wg.Done()
r.cfg.Logger.Log(logger.Info, pkg+"finished lexing")
}

20
revid/senders.go
View File

@ -61,19 +61,25 @@ const (
type httpSender struct {
client *netsender.Sender
log func(lvl int8, msg string, args ...interface{})
report func(sent int)
}
// newHttpSender returns a pointer to a new httpSender.
func newHttpSender(ns *netsender.Sender, log func(lvl int8, msg string, args ...interface{})) *httpSender {
func newHttpSender(ns *netsender.Sender, log func(lvl int8, msg string, args ...interface{}), report func(sent int)) *httpSender {
return &httpSender{
client: ns,
log: log,
report: report,
}
}
// Write implements io.Writer.
func (s *httpSender) Write(d []byte) (int, error) {
return len(d), httpSend(d, s.client, s.log)
err := httpSend(d, s.client, s.log)
if err == nil {
s.report(len(d))
}
return len(d), err
}
func (s *httpSender) Close() error { return nil }
@ -276,9 +282,10 @@ type rtmpSender struct {
ring *vring.Buffer
done chan struct{}
wg sync.WaitGroup
report func(sent int)
}
func newRtmpSender(url string, timeout uint, retries int, rb *vring.Buffer, log func(lvl int8, msg string, args ...interface{})) (*rtmpSender, error) {
func newRtmpSender(url string, timeout uint, retries int, rb *vring.Buffer, log func(lvl int8, msg string, args ...interface{}), report func(sent int)) (*rtmpSender, error) {
var conn *rtmp.Conn
var err error
for n := 0; n < retries; n++ {
@ -299,6 +306,7 @@ func newRtmpSender(url string, timeout uint, retries int, rb *vring.Buffer, log
log: log,
ring: rb,
done: make(chan struct{}),
report: report,
}
s.wg.Add(1)
go s.output()
@ -364,6 +372,7 @@ func (s *rtmpSender) Write(d []byte) (int, error) {
if err != nil {
s.log(logger.Warning, pkg+"rtmpSender: ring buffer write error", "error", err.Error())
}
s.report(len(d))
return len(d), nil
}
@ -404,9 +413,10 @@ type rtpSender struct {
log func(lvl int8, msg string, args ...interface{})
encoder *rtp.Encoder
data []byte
report func(sent int)
}
func newRtpSender(addr string, log func(lvl int8, msg string, args ...interface{}), fps uint) (*rtpSender, error) {
func newRtpSender(addr string, log func(lvl int8, msg string, args ...interface{}), fps uint, report func(sent int)) (*rtpSender, error) {
conn, err := net.Dial("udp", addr)
if err != nil {
return nil, err
@ -414,6 +424,7 @@ func newRtpSender(addr string, log func(lvl int8, msg string, args ...interface{
s := &rtpSender{
log: log,
encoder: rtp.NewEncoder(conn, int(fps)),
report: report,
}
return s, nil
}
@ -426,6 +437,7 @@ func (s *rtpSender) Write(d []byte) (int, error) {
if err != nil {
s.log(logger.Warning, pkg+"rtpSender: write error", err.Error())
}
s.report(len(d))
return len(d), nil
}