added 4x goroutines with mutex lock and waitgroups

codec/mjpeg/lex.go

@@ -86,6 +86,7 @@ func Lex(dst io.Writer, src io.Reader, delay time.Duration) error {
 			}
 			last = b
 		}
+		fmt.Print("lex: just before write")
 		<-tick
 		_, err = dst.Write(buf)
 		if err != nil {

filter/basic.go

@@ -36,6 +36,8 @@ import (
 	"image/jpeg"
 	"io"
 	"os"
+	"strconv"
+	"sync"
 	"time"
 
 	"golang.org/x/image/font"
@@ -57,85 +59,48 @@ func absDiff(a, b uint32) int {
 // Mixture of Gaussians method.
 type BasicFilter struct {
 	dst    io.WriteCloser
+	img    image.Image
 	bg     [][][3]uint32
+	bwImg  *image.RGBA
 	w      int
 	h      int
 	file   io.WriteCloser
+	motion int
 	debug  bool
 }
 
 // NewMOGFilter returns a pointer to a new MOGFilter struct.
 func NewBasicFilter(dst io.WriteCloser, debug bool) *BasicFilter {
+	bwImg := image.NewRGBA(image.Rect(0, 0, 0, 0))
 	var file io.WriteCloser
 	var err error
 	if debug {
 		file, err = os.Create("motion.mjpeg")
 		if err != nil {
-			panic("!!! TEST CODE !!!: file didnt work")
+			panic("debug file didn't create")
 		}
 	} else {
 		file = nil
 	}
-
+	return &BasicFilter{dst, nil, nil, bwImg, 0, 0, file, 0, debug}
-	return &BasicFilter{dst, nil, 0, 0, file, debug}
 }
 
 // Implements io.Closer.
 // Close frees resources used by gocv, because it has to be done manually, due to
 // it using c-go.
-func (m *BasicFilter) Close() error {
+func (b *BasicFilter) Close() error {
 	return nil
 }
 
-// Implements io.Writer.
+// Go routine for one row of the image to be processed
-// Write applies the motion filter to the video stream. Only frames with motion
+func (b *BasicFilter) Process(j int, wg *sync.WaitGroup) {
-// are written to the destination encoder, frames without are discarded.
+	defer wg.Done()
-func (b *BasicFilter) Write(f []byte) (int, error) {
-	t0 := time.Now()
-	//decode MJPEG
-	r := bytes.NewReader(f)
-	img, err := jpeg.Decode(r)
-	if err != nil {
-		return 0, fmt.Errorf("image can't be decoded: %w", err)
-	}
-	t1 := time.Now()
-
-	//get background image and save a new background image if needed
-	//first frame must always be sent
-	if b.bg == nil {
-		bounds := img.Bounds()
-		b.w = bounds.Max.X
-		b.h = bounds.Max.Y
-
-		b.bg = make([][][3]uint32, b.h)
-		for i, _ := range b.bg {
-			b.bg[i] = make([][3]uint32, b.w)
-		}
-
-		for j := 0; j < b.h; j++ {
-			for i := 0; i < b.w; i++ {
-				n := img.At(i, j)
-				R, G, B, _ := n.RGBA()
-				copy(b.bg[j][i][:], []uint32{R, G, B})
-			}
-		}
-		return len(f), nil
-	}
-
-	motion := 0
-	//for all pixels get the difference from the background image
-
-	bwImg := image.NewRGBA(image.Rect(0, 0, b.w, b.h))
-	for j := 0; j < b.h; j++ {
 	for i := 0; i < b.w; i++ {
 		// ti0 := time.Now()
-			n := img.At(i, j)
+		n := b.img.At(i, j)
 		// ti1 := time.Now()
 		R, G, B, _ := n.RGBA()
 		// ti2 := time.Now()
-			// diffR := int(math.Abs(float64(R) - float64(b.bg[j][i][0])))
-			// diffG := int(math.Abs(float64(G) - float64(b.bg[j][i][1])))
-			// diffB := int(math.Abs(float64(B) - float64(b.bg[j][i][2])))
 
 		diffB := absDiff(B, b.bg[j][i][2])
 		diffR := absDiff(R, b.bg[j][i][0])
@@ -146,14 +111,14 @@ func (b *BasicFilter) Write(f []byte) (int, error) {
 		diff := diffR + diffG + diffB
 
 		if diff > 45000 {
-				motion++
+			b.motion++
 		}
 
 		if b.debug {
 			if diff > 45000 {
-					bwImg.SetRGBA(i, j, color.RGBA{0xff, 0xff, 0xff, 0xff})
+				b.bwImg.SetRGBA(i, j, color.RGBA{0xff, 0xff, 0xff, 0xff})
 			} else {
-					bwImg.SetRGBA(i, j, color.RGBA{0x00, 0x00, 0x00, 0xff})
+				b.bwImg.SetRGBA(i, j, color.RGBA{0x00, 0x00, 0x00, 0xff})
 			}
 		}
 		// ti4 := time.Now()
@@ -169,21 +134,90 @@ func (b *BasicFilter) Write(f []byte) (int, error) {
 		// fmt.Print("Total: ", time.Now().Sub(ti0).Nanoseconds(), "\n\n")
 	}
 }
+
+// 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 (b *BasicFilter) Write(f []byte) (int, error) {
+	fmt.Print("in basic write\n")
+	t0 := time.Now()
+	//decode MJPEG
+	r := bytes.NewReader(f)
+	var err error
+	b.img, err = jpeg.Decode(r)
+	if err != nil {
+		return 0, fmt.Errorf("image can't be decoded: %w", err)
+	}
+
+	t1 := time.Now()
+
+	//get background image and save a new background image if needed
+	//first frame must always be sent
+	if b.bg == nil {
+		bounds := b.img.Bounds()
+		b.w = bounds.Max.X
+		b.h = bounds.Max.Y
+
+		b.bwImg = image.NewRGBA(image.Rect(0, 0, b.w, b.h))
+
+		b.bg = make([][][3]uint32, b.h)
+		for i, _ := range b.bg {
+			b.bg[i] = make([][3]uint32, b.w)
+		}
+
+		for j := 0; j < b.h; j++ {
+			for i := 0; i < b.w; i++ {
+				n := b.img.At(i, j)
+				R, G, B, _ := n.RGBA()
+				copy(b.bg[j][i][:], []uint32{R, G, B})
+			}
+		}
+		return len(f), nil
+	}
+
+	//for all pixels get the difference from the background image
+
+	// c := make(chan int)
+	// fmt.Print("channel made \n")
+	var m sync.Mutex
+
+	j := 0
+	m.Lock()
+	var wg sync.WaitGroup
+
+	for j < b.h {
+		wg.Add(4)
+		go b.Process(j, &wg)
+		go b.Process(j+1, &wg)
+		go b.Process(j+2, &wg)
+		go b.Process(j+3, &wg)
+		j = j + 4
+		wg.Wait()
+	}
+
+	if j >= b.h {
+		m.Unlock()
+	}
+
 	t2 := time.Now()
 
 	//visualise
 	if b.debug {
 		col := color.RGBA{200, 100, 0, 255}
 		d := &font.Drawer{
-			Dst:  bwImg,
+			Dst:  b.bwImg,
 			Src:  image.NewUniform(col),
 			Face: basicfont.Face7x13,
 			Dot:  fixed.P(20, 30),
 		}
-		if motion > 1000 {
+		var s string
-			d.DrawString("Motion")
+		if b.motion > 1000 {
+			s = strconv.Itoa(b.motion) + " Motion"
+		} else {
+			s = strconv.Itoa(b.motion)
 		}
-		err = jpeg.Encode(b.file, bwImg, nil)
+		d.DrawString(s)
+		err = jpeg.Encode(b.file, b.bwImg, nil)
 		if err != nil {
 			return len(f), err
 		}
@@ -196,7 +230,7 @@ func (b *BasicFilter) Write(f []byte) (int, error) {
 	fmt.Print("Total: ", time.Now().Sub(t0).Milliseconds(), "\n\n")
 
 	//choose a threshold that motion is detected for if greater
-	if motion < 1000 {
+	if b.motion < 1000 {
 		return len(f), nil
 	}
 	//discard non motion frames

revid/revid.go

@@ -361,7 +361,7 @@ func (r *Revid) setupPipeline(mtsEnc func(dst io.WriteCloser, rate float64) (io.
 			case config.FilterKNN:
 				r.filters[i] = filter.NewKNNFilter(dst, knnMinArea, knnThreshold, knnHistory, knnKernel, r.cfg.ShowWindows)
 			case config.FilterBasic:
-				r.filters[i] = filter.NewBasicFilter(dst, false)
+				r.filters[i] = filter.NewBasicFilter(dst, r.cfg.ShowWindows)
 			default:
 				panic("Undefined Filter")
 			}
@@ -452,7 +452,6 @@ func (r *Revid) Start() error {
 	if err != nil {
 		return fmt.Errorf("could not start input device: %w", err)
 	}
-
 	r.wg.Add(1)
 	go r.processFrom(r.input, 0)