Merge pull request #237 from KumarLabJax/add-purge-cli

add jabs-cli prune command; implement project metadata merging for jabs-init command

Author: gbeane

src/jabs/project/project.py

@@ -146,6 +146,60 @@ def project_paths(self) -> ProjectPaths:
         """get the project paths object for this project"""
         return self._paths
 
+    def get_derived_file_paths(self, video_name: str) -> list[Path]:
+        """Return a list of paths for files derived from a given video.
+
+        Includes:
+          - all files under features/<video base name>/** (recursive)
+          - all files under cache/convex_hulls/<video base name>/** (recursive)
+          - cache/<video base name>_pose_est_v*_cache.h5
+          - predictions/<video base name>.h5
+          - annotations/<video base name>.json
+
+        Excludes:
+          - video file
+          - pose file
+
+        Args:
+            video_name: File name (or key) of the video in this project.
+
+        Returns:
+            List of pathlib.Path objects for all related files.
+        """
+        paths: list[Path] = []
+        base = Path(video_name).with_suffix("").name
+
+        # Feature files (recursive under features/<base>/)
+        feature_root = self._paths.feature_dir / base
+        if feature_root.exists():
+            for p in feature_root.rglob("*"):
+                if p.is_file():
+                    paths.append(p)
+
+        # Cached convex hulls (recursive under cache/convex_hulls/<base>/)
+        if self._paths.cache_dir is not None:
+            ch_root = self._paths.cache_dir / "convex_hulls" / base
+            if ch_root.exists():
+                for p in ch_root.rglob("*"):
+                    if p.is_file():
+                        paths.append(p)
+
+            # Cached pose files: cache/<base>_pose_est_v*_cache.h5
+            for p in self._paths.cache_dir.glob(f"{base}_pose_est_v*_cache.h5"):
+                paths.append(p)
+
+        # Predictions file: predictions/<base>.h5
+        prediction = self._paths.prediction_dir / f"{base}.h5"
+        if prediction.exists():
+            paths.append(prediction)
+
+        # Annotation file
+        annotation = self._paths.annotations_dir / f"{base}.json"
+        if annotation.exists():
+            paths.append(annotation)
+
+        return paths
+
     @property
     def labeler(self) -> str | None:
         """return name of labeler

src/jabs/project/settings_manager.py

@@ -97,49 +97,73 @@ def video_metadata(self, video: str) -> dict:
         """
         return self._project_info["video_files"][video].get("metadata", {})
 
-    def set_project_metadata(self, metadata: dict) -> None:
-        """Set or replace project and per-video metadata.
+    def remove_video_from_project_file(self, video_name: str, sync=True) -> None:
+        """Remove a video entry from the project if it exists.
 
-        Removes any existing project-level and per-video metadata, then sets new metadata as provided.
-        Only the "metadata" field is updated for each video; other fields are preserved.
+        Note: does not remove any associated video or data files from disk, this only
+         removes the entry from the project.json file.
 
         Args:
-            metadata (dict): Dictionary containing new project-level metadata under the "project" key,
-                and per-video metadata under the "videos" key. Example:
+            video_name: Name of the video file to remove.
+            sync: If True, save the project file after removal to sync the on-disk copy. Defaults to True.
+        """
+        video_files = self._project_info.get("video_files", {})
+        video_files.pop(video_name, None)
+
+        if sync:
+            self.save_project_file()
+
+    def set_project_metadata(self, metadata: dict, replace: bool = False) -> None:
+        """Set or merge project and per-video metadata.
+
+        By default, existing metadata is merged: new fields are added and existing fields
+        are updated. If clear_existing=True, all existing metadata is cleared first.
+
+        Args:
+            metadata (dict): Dictionary containing new project-level metadata under the
+                "project" key, and per-video metadata under the "videos" key. Example:
                 {
                     "project": {...},
                     "videos": {
                         "video1": {...},
                         ...
                     }
                 }
-
-                See src/jabs/schema/metadata.py for the expected structure of the metadata.
+            replace (bool): If true, replace existing metadata instead of merge.
+                Defaults to False.
 
         Raises:
             KeyError: If metadata for a video is provided for a video not present in the project.
-        """
-        # Remove existing project-level metadata
-        self._project_info.pop("metadata", None)
 
-        # Remove existing metadata from each video
+        Note:
+            See src/jabs/schema/metadata.py for metadata schema.
+        """
         video_files = self._project_info.get("video_files", {})
-        for video_entry in video_files.values():
-            video_entry.pop("metadata", None)
 
-        # Set new project-level metadata if provided
+        if replace:
+            # Remove all existing metadata
+            self._project_info.pop("metadata", None)
+            for video_entry in video_files.values():
+                video_entry.pop("metadata", None)
+
+        # Merge or replace project-level metadata
         if "project" in metadata:
-            self._project_info["metadata"] = metadata["project"]
+            existing_project_meta = self._project_info.get("metadata", {}) if not replace else {}
+            merged_project_meta = dict(existing_project_meta)
+            merged_project_meta.update(metadata["project"])
+            self._project_info["metadata"] = merged_project_meta
 
