Merge pull request #172 from tphakala/capture-buffer-fix

refactor: Fix audio capture buffer

internal/analysis/processor/actions.go

@@ -103,8 +103,13 @@ func (a DatabaseAction) Execute(data interface{}) error {
 
 		// Save audio clip to file if enabled
 		if a.Settings.Realtime.Audio.Export.Enabled {
-			time.Sleep(1 * time.Second) // Sleep for 1 second to allow the audio buffer to fill
-			pcmData, _ := a.AudioBuffer.ReadSegment(a.Note.BeginTime, time.Now())
+			// export audio clip from capture buffer
+			pcmData, err := a.AudioBuffer.ReadSegment(a.Note.BeginTime, 15)
+			if err != nil {
+				log.Printf("Failed to read audio segment from buffer: %v", err)
+				return err
+			}
+
 			if err := myaudio.SavePCMDataToWAV(a.Note.ClipName, pcmData); err != nil {
 				log.Printf("error saving audio clip to %s: %s\n", a.Settings.Realtime.Audio.Export.Type, err)
 				return err

internal/analysis/realtime.go

@@ -85,7 +85,7 @@ func RealtimeAnalysis(settings *conf.Settings) error {
 	myaudio.InitRingBuffer(bufferSize)
 
 	// Audio buffer for extended audio clip capture
-	audioBuffer := myaudio.NewAudioBuffer(30, conf.SampleRate, 2)
+	audioBuffer := myaudio.NewAudioBuffer(60, conf.SampleRate, conf.BitDepth/8)
 
 	// init detection queue
 	queue.Init(5, 5)

internal/myaudio/audiobuffer.go

@@ -16,20 +16,24 @@ type AudioBuffer struct {
 	bufferSize     int
 	bufferDuration time.Duration
 	startTime      time.Time
+	initialized    bool
 	lock           sync.Mutex
 }
 
-// Initializes a new AudioBuffer with timestamp tracking
+// NewAudioBuffer initializes a new AudioBuffer with timestamp tracking
 func NewAudioBuffer(durationSeconds int, sampleRate, bytesPerSample int) *AudioBuffer {
 	bufferSize := durationSeconds * sampleRate * bytesPerSample
-	return &AudioBuffer{
-		data:           make([]byte, bufferSize),
+	alignedBufferSize := ((bufferSize + 2047) / 2048) * 2048 // Round up to the nearest multiple of 2048
+	ab := &AudioBuffer{
+		data:           make([]byte, alignedBufferSize),
 		sampleRate:     sampleRate,
 		bytesPerSample: bytesPerSample,
-		bufferSize:     bufferSize,
+		bufferSize:     alignedBufferSize,
 		bufferDuration: time.Second * time.Duration(durationSeconds),
-		startTime:      time.Now(),
+		initialized:    false,
 	}
+
+	return ab
 }
 
 // Write adds PCM audio data to the buffer, ensuring thread safety and accurate timekeeping.
@@ -38,6 +42,12 @@ func (ab *AudioBuffer) Write(data []byte) {
 	ab.lock.Lock()
 	defer ab.lock.Unlock()
 
+	if !ab.initialized {
+		// Initialize the buffer's start time based on the current time.
+		ab.startTime = time.Now()
+		ab.initialized = true
+	}
+
 	// Store the current write index to determine if we've wrapped around the buffer.
 	prevWriteIndex := ab.writeIndex
 
@@ -48,53 +58,66 @@ func (ab *AudioBuffer) Write(data []byte) {
 	ab.writeIndex = (ab.writeIndex + bytesWritten) % ab.bufferSize
 
 	// Determine if the write operation has overwritten old data.
-	if ab.writeIndex <= prevWriteIndex && (bytesWritten >= ab.bufferSize) {
+	if ab.writeIndex <= prevWriteIndex {
 		// If old data has been overwritten, adjust startTime to maintain accurate timekeeping.
 		ab.startTime = time.Now().Add(-ab.bufferDuration)
+		//log.Printf("Buffer has wrapped around, adjusting start time to %v", ab.startTime)
 	}
 }
 
 // ReadSegment extracts a segment of audio data based on precise start and end times, handling wraparounds.
-func (ab *AudioBuffer) ReadSegment(requestedStartTime, requestedEndTime time.Time) ([]byte, error) {
-	// Lock the buffer to prevent concurrent writes or reads from interfering with the update process.
-	ab.lock.Lock()
-	defer ab.lock.Unlock()
-
-	// Calculate time since the buffer's startTime for both requested start and end times
-	startOffset := requestedStartTime.Sub(ab.startTime)
-	endOffset := requestedEndTime.Sub(ab.startTime)
-
-	// Convert time offsets to buffer indices
-	startIndex := int(startOffset.Seconds()) * ab.sampleRate * ab.bytesPerSample
-	endIndex := int(endOffset.Seconds()) * ab.sampleRate * ab.bytesPerSample
-
-	// Normalize indices based on buffer size
-	startIndex = startIndex % ab.bufferSize
-	endIndex = endIndex % ab.bufferSize
-
-	// Check if requested times are within the buffer's timeframe
-	if startOffset < 0 || endOffset < 0 || endOffset <= startOffset {
-		return nil, errors.New("requested times are outside the buffer's current timeframe")
-	}
-
-	// Determine if the read segment wraps around the buffer's end
-	if startIndex < endIndex {
-		// Simple case: The segment does not wrap around
-		segmentSize := endIndex - startIndex
-		segment := make([]byte, segmentSize)
-		copy(segment, ab.data[startIndex:endIndex])
-		return segment, nil
-	} else {
-		// Wraparound case: The segment spans the end and restarts at the beginning of the buffer
-		segmentSize := (ab.bufferSize - startIndex) + endIndex
-		segment := make([]byte, segmentSize)
-
-		// Copy from startIndex to the end of the buffer
-		firstPartSize := ab.bufferSize - startIndex
-		copy(segment[:firstPartSize], ab.data[startIndex:])
-
-		// Copy from the beginning of the buffer to endIndex
-		copy(segment[firstPartSize:], ab.data[:endIndex])
-		return segment, nil
+// It waits until the current time is past the requested end time.
+func (ab *AudioBuffer) ReadSegment(requestedStartTime time.Time, duration int) ([]byte, error) {
+	requestedEndTime := requestedStartTime.Add(time.Duration(duration) * time.Second)
+
+	for {
+		ab.lock.Lock()
+
+		startOffset := requestedStartTime.Sub(ab.startTime)
+		endOffset := requestedEndTime.Sub(ab.startTime)
+
+		startIndex := int(startOffset.Seconds()) * ab.sampleRate * ab.bytesPerSample
+		endIndex := int(endOffset.Seconds()) * ab.sampleRate * ab.bytesPerSample
+
+		startIndex = startIndex % ab.bufferSize
+		endIndex = endIndex % ab.bufferSize
+
+		if startOffset < 0 {
+			if ab.writeIndex == 0 || ab.writeIndex+int(startOffset.Seconds())*ab.sampleRate*ab.bytesPerSample > ab.bufferSize {
+				ab.lock.Unlock()
+				return nil, errors.New("requested start time is outside the buffer's current timeframe")
+			}
+			startIndex = (ab.bufferSize + startIndex) % ab.bufferSize
+		}
+
+		if endOffset < 0 || endOffset <= startOffset {
+			ab.lock.Unlock()
+			return nil, errors.New("requested times are outside the buffer's current timeframe")
+		}
+
+		// Wait until the current time is past the requested end time
+		if time.Now().After(requestedEndTime) {
+			var segment []byte
+			if startIndex < endIndex {
+
+				//log.Printf("Reading segment from %d to %d", startIndex, endIndex)
+				segmentSize := endIndex - startIndex
+				segment = make([]byte, segmentSize)
+				copy(segment, ab.data[startIndex:endIndex])
+			} else {
+				//log.Printf("Buffer has wrapped, reading segment from %d to %d", startIndex, endIndex)
+				segmentSize := (ab.bufferSize - startIndex) + endIndex
+				segment = make([]byte, segmentSize)
+				firstPartSize := ab.bufferSize - startIndex
+				copy(segment[:firstPartSize], ab.data[startIndex:])
+				copy(segment[firstPartSize:], ab.data[:endIndex])
+			}
+			ab.lock.Unlock()
+			return segment, nil
+		}
+
+		//log.Printf("Buffer is not filled yet, waiting for data to be available")
+		ab.lock.Unlock()
+		time.Sleep(1 * time.Second) // Sleep briefly to avoid busy waiting
 	}
 }