more changes to processor

Author: noriah

catnip.go

@@ -10,6 +10,9 @@ import (
 	"github.com/pkg/errors"
 )
 
+const MaxChannelCount = 2
+const MaxSampleSize = 2048
+
 type SetupFunc func() error
 type StartFunc func(ctx context.Context) (context.Context, error)
 type CleanupFunc func() error
@@ -79,14 +82,11 @@ func Run(cfg *Config, ctx context.Context) error {
 		}
 	}
 
-	ctx = vis.Start(ctx)
-	defer vis.Stop()
-
 	kickChan := make(chan bool, 1)
-
 	mu := &sync.Mutex{}
 
-	go vis.Process(ctx, kickChan, mu)
+	ctx = vis.Start(ctx, kickChan, mu)
+	defer vis.Stop()
 
 	if err := audio.Start(ctx, inputBuffers, kickChan, mu); err != nil {
 		if !errors.Is(ctx.Err(), context.Canceled) {

config.go

@@ -2,6 +2,7 @@ package catnip
 
 import (
 	"errors"
+	"fmt"
 
 	"github.com/noriah/catnip/dsp"
 	"github.com/noriah/catnip/dsp/window"
@@ -49,9 +50,6 @@ func NewZeroConfig() Config {
 		SampleRate:   44100,
 		SampleSize:   1024,
 		ChannelCount: 1,
-		ProcessRate:  0,
-		Combine:      false,
-		UseThreaded:  false,
 	}
 }
 
@@ -65,13 +63,14 @@ func (cfg *Config) Validate() error {
 	}
 
 	switch {
-
-	case cfg.ChannelCount > 2:
-		return errors.New("too many channels (2 max)")
+	case cfg.ChannelCount > MaxChannelCount:
+		return fmt.Errorf("too many channels (%d max)", MaxChannelCount)
 
 	case cfg.ChannelCount < 1:
 		return errors.New("too few channels (1 min)")
 
+	case cfg.SampleSize > MaxSampleSize:
+		return fmt.Errorf("sample size too large (%d max)", MaxSampleSize)
 	}
 
 	return nil

dsp/analyzer.go

@@ -31,11 +31,10 @@ type Analyzer interface {
 
 // analyzer is an audio spectrum in a buffer
 type analyzer struct {
-	cfg           AnalyzerConfig // the analyzer config
-	bins          []bin          // bins for processing
-	binCount      int            // number of bins we look at
-	fftSize       int            // number of fft bins
-	dontNormalize bool           // dont normalize the output
+	cfg      AnalyzerConfig // the analyzer config
+	bins     []bin          // bins for processing
+	binCount int            // number of bins we look at
+	fftSize  int            // number of fft bins
 }
 
 // Bin is a helper struct for spectrum
@@ -109,10 +108,9 @@ func MinNonZeroSampleValue() BinMethod {
 
 func NewAnalyzer(cfg AnalyzerConfig) Analyzer {
 	return &analyzer{
-		cfg:           cfg,
-		bins:          make([]bin, cfg.SampleSize),
-		fftSize:       cfg.SampleSize/2 + 1,
-		dontNormalize: cfg.DontNormalize,
+		cfg:     cfg,
+		bins:    make([]bin, cfg.SampleSize),
+		fftSize: cfg.SampleSize/2 + 1,
 	}
 }
 
@@ -154,7 +152,7 @@ func (az *analyzer) ProcessBin(idx int, src []complex128) float64 {
 		return 0.0
 	}
 
-	if !az.dontNormalize {
+	if !az.cfg.DontNormalize {
 		mag = math.Log(mag)
 	}
 

go.mod

@@ -1,6 +1,6 @@
 module github.com/noriah/catnip
 
-go 1.18
+go 1.20
 
 require (
 	github.com/integrii/flaggy v1.4.4

graphic/display.go

@@ -6,6 +6,7 @@ import (
 	"sync/atomic"
 
 	"github.com/noriah/catnip/util"
+
 	"github.com/nsf/termbox-go"
 )
 
@@ -393,19 +394,15 @@ func (d *Display) SetInvertDraw(invert bool) {
 }
 
 // Bins returns the number of bars we will draw.
-func (d *Display) Bins(sets ...int) int {
-	x := 1
-	if len(sets) > 0 {
-		x = sets[0]
-	}
+func (d *Display) Bins(chCount int) int {
 
 	switch d.drawType {
 	case DrawUp, DrawDown:
-		return (d.termWidth / d.binSize) / x
+		return (d.termWidth / d.binSize) / chCount
 	case DrawUpDown:
 		return d.termWidth / d.binSize
 	case DrawLeft, DrawRight:
-		return (d.termHeight / d.binSize) / x
+		return (d.termHeight / d.binSize) / chCount
 	case DrawLeftRight:
 		return d.termHeight / d.binSize
 	default:

processor/processor.go

@@ -12,14 +12,14 @@ import (
 )
 
 type Output interface {
-	Bins(...int) int
+	Bins(int) int
 	Write([][]float64, int) error
 }
 
 type Processor interface {
-	Start(ctx context.Context) context.Context
+	Start(ctx context.Context, kickChan chan bool, mu *sync.Mutex) context.Context
 	Stop()
-	Process(context.Context, chan bool, *sync.Mutex)
+	Process()
 }
 
 type Config struct {
@@ -49,6 +49,9 @@ type processor struct {
 
 	plans []*fft.Plan
 
+	mu        *sync.Mutex
+	ctxCancel context.CancelFunc
+
 	anlz  dsp.Analyzer
 	out   Output
 	smth  dsp.Smoother
@@ -80,16 +83,20 @@ func New(cfg Config) *processor {
 	return vis
 }
 
-func (vis *processor) Start(ctx context.Context) context.Context {
+func (vis *processor) Start(ctx context.Context, kickChan chan bool, mu *sync.Mutex) context.Context {
+	newCtx, cancel := context.WithCancel(ctx)
+	vis.ctxCancel = cancel
+	vis.mu = mu
+	go vis.run(newCtx, kickChan)
 
-	return ctx
+	return newCtx
 }
 
-func (vis *processor) Stop() {}
-
-// Process runs processing on sample sets and calls Write on the output once per sample set.
-func (vis *processor) Process(ctx context.Context, kickChan chan bool, mu *sync.Mutex) {
+func (vis *processor) Stop() {
+	vis.ctxCancel()
+}
 
+func (vis *processor) run(ctx context.Context, kickChan chan bool) {
 	if vis.processRate <= 0 {
 		// if we do not have a framerate set, allow at most 1 second per sampling
 		vis.processRate = 1
@@ -100,33 +107,7 @@ func (vis *processor) Process(ctx context.Context, kickChan chan bool, mu *sync.
 	defer ticker.Stop()
 
 	for {
-		mu.Lock()
-		for idx := range vis.barBufs {
-			if vis.wndwr != nil {
-				vis.wndwr(vis.inputBufs[idx])
-			}
-			vis.plans[idx].Execute()
-		}
-		mu.Unlock()
-
-		if n := vis.out.Bins(vis.channelCount); n != vis.bars {
-			vis.bars = vis.anlz.Recalculate(n)
-		}
-
-		for idx, fftBuf := range vis.fftBufs {
-			buf := vis.barBufs[idx]
-
-			for bIdx := range buf[:vis.bars] {
-				buf[bIdx] = vis.anlz.ProcessBin(bIdx, fftBuf)
-			}
-		}
-
-		if vis.smth != nil {
-			vis.smth.SmoothBuffers(vis.barBufs)
-		}
-
-		vis.out.Write(vis.barBufs, vis.channelCount)
-
+		vis.Process()
 		select {
 		case <-ctx.Done():
 			return
@@ -137,3 +118,33 @@ func (vis *processor) Process(ctx context.Context, kickChan chan bool, mu *sync.
 		ticker.Reset(dur)
 	}
 }
+
+// Process runs processing on sample sets and calls Write on the output once per sample set.
+func (vis *processor) Process() {
+	vis.mu.Lock()
+	for idx := range vis.barBufs {
+		if vis.wndwr != nil {
+			vis.wndwr(vis.inputBufs[idx])
+		}
+		vis.plans[idx].Execute()
+	}
+	vis.mu.Unlock()
+
+	if n := vis.out.Bins(vis.channelCount); n != vis.bars {
+		vis.bars = vis.anlz.Recalculate(n)
+	}
+
+	for idx, fftBuf := range vis.fftBufs {
+		buf := vis.barBufs[idx]
+
+		for bIdx := range buf[:vis.bars] {
+			buf[bIdx] = vis.anlz.ProcessBin(bIdx, fftBuf)
+		}
+	}
+
+	if vis.smth != nil {
+		vis.smth.SmoothBuffers(vis.barBufs)
+	}
+
+	vis.out.Write(vis.barBufs, vis.channelCount)
+}

processor/processor_test.go

@@ -0,0 +1,66 @@
+package processor
+
+import (
+	"sync"
+	"testing"
+
+	"github.com/noriah/catnip/input"
+)
+
+const (
+	BinSize = 1024
+	ChCount = 2
+)
+
+type testOutput struct{}
+
+func (tO *testOutput) Bins(int) int {
+	return BinSize
+}
+
+func (tO *testOutput) Write([][]float64, int) error {
+	return nil
+}
+
+type testAnalyzer struct{}
+
+func (tA *testAnalyzer) BinCount() int {
+	return BinSize
+}
+
+func (tA *testAnalyzer) ProcessBin(int, []complex128) float64 {
+	return 0.0
+}
+
+func (ta *testAnalyzer) Recalculate(int) int {
+	return BinSize
+}
+
+type Analyzer interface {
+	BinCount() int
+	ProcessBin(int, []complex128) float64
+	Recalculate(int) int
+}
+
+func BenchmarkSlices(b *testing.B) {
+
+	inputBuffers := input.MakeBuffers(ChCount, BinSize)
+
+	cfg := Config{
+		SampleRate:   122880.0,
+		SampleSize:   BinSize,
+		ChannelCount: ChCount,
+		Buffers:      inputBuffers,
+		Output:       &testOutput{},
+		Analyzer:     &testAnalyzer{},
+	}
+
+	proc := New(cfg)
+	proc.mu = &sync.Mutex{}
+
+	b.ResetTimer()
+
+	for i := 0; i < b.N; i++ {
+		proc.Process()
+	}
+}

processor/threaded.go

@@ -90,7 +90,7 @@ func (vis *threadedProcessor) channelProcessor(ch int, kick <-chan bool) {
 	}
 }
 
-func (vis *threadedProcessor) Start(ctx context.Context) context.Context {
+func (vis *threadedProcessor) Start(ctx context.Context, kickChan chan bool, mu *sync.Mutex) context.Context {
 	vis.ctx, vis.cancel = context.WithCancel(ctx)
 
 	for i, kick := range vis.kicks {
@@ -107,7 +107,7 @@ func (vis *threadedProcessor) Stop() {
 }
 
 // Process runs one draw refresh with the visualizer on the termbox screen.
-func (vis *threadedProcessor) Process(ctx context.Context, kickChan chan bool, mu *sync.Mutex) {
+func (vis *threadedProcessor) Process() {
 	if n := vis.out.Bins(vis.channelCount); n != vis.bars {
 		vis.bars = vis.anlz.Recalculate(n)
 	}