Implement embedding extraction for LargeRecordingAnalyzer

Author: joeweiss

src/birdnetlib/analyzer.py

@@ -568,3 +568,27 @@ def analyze_recording(self, recording):
 
         self.results = results
         recording.detection_list = self.detections
+
+    def extract_embeddings_for_recording(self, recording):
+        print("extract_embeddings_for_recording", recording.filename)
+        start = 0
+        end = recording.sample_secs
+        results = []
+        for segment in read_audio_segments(recording.path, sr=48000):
+            c = segment["segment"]
+            if len(c) < recording.sample_secs * 48000:
+                # If below the minimum segment duration, continue.
+                del c
+                continue
+            start = segment["start_sec"]
+            end = segment["end_sec"]
+
+            data = np.array([c], dtype="float32")
+            e = self._return_embeddings(data)[0].tolist()
+            results.append({"start_time": start, "end_time": end, "embeddings": e})
+
+            # Increment start and end
+            start += recording.sample_secs - recording.overlap
+            end = start + recording.sample_secs
+
+        self.embeddings = results

src/birdnetlib/main.py

@@ -73,12 +73,6 @@ def analyze(self):
         self.analyzed = True
 
     def extract_embeddings(self):
-        # Check that analyzer is not LargeRecordingAnalyzer
-        if isinstance(self.analyzer, LargeRecordingAnalyzer):
-            raise IncompatibleAnalyzerError(
-                "LargeRecordingAnalyzer can only be used with the LargeRecording class"
-            )
-
         # Read and analyze.
         self.read_audio_data()
         self.analyzer.extract_embeddings_for_recording(self)
@@ -480,9 +474,9 @@ def analyze(self):
         self.analyzed = True
 
     def extract_embeddings(self):
-        raise NotImplementedError(
-            "Extraction of embeddings is not yet implemented for LargeRecordingAnalyzer. Use Analyzer if possible."
-        )
+        self.analyzer.extract_embeddings_for_recording(self)
+        self.embeddings_list = self.analyzer.embeddings
+        self.embeddings_extracted = True
 
     def get_extract_array(self, start_sec, end_sec):
         # Returns ndarray trimmed for start_sec:end_sec

tests/test_embeddings.py

@@ -101,8 +101,6 @@ def test_largefile_embeddings():
     # pprint(commandline_results)
     assert len(commandline_results) == 40
 
-    # TODO: Implement for LargeRecording.
-    # Confirm that LargeRecording return not implemented.
     large_analyzer = LargeRecordingAnalyzer()
     recording = LargeRecording(
         large_analyzer,
@@ -113,6 +111,16 @@ def test_largefile_embeddings():
         min_conf=min_conf,
         return_all_detections=True,
     )
-    msg = "Extraction of embeddings is not yet implemented for LargeRecordingAnalyzer. Use Analyzer if possible."
-    with pytest.raises(NotImplementedError, match=msg):
-        recording.extract_embeddings()
+
+    recording.extract_embeddings()
+
+    # Check that birdnetlib results match command line results.
+    assert len(recording.embeddings) == 40
+    for idx, i in enumerate(commandline_results):
+        # Specify the tolerance level
+        tolerance = 1e-4  # 4 decimal points tolerance between BirdNET and birdnetlib.
+
+        # Assert that the arrays are almost equal within the tolerance
+        assert np.allclose(
+            i["embeddings"], recording.embeddings[idx]["embeddings"], atol=tolerance
+        )