On file complete hook

.gitignore

@@ -1,5 +1,6 @@
 # Agents
 .claude
+CLAUDE.md
 
 # Custom
 *.ods*#

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added an `on_file_complete` callback to acoustic `predict(..)`, `predict_session(..)`, `encode(..)` and `encode_session(..)` (all models: 2.4, 3.0, Perch V2). It fires once per input file the moment that file is fully processed, receiving a single-file result (`AcousticFilePredictionResult` / `AcousticFileEncodingResult`); invalid files are reported with their input marked unprocessable. This enables streaming per-file persistence (e.g. resumable multi-file analysis) and live output. The callback runs on a background thread with a copy of the caller's context, off the inference hot path, so it does not regress throughput. File inputs only (not `run_arrays`); a callback that raises cancels the run.
 - Added support for the BirdNET V3.0 (preview) acoustic model with four backends: TFLite/LiteRT (`tf`), ProtoBuf (`pb`), PyTorch (`pt`) and ONNX (`onnx`). Both `predict(..)` and `encode(..)` are supported on all backends. Load via `birdnet.load("acoustic", "3.0", <backend>)`. The `pt` and `onnx` backends require the new `birdnet[pt]` and `birdnet[onnx]` install extras (#41).
 - Added support for the BirdNET-Geomodel V3.0 with TFLite/LiteRT (`tf`) and ProtoBuf (`pb`) backends. Load via `birdnet.load("geo", "3.0", <backend>)` (#41).
 

README.md

@@ -79,9 +79,9 @@ pip install birdnet[onnx] --user
 
 If you encounter issues with audio file reading, please ensure that `libsndfile` is installed on your system.
 
-- **Ubuntu/Debian**: `sudo apt-get install libsndfile1`
-- **macOS** (using Homebrew): `brew install libsndfile`
-- **Windows**: Download and install the precompiled binaries from the [official website](https://github.com/libsndfile/libsndfile/releases/), extract them and add the folder to path.
+* **Ubuntu/Debian**: `sudo apt-get install libsndfile1`
+* **macOS** (using Homebrew): `brew install libsndfile`
+* **Windows**: Download and install the precompiled binaries from the [official website](https://github.com/libsndfile/libsndfile/releases/), extract them and add the folder to path.
 
 ## Supported operations, precisions and devices
 
@@ -191,15 +191,15 @@ The full result is at [example/location.csv](example/location.csv).
 
 If something goes wrong, you can find the log file in the following locations:
 
-- Windows: `C:\Users\{user}\AppData\Local\Temp\birdnet.log`
-- Linux/MacOS: `/tmp/birdnet.log`
+* Windows: `C:\Users\{user}\AppData\Local\Temp\birdnet.log`
+* Linux/MacOS: `/tmp/birdnet.log`
 
 ## File formats
 
 The audio models support all formats compatible with the SoundFile library (see [here](https://python-soundfile.readthedocs.io/en/0.11.0/#read-write-functions)). This includes, but is not limited to, WAV, FLAC, OGG, and AIFF. The flexibility of supported formats ensures that the models can handle a wide variety of audio input types, making them adaptable to different use cases and environments.
 
-- Supported: AIFC, AIFF, AU, AVR, CAF, FLAC, HTK, IRCAM, MAT4, MAT5, MP3, MPC2K, NIST, OGG, OPUS, PAF, PVF, RAW, RF64, SD2, SDS, SVX, VOC, W64, WAV, WAVEX, WVE, XI
-- Not supportet at the moment: AAC, M4A, WMA
+* Supported: AIFC, AIFF, AU, AVR, CAF, FLAC, HTK, IRCAM, MAT4, MAT5, MP3, MPC2K, NIST, OGG, OPUS, PAF, PVF, RAW, RF64, SD2, SDS, SVX, VOC, W64, WAV, WAVEX, WVE, XI
+* Not supported at the moment: AAC, M4A, WMA
 
 ## Model formats and execution details
 
@@ -214,8 +214,8 @@ Ensure your environment is configured to utilize the appropriate model and avail
 
 ## License
 
-- **Source Code**: The source code for this project is licensed under the [MIT License](https://opensource.org/licenses/MIT).
-- **Models**: The models used in this project are licensed under the [Creative Commons Attribution-NonCommercial-ShareAlike 4.0 International License (CC BY-NC-SA 4.0)](https://creativecommons.org/licenses/by-nc-sa/4.0/).
+* **Source Code**: The source code for this project is licensed under the [MIT License](https://opensource.org/licenses/MIT).
+* **Models**: The models used in this project are licensed under the [Creative Commons Attribution-NonCommercial-ShareAlike 4.0 International License (CC BY-NC-SA 4.0)](https://creativecommons.org/licenses/by-nc-sa/4.0/).
 
 Please ensure you review and adhere to the specific license terms provided with each model. Note that educational and research purposes are considered non-commercial use cases.
 

benchmarks/on_file_complete_benchmark.py

@@ -0,0 +1,156 @@
+# ruff: noqa: T201, ANN001
+"""Throughput benchmark for the per-file completion callback (``on_file_complete``).
+
+Compares prediction throughput across three conditions on the same synthetic
+workload to show that enabling the callback does not regress throughput:
+
+  * baseline  – ``on_file_complete=None`` (feature fully inert)
+  * noop      – a callback that does nothing (isolates pipeline overhead:
+                markers, per-file bookkeeping, slice copies, result construction)
+  * persist   – a callback that writes each file's result to CSV (the resumable
+                analysis use case; includes real disk I/O)
+
+Run:
+    python benchmarks/on_file_complete_benchmark.py \
+        --files 48 --seconds 15 --reps 3 --workers 8
+"""
+
+from __future__ import annotations
+
+import argparse
+import statistics
+import tempfile
+import time
+from pathlib import Path
+
+import numpy as np
+import soundfile as sf
+
+from birdnet.model_loader import load
+
+SAMPLE_RATE = 48_000
+
+
+def make_audio_files(out_dir: Path, n_files: int, seconds: float) -> list[str]:
+  rng = np.random.default_rng(1234)
+  n_samples = int(seconds * SAMPLE_RATE)
+  paths: list[str] = []
+  for i in range(n_files):
+    # Low-amplitude noise -> a few detections per file, realistic write sizes.
+    audio = (rng.standard_normal(n_samples) * 0.05).astype(np.float32)
+    p = out_dir / f"bench_{i:04d}.wav"
+    sf.write(p, audio, SAMPLE_RATE)
+    paths.append(str(p))
+  return paths
+
+
+def run_once(model, files, *, on_file_complete, workers, batch_size, top_k) -> float:
+  start = time.perf_counter()
+  with model.predict_session(
+    n_workers=workers,
+    batch_size=batch_size,
+    top_k=top_k,
+    on_file_complete=on_file_complete,
+  ) as session:
+    session.run(files)
+  return time.perf_counter() - start
+
+
+def summarise(name: str, times: list[float], total_audio_s: float) -> dict:
+  best = min(times)
+  median = statistics.median(times)
+  return {
+    "name": name,
+    "median_s": median,
+    "best_s": best,
+    "xrt_median": total_audio_s / median,
+    "xrt_best": total_audio_s / best,
+  }
+
+
+def main() -> None:
+  parser = argparse.ArgumentParser()
+  parser.add_argument("--files", type=int, default=48)
+  parser.add_argument("--seconds", type=float, default=15.0)
+  parser.add_argument("--reps", type=int, default=3)
+  parser.add_argument("--workers", type=int, default=8)
+  parser.add_argument("--batch-size", type=int, default=8)
+  parser.add_argument("--top-k", type=int, default=5)
+  args = parser.parse_args()
+
+  total_audio_s = args.files * args.seconds
+
+  with tempfile.TemporaryDirectory() as tmp:
+    tmp_dir = Path(tmp)
+    csv_dir = tmp_dir / "csv"
+    csv_dir.mkdir()
+    print(
+      f"Generating {args.files} files x {args.seconds}s "
+      f"({total_audio_s / 60:.1f} min audio)..."
+    )
+    files = make_audio_files(tmp_dir, args.files, args.seconds)
+
+    model = load("acoustic", "2.4", "tf", precision="fp32", library="tflite")
+
+    def persist(result) -> None:  # noqa: ANN001
+      name = Path(str(result.inputs[0])).stem
+      result.to_csv(csv_dir / f"{name}.csv", silent=True)
+
+    conditions = {
+      "baseline (no callback)": None,
+      "noop callback": lambda _r: None,
+      "persist callback (to_csv)": persist,
+    }
+
+    common = {
+      "workers": args.workers,
+      "batch_size": args.batch_size,
+      "top_k": args.top_k,
+    }
+
+    # Warm up (model load, tflite graph, process spawn) — not measured.
+    print("Warming up...")
+    run_once(model, files, on_file_complete=None, **common)
+
+    # Interleave conditions across reps so system drift (thermal, scheduling)
+    # affects every condition equally rather than biasing one block.
+    times: dict[str, list[float]] = {name: [] for name in conditions}
+    for r in range(args.reps):
+      for name, cb in conditions.items():
+        dt = run_once(model, files, on_file_complete=cb, **common)
+        times[name].append(dt)
+        print(f"  rep {r + 1}/{args.reps}  {name:28s}: {dt:6.2f}s")
+
+    results = [
+      summarise(name, times[name], total_audio_s) for name in conditions
+    ]
+
+    baseline = results[0]
+    print("\n" + "=" * 78)
+    print(
+      f"Workload: {args.files} files x {args.seconds}s = {total_audio_s / 60:.1f} min "
+      f"| workers={args.workers} batch={args.batch_size} top_k={args.top_k} "
+      f"| reps={args.reps}"
+    )
+    print("=" * 78)
+    header = (
+      f"{'condition':30s}{'median s':>11s}{'best s':>10s}"
+      f"{'xRT (best)':>13s}{'vs baseline':>14s}"
+    )
+    print(header)
+    print("-" * 78)
+    for res in results:
+      delta = (res["best_s"] - baseline["best_s"]) / baseline["best_s"] * 100
+      print(
+        f"{res['name']:30s}{res['median_s']:>11.2f}{res['best_s']:>10.2f}"
+        f"{res['xrt_best']:>12.1f}x{delta:>+13.1f}%"
+      )
+    print("=" * 78)
+    print(
+      "Lower 'vs baseline' magnitude = less overhead. Positive = slower than "
+      "baseline; within run-to-run noise (~a few %) means no regression."
+    )
+
+
+if __name__ == "__main__":
+  main()

src/birdnet/acoustic/inference/configs.py

@@ -219,6 +219,15 @@ class OutputConfig:
   show_stats: None | Literal["minimal", "progress", "benchmark"]
   progress_callback: Callable[[AcousticProgressStats], None] | None
 
+  # Fired once per input file as soon as that file is fully processed, with a
+  # single-file result. Unlike ``progress_callback`` this is independent of
+  # ``show_stats`` and enables streaming per-file persistence (e.g. resumable
+  # analysis). Only supported for file inputs. Typed loosely to avoid importing
+  # the (heavy) result module here; the concrete type depends on the session:
+  # ``AcousticFilePredictionResult`` for predictions,
+  # ``AcousticFileEncodingResult`` for encodings.
+  file_completion_callback: Callable[[Any], None] | None = None
+
   @classmethod
   def validate_show_stats(
     cls,

src/birdnet/acoustic/inference/core/consumer.py

@@ -1,9 +1,13 @@
 from __future__ import annotations
 
+import queue
 from multiprocessing import Queue
 from multiprocessing.synchronize import Event
+from pathlib import Path
 from queue import Empty
 
+import numpy as np
+
 from birdnet.acoustic.inference.core.logs import get_logger_from_session
 from birdnet.acoustic.inference.core.tensor import AcousticTensorBase
 
@@ -16,13 +20,38 @@ def __init__(
     worker_queue: Queue,
     tensor: AcousticTensorBase,
     cancel_event: Event,
+    *,
+    n_inputs: int = 0,
+    inputs: list[Path] | None = None,
+    completion_marker_queue: Queue | None = None,
+    completion_dispatch_queue: queue.Queue | None = None,
   ) -> None:
     self._n_workers = n_workers
     self._queue = worker_queue
     self._tensor = tensor
     self._cancel_event = cancel_event
     self._logger = get_logger_from_session(session_id, __name__)
 
+    # Per-file completion tracking (``on_file_complete``). Fully inert unless a
+    # marker queue is provided, so the default hot path is byte-for-byte
+    # unchanged.
+    self._track_completion = completion_marker_queue is not None
+    self._marker_queue = completion_marker_queue
+    self._dispatch_queue = completion_dispatch_queue
+    self._inputs = inputs
+    self._n_inputs = n_inputs
+    if self._track_completion:
+      assert self._dispatch_queue is not None
+      assert self._inputs is not None
+      assert self._n_inputs == len(self._inputs)
+      self._written = np.zeros(n_inputs, dtype=np.int64)
+      self._expected = np.zeros(n_inputs, dtype=np.int64)
+      self._duration = np.zeros(n_inputs, dtype=np.float64)
+      self._invalid = np.zeros(n_inputs, dtype=bool)
+      self._marker_received = np.zeros(n_inputs, dtype=bool)
+      self._n_markers = 0
+      self._pending: set[int] = set()
+
   def _log(self, message: str) -> None:
     self._logger.debug(f"C: {message}")
 
@@ -34,6 +63,9 @@ def __call__(self) -> None:
         "Consumer encountered an exception.", exc_info=e, stack_info=True
       )
       self._cancel_event.set()
+    finally:
+      if self._track_completion:
+        self._end_dispatch()
 
   def _run_main_loop(self) -> None:
     finished_workers = 0
@@ -74,3 +106,82 @@ def _run_main_loop(self) -> None:
         f"Total received: {n_received_predictions}"
       )
       self._tensor.write_block(*block)
+
+      if self._track_completion:
+        # block[0] holds the per-segment input indices of this batch. bincount
+        # is C-speed (unlike np.add.at) so this stays cheap on the hot path.
+        self._written += np.bincount(block[0], minlength=self._n_inputs)
+        self._drain_markers()
+        self._dispatch_ready()
+
+  # -- per-file completion helpers (only used when tracking is enabled) --------
+
+  def _apply_marker(self, marker: tuple[int, int, bool, float]) -> None:
+    idx, n_emitted, is_invalid, duration = marker
+    if self._marker_received[idx]:
+      return
+    self._marker_received[idx] = True
+    self._expected[idx] = n_emitted
+    self._invalid[idx] = is_invalid
+    self._duration[idx] = duration
+    self._n_markers += 1
+    self._pending.add(idx)
+
+  def _drain_markers(self) -> None:
+    assert self._marker_queue is not None
+    while True:
+      try:
+        marker = self._marker_queue.get_nowait()
+      except Empty:
+        break
+      self._apply_marker(marker)
+
+  def _finalize_markers(self) -> None:
+    # After all workers finished, every segment has been written; block until we
+    # have every producer marker so no completed file is missed.
+    assert self._marker_queue is not None
+    while self._n_markers < self._n_inputs:
+      if self._cancel_event.is_set():
+        return
+      try:
+        marker = self._marker_queue.get(timeout=1.0)
+      except Empty:
+        continue
+      self._apply_marker(marker)
+
+  def _dispatch_ready(self) -> None:
+    if not self._pending:
+      return
+    done: list[int] = []
+    for idx in self._pending:
+      if self._invalid[idx]:
+        # Invalid/partial files are reported as zero-detection, matching how the
+        # aggregate result masks unprocessable inputs.
+        self._emit(idx, valid=False)
+        done.append(idx)
+      elif self._written[idx] >= self._expected[idx]:
+        self._emit(idx, valid=True)
+        done.append(idx)
+    for idx in done:
+      self._pending.discard(idx)
+
+  def _emit(self, idx: int, *, valid: bool) -> None:
+    assert self._dispatch_queue is not None
+    assert self._inputs is not None
+    n_segments = int(self._expected[idx]) if valid else 0
+    # Runs on this (consumer) thread, so the tensor is never read and written
+    # concurrently; the copied arrays are what crosses to the dispatcher thread.
+    # The array tuple is tensor-specific (predictions vs embeddings); the
+    # strategy that built the tensor knows how to turn it back into a result.
+    arrays = self._tensor.copy_file_slice(idx, n_segments)  # type: ignore[attr-defined]
+    self._dispatch_queue.put(
+      (self._inputs[idx], arrays, not valid, float(self._duration[idx]))
+    )
+
+  def _end_dispatch(self) -> None:
+    if not self._cancel_event.is_set():
+      self._finalize_markers()
+      self._dispatch_ready()
+    assert self._dispatch_queue is not None
+    # Sentinel tells the dispatcher this run is done.
+    self._dispatch_queue.put(None)