Video files organize

dandi/cli/cmd_organize.py

@@ -34,6 +34,22 @@
     default="auto",
     show_default=True,
 )
+@click.option(
+    "--modify-external-file-fields",
+    is_flag=True,
+    default=False,
+    help="using this options will rewrite the 'external_file' argument of the "
+    "ImageSeries in the nwb file. The value written will correspond to the"
+    " new location of the video files after being organized. This option will "
+    "work only if --files-mode is copy/move since the original nwb file is "
+    "modified",
+)
+@click.option(
+    "--media-files-mode",
+    type=click.Choice(["copy", "move", "symlink", "hardlink"]),
+    default=None,
+    help="Supply one of copy/move/symlink/hardlink as the value",
+)
 @click.argument("paths", nargs=-1, type=click.Path(exists=True))
 @devel_debug_option()
 @map_to_click_exceptions
@@ -43,6 +59,8 @@ def organize(
     invalid="fail",
     files_mode="auto",
     devel_debug=False,
+    modify_external_file_fields=False,
+    media_files_mode=None,
 ):
     """(Re)organize files according to the metadata.
 
@@ -80,11 +98,13 @@ def organize(
     from ..dandiset import Dandiset
     from ..metadata import get_metadata
     from ..organize import (
+        _create_external_file_names,
         create_unique_filenames_from_metadata,
         detect_link_type,
         filter_invalid_metadata_rows,
+        organize_external_files,
     )
-    from ..pynwb_utils import ignore_benign_pynwb_warnings
+    from ..pynwb_utils import ignore_benign_pynwb_warnings, rename_nwb_external_files
     from ..utils import Parallel, copy_file, delayed, find_files, load_jsonl, move_file
 
     in_place = False  # If we deduce that we are organizing in-place
@@ -104,6 +124,11 @@ def act(func, *args, **kwargs):
             lgr.debug("%s %s %s", func.__name__, args, kwargs)
             return func(*args, **kwargs)
 
+    if modify_external_file_fields and files_mode not in ["copy", "move"]:
+        raise ValueError(
+            "files_mode needs to be one of 'copy/move' for the rewrite option to work"
+        )
+
     if dandiset_path is None:
         dandiset = Dandiset.find(os.curdir)
         if not dandiset:
@@ -140,7 +165,7 @@ def act(func, *args, **kwargs):
                 "Only 'dry' or 'move' mode could be used to operate in-place "
                 "within a dandiset (no paths were provided)"
             )
-        lgr.info(f"We will organize {dandiset_path} in-place")
+        lgr.info("We will organize %s in-place", dandiset_path)
         in_place = True
         paths = dandiset_path
 
@@ -214,6 +239,36 @@ def _get_metadata(path):
 
     metadata = create_unique_filenames_from_metadata(metadata)
 
+    # update metadata with external_file information:
+    external_files_missing_in_nwbfiles = [
+        len(m["external_file_objects"]) == 0 for m in metadata
+    ]
+
+    if all(external_files_missing_in_nwbfiles) and modify_external_file_fields:
+        lgr.warning(
+            "modify_external_file_fields specified but no external_files found "
+            "linked to any nwbfile found in %s",
+            paths,
+        )
+    elif (
+        not all(external_files_missing_in_nwbfiles) and not modify_external_file_fields
+    ):
+        raise ValueError(
+            "modify_external_file_fields option not specified but found "
+            "external video files linked to the nwbfiles "
+            f"""{[metadata[no]['path']
+                  for no, a in enumerate(external_files_missing_in_nwbfiles)
+                  if not a]}"""
+        )
+
+    if modify_external_file_fields and media_files_mode is None:
+        media_files_mode = "symlink"
+        lgr.warning(
+            "media_files_mode not specified, setting to recommended mode: 'symlink' "
+        )
+
+    metadata = _create_external_file_names(metadata)
+
     # Verify first that the target paths do not exist yet, and fail if they do
     # Note: in "simulate" mode we do early check as well, so this would be
     # duplicate but shouldn't hurt
@@ -313,10 +368,15 @@ def _get_metadata(path):
             if op.exists(d):
                 try:
                     os.rmdir(d)
-                    lgr.info(f"Removed empty directory {d}")
+                    lgr.info("Removed empty directory %s", d)
                 except Exception as exc:
                     lgr.debug("Failed to remove directory %s: %s", d, exc)
 
+    # create video file name and re write nwb file external files:
+    if modify_external_file_fields:
+        rename_nwb_external_files(metadata, dandiset_path)
+        organize_external_files(metadata, dandiset_path, media_files_mode)
+
     def msg_(msg, n, cond=None):
         if hasattr(n, "__len__"):
             n = len(n)

dandi/consts.py

@@ -148,6 +148,9 @@ class DandiInstance(NamedTuple):
 #: of retries)
 RETRY_STATUSES = (500, 502, 503, 504)
 
+VIDEO_FILE_EXTENSIONS = [".mp4", ".avi", ".wmv", ".mov", ".flv"]
+VIDEO_FILE_MODULES = ["processing", "acquisition"]
+
 #: Maximum allowed depth of a Zarr directory tree
 MAX_ZARR_DEPTH = 5
 

dandi/organize.py

@@ -9,13 +9,15 @@
 import os.path as op
 from pathlib import Path
 import re
+from typing import List
+import uuid
 
 import numpy as np
 
 from . import get_logger
 from .exceptions import OrganizeImpossibleError
 from .pynwb_utils import get_neurodata_types_to_modalities_map, get_object_id
-from .utils import ensure_datetime, flattened, yaml_load
+from .utils import copy_file, ensure_datetime, flattened, move_file, yaml_load
 
 lgr = get_logger()
 
@@ -172,6 +174,83 @@ def create_unique_filenames_from_metadata(metadata):
     return metadata
 
 
+def _create_external_file_names(metadata: list) -> List[dict]:
+    """
+    Renames the external_file attribute in an ImageSeries according to the rule:
+    Example, the Initial name of file:
+        external_file = [name1.mp4]
+    rename to:
+        external_file = [dandiset-path-of-nwbfile/
+                dandi-renamed-nwbfile_name(folder without extension .nwb)/
+                f'{ImageSeries.object_id}_external_file_0.mp4'
+    This is stored in a new field in the
+    metadata['external_file_objects'][0]['external_files_renamed']
+    Parameters
+    ----------
+    metadata: list
+        list of metadata dictionaries created during the call to pynwb_utils._get_pynwb_metadata
+    Returns
+    -------
+    metadata: list
+        updated list of metadata dictionaries
+    """
+    metadata = deepcopy(metadata)
+    for meta in metadata:
+        if "dandi_path" not in meta or "external_file_objects" not in meta:
+            continue
+        nwb_folder_name = op.splitext(op.basename(meta["dandi_path"]))[0]
+        for ext_file_dict in meta["external_file_objects"]:
+            renamed_path_list = []
+            uuid_str = ext_file_dict.get("id", str(uuid.uuid4()))
+            for no, ext_file in enumerate(ext_file_dict["external_files"]):
+                renamed = op.join(
+                    nwb_folder_name, f"{uuid_str}_external_file_{no}{ext_file.suffix}"
+                )
+                renamed_path_list.append(str(renamed))
+            ext_file_dict["external_files_renamed"] = renamed_path_list
+    return metadata
+
+
+def organize_external_files(
+    metadata: list, dandiset_path: str, files_mode: str
+) -> None:
+    """
+    Organizes the external_files into the new Dandiset folder structure.
+
+    Parameters
+    ----------
+    metadata: list
+        list of metadata dictionaries created during the call to pynwb_utils._get_pynwb_metadata
+    dandiset_path: str
+        full path of the main dandiset folder.
+    files_mode: str
+        one of "symlink", "copy", "move", "hardlink"
+
+    """
+    for e in metadata:
+        for ext_file_dict in e["external_file_objects"]:
+            for no, (name_old, name_new) in enumerate(
+                zip(
+                    ext_file_dict["external_files"],
+                    ext_file_dict["external_files_renamed"],
+                )
+            ):
+                new_path = op.join(dandiset_path, op.dirname(e["dandi_path"]), name_new)
+                name_old_str = str(name_old)
+                if not op.exists(op.dirname(new_path)):
+                    os.makedirs(op.dirname(new_path))
+                if files_mode == "symlink":
+                    os.symlink(name_old_str, new_path)
+                elif files_mode == "hardlink":
+                    os.link(name_old_str, new_path)
+                elif files_mode == "copy":
+                    copy_file(name_old_str, new_path)
+                elif files_mode == "move":
+                    move_file(name_old_str, new_path)
+                else:
+                    raise NotImplementedError(files_mode)
+
+
 def _assign_obj_id(metadata, non_unique):
     msg = "%d out of %d paths are not unique" % (len(non_unique), len(metadata))
 

dandi/pynwb_utils.py

@@ -17,6 +17,8 @@
 
 from . import __version__, get_logger
 from .consts import (
+    VIDEO_FILE_EXTENSIONS,
+    VIDEO_FILE_MODULES,
     metadata_nwb_computed_fields,
     metadata_nwb_file_fields,
     metadata_nwb_subject_fields,
@@ -230,9 +232,93 @@ def _get_pynwb_metadata(path: Union[str, Path]) -> Dict[str, Any]:
             key = f[len("number_of_") :]
             out[f] = len(getattr(nwb, key, []) or [])
 
+        # get external_file data:
+        out["external_file_objects"] = _get_image_series(nwb)
+
+    return out
+
+
+def _get_image_series(nwb: pynwb.NWBFile) -> List[dict]:
+    """
+    This method supports _get_pynwb_metadata() in retrieving all ImageSeries
+    related metadata from an open nwb file.
+    Specifically it pulls out the ImageSeries uuid, name and all the
+    externally linked files
+    named under the argument 'external_file'.
+    Parameters
+    ----------
+    nwb: pynwb.NWBFile
+    Returns
+    -------
+    out: List[dict]
+        list of dicts : [{id: <ImageSeries uuid>, name: <IMageSeries name>,
+        external_files=[ImageSeries.external_file]}]
+        if no ImageSeries found in the given modules to check, then it returns an empty list.
+    """
+    out = []
+    for module_name in VIDEO_FILE_MODULES:
+        module_cont = getattr(nwb, module_name)
+        for name, ob in module_cont.items():
+            if isinstance(ob, pynwb.image.ImageSeries) and ob.external_file is not None:
+                out_dict = dict(id=ob.object_id, name=ob.name, external_files=[])
+                for ext_file in ob.external_file:
+                    if Path(ext_file).suffix in VIDEO_FILE_EXTENSIONS:
+                        out_dict["external_files"].append(Path(ext_file))
+                    else:
+                        lgr.warning(
+                            "external file %s should be one of: %s",
+                            ext_file,
+                            VIDEO_FILE_EXTENSIONS,
+                        )
+                out.append(out_dict)
     return out
 
 
+def rename_nwb_external_files(metadata: list, dandiset_path: str) -> None:
+    """
+    This method, renames the external_file attribute in an ImageSeries datatype in an open nwb file.
+    It pulls information about the ImageSEries objects
+    from metadata: metadata["external_file_objects"]
+    populated during _get_pynwb_metadata() call.
+
+    Parameters
+    ----------
+    metadata: List[dict]
+        list of dictionaries containing the metadata gathered from the nwbfile
+    dandiset_path: str
+        base path of dandiset
+    """
+    for meta in metadata:
+        if not all(i in meta for i in ["path", "dandi_path", "external_file_objects"]):
+            lgr.warning(
+                "could not rename external files, update metadata"
+                'with "path", "dandi_path", "external_file_objects"'
+            )
+            return
+        dandiset_nwbfile_path = op.join(dandiset_path, meta["dandi_path"])
+        with NWBHDF5IO(dandiset_nwbfile_path, mode="r+", load_namespaces=True) as io:
+            nwb = io.read()
+            for ext_file_dict in meta["external_file_objects"]:
+                # retrieve nwb neurodata object of the given object id:
+                container_list = [
+                    child
+                    for child in nwb.children
+                    if ext_file_dict["id"] == child.object_id
+                ]
+                if len(container_list) == 0:
+                    continue
+                else:
+                    container = container_list[0]
+                # rename all external files:
+                for no, (name_old, name_new) in enumerate(
+                    zip(
+                        ext_file_dict["external_files"],
+                        ext_file_dict["external_files_renamed"],
+                    )
+                ):
+                    container.external_file[no] = str(name_new)
+
+
 @validate_cache.memoize_path
 def validate(path: Union[str, Path], devel_debug: bool = False) -> List[str]:
     """Run validation on a file and return errors