-        # Update per-video metadata
+        # Merge or replace per-video metadata
         for video_name, video_metadata in metadata.get("videos", {}).items():
-            # get the existing video entry, raise KeyError if not found
+            if video_name not in video_files:
+                raise KeyError(f"Video '{video_name}' not found in project.")
+
             video_entry = video_files[video_name]
-            if video_metadata:
-                video_entry["metadata"] = video_metadata
-            video_files[video_name] = video_entry
+            existing_video_meta = video_entry.get("metadata", {}) if not replace else {}
+            merged_video_meta = dict(existing_video_meta)
+            merged_video_meta.update(video_metadata)
+            video_entry["metadata"] = merged_video_meta
 
-        # Save changes
         self.save_project_file()
 
     def save_behavior(self, behavior: str, data: dict):

src/jabs/scripts/cli.py

@@ -13,7 +13,7 @@
 from rich.console import Console
 
 from jabs.classifier import Classifier
-from jabs.project import Project, export_training_data
+from jabs.project import Project, export_training_data, get_videos_to_prune
 from jabs.types import ClassifierType
 
 # find out which classifiers are supported in this environment
@@ -156,6 +156,82 @@ def rename_behavior(ctx, directory: Path, old_name: str, new_name: str) -> None:
         jabs_project.rename_behavior(old_name, new_name)
 
 
+@cli.command(name="prune")
+@click.argument(
+    "directory",
+    type=click.Path(
+        exists=True,
+        file_okay=False,
+        dir_okay=True,
+        path_type=Path,
+    ),
+)
+@click.option(
+    "--behavior",
+    type=str,
+    default=None,
+    help="Filter by behavior name. If provided, only videos labeled for this behavior will be retained; otherwise, all videos with any labeled behavior are kept.",
+)
+@click.pass_context
+def prune(ctx, directory: Path, behavior: str | None):
+    """Prune unused videos from a JABS project directory."""
+    if not Project.is_valid_project_directory(directory):
+        raise click.ClickException(f"Invalid JABS project directory: {directory}")
+
+    project = Project(directory)
+    videos_to_prune = get_videos_to_prune(project, behavior)
+
+    if not videos_to_prune:
+        click.echo("No videos to prune.")
+        return
+
+    click.echo(
+        f"Found {len(videos_to_prune)} videos to prune out of {len(project.video_manager.videos)} total videos."
+    )
+    click.echo("The following videos will be removed:")
+    for video_path in videos_to_prune:
+        click.echo(f" - {video_path.video_path.name}")
+
+    confirm = click.confirm("Do you want to proceed with pruning these videos?", default=False)
+
+    if not confirm:
+        click.echo("Pruning cancelled.")
+        return
+
+    for video_paths in videos_to_prune:
+        # get related files that also need to be cleaned up
+        derived_files = project.get_derived_file_paths(video_paths.video_path.name)
+
+        # Remove files, ignore file not found errors
+        try:
+            video_paths.video_path.unlink()
+        except FileNotFoundError:
+            pass
+        except Exception as e:
+            click.echo(f"Warning: failed to delete video or pose file: {e}")
+
+        try:
+            video_paths.pose_path.unlink()
+        except FileNotFoundError:
+            pass
+        except Exception as e:
+            click.echo(f"Warning: failed to delete video or pose file: {e}")
+
+        for file in derived_files:
+            try:
+                file.unlink()
+            except FileNotFoundError:
+                continue
+            except Exception as e:
+                click.echo(f"Warning: failed to delete derived file {file}: {e}")
+
+        # remove from the project.json file
+        project.settings_manager.remove_video_from_project_file(
+            video_paths.video_path.name, sync=False
+        )
+    project.settings_manager.save_project_file()
+
+
 def main():
     """Entry point for the JABS CLI."""
     cli(obj={})

src/jabs/scripts/initialize_project.py

@@ -227,6 +227,7 @@ def main():
     distance_unit = project.feature_manager.distance_unit
     if metadata:
         has_metadata = False
+        replace = True
 
         if project.settings_manager.project_metadata != {}:
             has_metadata = True
@@ -236,14 +237,16 @@ def main():
                 has_metadata = True
                 break
 
-        if has_metadata and not args.force:
+        if has_metadata:
             response = (
-                input("Warning: Project already has metadata. Overwrite? [y/N]: ").strip().lower()
+                input(
+                    "Metadata already exists. Apply new metadata by [M]erge (default) or [R]eplace (clear existing)? [M/r]: "
+                )
+                .strip()
+                .lower()
             )
-            if response != "y":
-                print("Aborting. Use --force to overwrite without prompt.")
-                sys.exit(1)
-        project.settings_manager.set_project_metadata(metadata)
+            replace = response == "r"
+        project.settings_manager.set_project_metadata(metadata, replace=replace)
 
     # iterate over each video and try to pair it with an h5 file
     # this test is quick, don't bother to parallelize