Continue interrupted analysis

birdnet_analyzer/analyze/core.py

@@ -12,11 +12,9 @@
     from birdnet.acoustic.inference.core.perf_tracker import (
         AcousticProgressStats,
     )
-    from birdnet.acoustic.inference.core.prediction.prediction_result import (
-        AcousticResultBase,
-    )
     from birdnet.globals import ACOUSTIC_MODEL_VERSIONS, MODEL_LANGUAGES
 
+    from birdnet_analyzer.analyze.resume import RunMetadata
     from birdnet_analyzer.config import ADDITIONAL_COLUMNS, RESULT_TYPES
 
 
@@ -103,13 +101,50 @@ def analyze(
     Notes:
         - The function ensures the BirdNET model is available before analysis.
         - Analysis parameters are saved to a file in the output directory.
+        - Directory analyses are resumable: per-file results are journaled to
+          ``<output>/.birdnet-resume/`` as they complete, an interrupted run
+          picks up where it left off (if the analysis parameters match), and
+          the journal is removed after outputs are written. When resuming a
+          run that had already completed every file, no new inference happens
+          and ``None`` is returned instead of a prediction result.
     """
     import birdnet_analyzer.config as cfg
-    from birdnet_analyzer.model_utils import run_geomodel, run_inference
+    from birdnet_analyzer.analyze.resume import ResumeJournal, RunMetadata
+    from birdnet_analyzer.model_utils import (
+        run_geomodel,
+        run_inference,
+        supports_on_file_complete,
+    )
     from birdnet_analyzer.utils import save_params_to_file
 
     species_list_file = slist if isinstance(slist, (str, Path)) else ""
     rtypes: list[RESULT_TYPES] = [rtype] if isinstance(rtype, str) else rtype
+    resume_params = _resume_fingerprint_params(
+        audio_input=audio_input,
+        model=model,
+        birdnet=birdnet,
+        min_conf=min_conf,
+        classifier=classifier,
+        cc_species_list=cc_species_list,
+        lat=lat,
+        lon=lon,
+        week=week,
+        slist=slist,
+        sensitivity=sensitivity,
+        overlap=overlap,
+        fmin=fmin,
+        fmax=fmax,
+        audio_speed=audio_speed,
+        top_n=top_n,
+        sf_thresh=sf_thresh,
+        locale=locale,
+    )
+
+    if not output:
+        if os.path.isfile(audio_input):
+            output = os.path.dirname(audio_input)
+        else:
+            output = audio_input
 
     if lat is not None and lon is not None:
         if slist is not None:
@@ -122,51 +157,84 @@ def analyze(
             lat, lon, week=week, language=locale, threshold=sf_thresh
         ).to_set()
 
-    predictions = run_inference(
-        audio_input,
-        model=model,
-        top_k=top_n,
-        batch_size=batch_size,
-        prefetch_ratio=3,
-        overlap_duration_s=overlap,
-        bandpass_fmin=fmin,
-        bandpass_fmax=fmax,
-        sigmoid_sensitivity=sensitivity,
-        speed=audio_speed,
-        min_confidence=min_conf,
-        custom_species_list=slist,
-        label_language=locale,
-        classifier=classifier,
-        cc_species_list=cc_species_list,
-        version=birdnet,
-        callback=on_update,
-        n_workers=n_workers,
-        n_producers=n_producers,
-    )
+    # For directory analyses, keep a crash-safe journal of per-file results in
+    # the output directory so an interrupted run can be resumed: files that
+    # already have a stored result are skipped, everything else is analyzed and
+    # persisted as it completes.
+    journal = None
+    input_files: list[Path] = []
+    inference_input = audio_input
+
+    if not _return_only and os.path.isdir(audio_input) and supports_on_file_complete():
+        from birdnet.acoustic.inference.configs import InferenceConfig
+
+        input_files = InferenceConfig.validate_input_files(audio_input)
+        journal = ResumeJournal.open(
+            output, resume_params, n_files_total=len(input_files)
+        )
+        completed = journal.completed_subset(input_files)
+        inference_input = [f for f in input_files if f not in completed]
+
+    predictions = None
+
+    if inference_input:
+        predictions = run_inference(
+            inference_input,
+            model=model,
+            top_k=top_n,
+            batch_size=batch_size,
+            prefetch_ratio=3,
+            overlap_duration_s=overlap,
+            bandpass_fmin=fmin,
+            bandpass_fmax=fmax,
+            sigmoid_sensitivity=sensitivity,
+            speed=audio_speed,
+            min_confidence=min_conf,
+            custom_species_list=slist,
+            label_language=locale,
+            classifier=classifier,
+            cc_species_list=cc_species_list,
+            version=birdnet,
+            callback=on_update,
+            n_workers=n_workers,
+            n_producers=n_producers,
+            on_file_complete=journal.on_file_complete if journal else None,
+        )
 
     if _return_only:
         return predictions
 
+    if predictions is not None:
+        meta = RunMetadata.from_result(predictions)
+        df = predictions.to_dataframe()
+    else:
+        # Everything was already completed by an interrupted run; outputs are
+        # built entirely from the journal.
+        meta = journal.metadata() if journal else None
+        df = None
+
+        if meta is None:
+            raise RuntimeError(
+                "Resume state in the output directory is incomplete. Delete "
+                f"'{Path(output) / '.birdnet-resume'}' and run the analysis again."
+            )
+
+    if journal is not None:
+        df = journal.combined_dataframe(df, input_files)
+
     audio_input_path: Path = Path(audio_input)
-    df = predictions.to_dataframe()
     df = _merge_consecutive_segments(
-        df, merge_consecutive, hop_size=predictions.hop_duration_s
+        df, merge_consecutive, hop_size=meta.hop_duration_s
     )
 
-    if not output:
-        if os.path.isfile(audio_input):
-            output = os.path.dirname(audio_input)
-        else:
-            output = audio_input
-
     if split_tables:
         _split_tables(
             df,
             audio_input_path,
             Path(output),
             fmin,
             fmax,
-            predictions,
+            meta,
             audio_speed,
             rtypes,
             additional_columns,
@@ -184,8 +252,8 @@ def analyze(
                 df,
                 fmin,
                 fmax,
-                predictions.model_fmin,
-                predictions.model_fmax,
+                meta.model_fmin,
+                meta.model_fmax,
                 audio_speed,
                 Path(output) / cfg.OUTPUT_RAVEN_FILENAME,
             )
@@ -202,7 +270,7 @@ def analyze(
                 min_conf=min_conf,
                 sensitivity=sensitivity,
                 species_list_file=species_list_file,
-                model_path=predictions.model_path,
+                model_path=meta.model_path,
             )
 
         if "kaleidoscope" in rtypes:
@@ -223,7 +291,7 @@ def analyze(
                 min_conf=min_conf,
                 sensitivity=sensitivity,
                 species_list_file=species_list_file,
-                model_path=predictions.model_path,
+                model_path=meta.model_path,
             )
 
     if save_params:
@@ -259,8 +327,8 @@ def analyze(
             (
                 model,
                 birdnet,
-                predictions.segment_duration_s,
-                predictions.model_sr,
+                meta.segment_duration_s,
+                meta.model_sr,
                 overlap,
                 fmin,
                 fmax,
@@ -285,16 +353,39 @@ def analyze(
             ),
         )
 
+    if journal is not None:
+        journal.finalize()
+
     return predictions
 
 
+def _resume_fingerprint_params(**params) -> dict:
+    """Normalize the parameters that identify a resumable run.
+
+    Only parameters that affect which detections are produced belong here;
+    output formatting (rtype, merge_consecutive, split_tables, ...) may change
+    between an interrupted run and its resume.
+    """
+    slist = params["slist"]
+
+    if slist is None or isinstance(slist, (str, Path)):
+        params["slist"] = str(slist) if slist is not None else None
+    else:
+        params["slist"] = sorted(map(str, slist))
+
+    params["audio_input"] = os.path.normcase(
+        os.path.normpath(os.path.abspath(params["audio_input"]))
+    )
+    return params
+
+
 def _split_tables(
     df: pd.DataFrame,
     audio_input_path: Path,
     output: Path,
     bandpass_fmin: int,
     bandpass_fmax: int,
-    predictions: AcousticResultBase,
+    meta: RunMetadata,
     audio_speed: float,
     rtypes: Sequence[str],
     additional_columns,
@@ -321,8 +412,8 @@ def _split_tables(
                 df_file,
                 bandpass_fmin,
                 bandpass_fmax,
-                predictions.model_fmin,
-                predictions.model_fmax,
+                meta.model_fmin,
+                meta.model_fmax,
                 audio_speed,
                 output / (file_shorthand + ".BirdNET.selection.table.txt"),
             )
@@ -339,7 +430,7 @@ def _split_tables(
                 min_conf=min_conf,
                 sensitivity=sensitivity,
                 species_list_file=species_list_file,
-                model_path=predictions.model_path,
+                model_path=meta.model_path,
             )
 
         if "kaleidoscope" in rtypes:
@@ -364,7 +455,7 @@ def _split_tables(
                 min_conf=min_conf,
                 sensitivity=sensitivity,
                 species_list_file=species_list_file,
-                model_path=predictions.model_path,
+                model_path=meta.model_path,
             )