Merge pull request #2279 from samuelgarcia/improve_relative_to

Improve the BaseExtractor.to_dict() relative_to machanism to make it safer on windows.

src/spikeinterface/core/base.py

@@ -15,7 +15,14 @@
 import numpy as np
 
 from .globals import get_global_tmp_folder, is_set_global_tmp_folder
-from .core_tools import check_json, is_dict_extractor, recursive_path_modifier, SIJsonEncoder
+from .core_tools import (
+    check_json,
+    is_dict_extractor,
+    SIJsonEncoder,
+    make_paths_relative,
+    make_paths_absolute,
+    check_paths_relative,
+)
 from .job_tools import _shared_job_kwargs_doc
 
 
@@ -322,6 +329,9 @@ def to_dict(
         to a dictionary. The resulting dictionary can be used to re-initialize the extractor
         through the `load_extractor_from_dict` function.
 
+        In some situations, 'relative_to' is not possible, (for instance different drives on Windows or url-like path), then
+        the relative will ignored and absolute (relative_to=None) will be done instead.
+
         Examples
         --------
         >>> dump_dict = original_extractor.to_dict()
@@ -383,7 +393,8 @@ def to_dict(
             to_dict_kwargs = dict(
                 include_annotations=include_annotations,
                 include_properties=include_properties,
-                relative_to=None,  # '_make_paths_relative' is already recursive!
+                # make_paths_relative() will make the recusrivity later:
+                relative_to=None,
                 folder_metadata=folder_metadata,
                 recursive=recursive,
             )
@@ -414,10 +425,9 @@ def to_dict(
             "module": module,
             "kwargs": kwargs,
             "version": module_version,
-            "relative_paths": (relative_to is not None),
         }
 
-        dump_dict["version"] = module_version  # Can be spikeinterface, spikefores, etc.
+        dump_dict["version"] = module_version  # Can be spikeinterface, spikeforest, etc.
 
         if include_annotations:
             dump_dict["annotations"] = self._annotations
@@ -431,10 +441,20 @@ def to_dict(
             # include only main properties
             dump_dict["properties"] = {k: self._properties.get(k, None) for k in self._main_properties}
 
-        if relative_to is not None:
+        if relative_to is None:
+            dump_dict["relative_paths"] = False
+        else:
             relative_to = Path(relative_to).resolve().absolute()
             assert relative_to.is_dir(), "'relative_to' must be an existing directory"
-            dump_dict = _make_paths_relative(dump_dict, relative_to)
+
+            if check_paths_relative(dump_dict, relative_to):
+                dump_dict["relative_paths"] = True
+                dump_dict = make_paths_relative(dump_dict, relative_to)
+            else:
+                # A warning will be very annoying for end user.
+                # So let's switch back to absolute path, but silently!
+                # warnings.warn("Try to BaseExtractor.to_dict() using relative_to but there is no common folder")
+                dump_dict["relative_paths"] = False
 
         if folder_metadata is not None:
             if relative_to is not None:
@@ -463,7 +483,7 @@ def from_dict(dictionary: dict, base_folder: Optional[Union[Path, str]] = None)
         # for pickle dump relative_path was not in the dict, this ensure compatibility
         if dictionary.get("relative_paths", False):
             assert base_folder is not None, "When  relative_paths=True, need to provide base_folder"
-            dictionary = _make_paths_absolute(dictionary, base_folder)
+            dictionary = make_paths_absolute(dictionary, base_folder)
         extractor = _load_extractor_from_dict(dictionary)
         folder_metadata = dictionary.get("folder_metadata", None)
         if folder_metadata is not None:
@@ -1008,18 +1028,6 @@ def save_to_zarr(
         return cached
 
 
-def _make_paths_relative(d, relative) -> dict:
-    relative = str(Path(relative).resolve().absolute())
-    func = lambda p: os.path.relpath(str(p), start=relative)
-    return recursive_path_modifier(d, func, target="path", copy=True)
-
-
-def _make_paths_absolute(d, base):
-    base = Path(base)
-    func = lambda p: str((base / p).resolve().absolute())
-    return recursive_path_modifier(d, func, target="path", copy=True)
-
-
 def _load_extractor_from_dict(dic) -> BaseExtractor:
     """
     Convert a dictionary into an instance of BaseExtractor or its subclass.

src/spikeinterface/core/binaryfolder.py

@@ -3,9 +3,8 @@
 
 import numpy as np
 
-from .base import _make_paths_absolute
 from .binaryrecordingextractor import BinaryRecordingExtractor
-from .core_tools import define_function_from_class
+from .core_tools import define_function_from_class, make_paths_absolute
 
 
 class BinaryFolderRecording(BinaryRecordingExtractor):
@@ -40,7 +39,7 @@ def __init__(self, folder_path):
 
         assert d["relative_paths"]
 
-        d = _make_paths_absolute(d, folder_path)
+        d = make_paths_absolute(d, folder_path)
 
         BinaryRecordingExtractor.__init__(self, **d["kwargs"])
 

src/spikeinterface/core/core_tools.py

@@ -1,4 +1,4 @@
-from pathlib import Path
+from pathlib import Path, WindowsPath
 from typing import Union
 import os
 import sys
@@ -851,6 +851,118 @@ def recursive_path_modifier(d, func, target="path", copy=True) -> dict:
                     raise ValueError(f"{k} key for path  must be str or list[str]")
 
 
+def _get_paths_list(d):
+    # this explore a dict and get all paths flatten in a list
+    # the trick is to use a closure func called by recursive_path_modifier()
+    path_list = []
+
+    def append_to_path(p):
+        path_list.append(p)
+
+    recursive_path_modifier(d, append_to_path, target="path", copy=True)
+    return path_list
+
+
+def _relative_to(p, relative_folder):
+    # custum os.path.relpath() with more checks
+
+    relative_folder = Path(relative_folder).resolve()
+    p = Path(p).resolve()
+    # the as_posix transform \\ to / on window which make better json files
+    rel_to = os.path.relpath(p.as_posix(), start=relative_folder.as_posix())
+    return Path(rel_to).as_posix()
+
+
+def check_paths_relative(input_dict, relative_folder) -> bool:
+    """
+    Check if relative path is possible to be applied on a dict describing an BaseExtractor.
+
+    For instance on windows, if some data are on a drive "D:/" and the folder is on drive "C:/" it returns False.
+
+    Parameters
+    ----------
+    input_dict: dict
+        A dict describing an extactor obtained by BaseExtractor.to_dict()
+    relative_folder: str or Path
+        The folder to be relative to.
+
+    Returns
+    -------
+    relative_possible: bool
+    """
+    path_list = _get_paths_list(input_dict)
+    relative_folder = Path(relative_folder).resolve().absolute()
+    not_possible = []
+    for p in path_list:
+        p = Path(p)
+        # check path is not an URL
+        if "http" in str(p):
+            not_possible.append(p)
+            continue
+
+        # If windows path check have same drive
+        if isinstance(p, WindowsPath) and isinstance(relative_folder, WindowsPath):
+            # check that on same drive
+            # important note : for UNC path on window the "//host/shared" is the drive
+            if p.resolve().absolute().drive != relative_folder.drive:
+                not_possible.append(p)
+                continue
+
+        # check relative is possible
+        try:
+            p2 = _relative_to(p, relative_folder)
+        except ValueError:
+            not_possible.append(p)
+            continue
+
+    return len(not_possible) == 0
+
+
+def make_paths_relative(input_dict, relative_folder) -> dict:
+    """
+    Recursively transform a dict describing an BaseExtractor to make every path relative to a folder.
+
+    Parameters
+    ----------
+    input_dict: dict
+        A dict describing an extactor obtained by BaseExtractor.to_dict()
+    relative_folder: str or Path
+        The folder to be relative to.
+
+    Returns
+    -------
+    output_dict: dict
+        A copy of the input dict with modified paths.
+    """
+    relative_folder = Path(relative_folder).resolve().absolute()
+    func = lambda p: _relative_to(p, relative_folder)
+    output_dict = recursive_path_modifier(input_dict, func, target="path", copy=True)
+    return output_dict
+
+
+def make_paths_absolute(input_dict, base_folder):
+    """
+    Recursively transform a dict describing an BaseExtractor to make every path absolute given a base_folder.
+
+    Parameters
+    ----------
+    input_dict: dict
+        A dict describing an extactor obtained by BaseExtractor.to_dict()
+    base_folder: str or Path
+        The folder to be relative to.
+
+    Returns
+    -------
+    output_dict: dict
+        A copy of the input dict with modified paths.
+    """
+    base_folder = Path(base_folder)
+    # use as_posix instead of str to make the path unix like even on window
+    func = lambda p: (base_folder / p).resolve().absolute().as_posix()
+    output_dict = recursive_path_modifier(input_dict, func, target="path", copy=True)
+    return output_dict
+
+
 def recursive_key_finder(d, key):
     # Find all values for a key on a dictionary, even if nested
     for k, v in d.items():

src/spikeinterface/core/npyfoldersnippets.py

@@ -3,9 +3,8 @@
 
 import numpy as np
 
-from .base import _make_paths_absolute
 from .npysnippetsextractor import NpySnippetsExtractor
-from .core_tools import define_function_from_class
+from .core_tools import define_function_from_class, make_paths_absolute
 
 
 class NpyFolderSnippets(NpySnippetsExtractor):
@@ -41,7 +40,7 @@ def __init__(self, folder_path):
 
         assert d["relative_paths"]
 
-        d = _make_paths_absolute(d, folder_path)
+        d = make_paths_absolute(d, folder_path)
 
         NpySnippetsExtractor.__init__(self, **d["kwargs"])
 

src/spikeinterface/core/sortingfolder.py

@@ -3,10 +3,9 @@
 
 import numpy as np
 
-from .base import _make_paths_absolute
 from .basesorting import BaseSorting, BaseSortingSegment
 from .npzsortingextractor import NpzSortingExtractor
-from .core_tools import define_function_from_class
+from .core_tools import define_function_from_class, make_paths_absolute
 from .numpyextractors import NumpySortingSegment
 
 
@@ -107,7 +106,7 @@ def __init__(self, folder_path):
 
         assert d["relative_paths"]
 
-        d = _make_paths_absolute(d, folder_path)
+        d = make_paths_absolute(d, folder_path)
 
         NpzSortingExtractor.__init__(self, **d["kwargs"])