Search_for_wildcards function updated to add @DATETO@ functionality

datashuttle/utils/folders.py

@@ -17,11 +17,14 @@
     from datashuttle.utils.custom_types import TopLevelFolder
 
 import glob
+import re
+from datetime import datetime
 from pathlib import Path
 
 from datashuttle.configs import canonical_folders, canonical_tags
 from datashuttle.utils import ssh, utils, validation
 from datashuttle.utils.custom_exceptions import NeuroBlueprintError
+from datashuttle.utils.utils import get_values_from_bids_formatted_name
 
 # -----------------------------------------------------------------------------
 # Create Folders
@@ -401,27 +404,65 @@ def search_for_wildcards(
     """
     new_all_names: List[str] = []
     for name in all_names:
-        if canonical_tags.tags("*") in name:
-            name = name.replace(canonical_tags.tags("*"), "*")
-
-            matching_names: List[str]
+        if canonical_tags.tags("*") in name or "@DATETO@" in name:
+            search_str = name.replace(canonical_tags.tags("*"), "*")
+            # If a date-range tag is present, extract dates and update the search string.
+            if "@DATETO@" in name:
+                m = re.search(r"(\d{8})@DATETO@(\d{8})", name)
+                if not m:
+                    raise ValueError(
+                        "Invalid date range format in name: " + name
+                    )
+                start_str, end_str = m.groups()
+                try:
+                    start_date = datetime.strptime(start_str, "%Y%m%d")
+                    end_date = datetime.strptime(end_str, "%Y%m%d")
+                except ValueError as e:
+                    raise ValueError("Invalid date in date range: " + str(e))
+                # Replace the date-range substring with "date-*"
+                search_str = re.sub(r"\d{8}@DATETO@\d{8}", "date-*", name)
+            # Use the helper function to perform the glob search.
             if sub:
-                matching_names = search_sub_or_ses_level(  # type: ignore
-                    cfg, base_folder, local_or_central, sub, search_str=name
+                matching_names: List[str] = search_sub_or_ses_level(
+                    cfg,
+                    base_folder,
+                    local_or_central,
+                    sub,
+                    search_str=search_str,
                 )[0]
             else:
-                matching_names = search_sub_or_ses_level(  # type: ignore
-                    cfg, base_folder, local_or_central, search_str=name
+                matching_names = search_sub_or_ses_level(
+                    cfg, base_folder, local_or_central, search_str=search_str
                 )[0]
-
+            # If a date-range tag was provided, further filter the results.
+            if "@DATETO@" in name:
+                filtered_names: List[str] = []
+                for candidate in matching_names:
+                    candidate_basename = (
+                        candidate
+                        if isinstance(candidate, str)
+                        else candidate.name
+                    )
+                    values_list = get_values_from_bids_formatted_name(
+                        [candidate_basename], "date"
+                    )
+                    if not values_list:
+                        continue
+                    candidate_date_str = values_list[0]
+                    try:
+                        candidate_date = datetime.strptime(
+                            candidate_date_str, "%Y%m%d"
+                        )
+                    except ValueError:
+                        continue
+                    if start_date <= candidate_date <= end_date:
+                        filtered_names.append(candidate)
+                matching_names = filtered_names
             new_all_names += matching_names
         else:
             new_all_names += [name]
-
-    new_all_names = list(
-        set(new_all_names)
-    )  # remove duplicate names in case of wildcard overlap
-
+    # Remove duplicates in case of wildcard overlap.
+    new_all_names = list(set(new_all_names))
     return new_all_names
 
 
@@ -440,7 +481,7 @@ def search_sub_or_ses_level(
     search_str: str = "*",
     verbose: bool = True,
     return_full_path: bool = False,
-) -> Tuple[List[str] | List[Path], List[str]]:
+) -> Tuple[Union[List[str], List[Path]], List[str]]:
     """
     Search project folder at the subject or session level.
     Only returns folders

datashuttle/utils/validation.py

@@ -321,7 +321,7 @@ def replace_tags_in_regexp(regexp: str) -> str:
     Note `replace_date_time_tags_in_name()` operates in place on a list.
     """
     regexp_list = [regexp]
-    date_regexp = "\d\d\d\d\d\d\d\d"
+    date_regexp = r"\d{8}"
     time_regexp = "\d\d\d\d\d\d"
 
     formatting.replace_date_time_tags_in_name(

tests/test_date_search_range.py

@@ -0,0 +1,134 @@
+import glob
+import os
+import re
+import shutil
+import tempfile
+from pathlib import Path
+from typing import List
+
+import pytest
+
+from datashuttle.utils.folders import search_for_wildcards
+
+
+# Dummy implementation for canonical_tags
+class DummyCanonicalTags:
+    @staticmethod
+    def tags(x: str) -> str:
+        if x == "*":
+            return "@*@"
+        return x
+
+
+# Patch canonical_tags so that tags("*") returns "@*@"
+@pytest.fixture(autouse=True)
+def patch_canonical_tags(monkeypatch):
+    from datashuttle.configs import canonical_tags
+
+    monkeypatch.setattr(canonical_tags, "tags", DummyCanonicalTags.tags)
+
+
+# Dummy implementation for search_sub_or_ses_level that simply performs globbing.
+def dummy_search_sub_or_ses_level(
+    cfg, base_folder: Path, local_or_central: str, *args, search_str: str
+):
+    pattern = os.path.join(str(base_folder), search_str)
+    matches: List[str] = sorted(glob.glob(pattern))
+    return (matches,)
+
+
+# Patch search_sub_or_ses_level in the module where search_for_wildcards is defined.
+@pytest.fixture(autouse=True)
+def patch_search_sub_or_ses_level(monkeypatch):
+    monkeypatch.setattr(
+        "datashuttle.utils.folders.search_sub_or_ses_level",
+        dummy_search_sub_or_ses_level,
+    )
+
+
+# Dummy implementation for get_values_from_bids_formatted_name.
+def dummy_get_values_from_bids_formatted_name(name: str, key: str) -> dict:
+    # Expect name format: "sub-01_date-YYYYMMDD"
+    m = re.search(r"date-(\d{8})", name)
+    if m:
+        return {key: m.group(1)}
+    return {}
+
+
+# Patch get_values_from_bids_formatted_name.
+@pytest.fixture(autouse=True)
+def patch_get_values_from_bids(monkeypatch):
+    monkeypatch.setattr(
+        "datashuttle.utils.utils.get_values_from_bids_formatted_name",
+        dummy_get_values_from_bids_formatted_name,
+    )
+
+
+# Fixture to create a temporary directory with a simulated folder structure.
+@pytest.fixture
+def temp_project_dir() -> Path:  # type: ignore
+    temp_dir = Path(tempfile.mkdtemp())
+    # Create folders with names in the format "sub-01_date-YYYYMMDD"
+    folder_dates = [
+        "20250305",
+        "20250306",
+        "20250307",
+        "20250308",
+        "20250309",
+        "20250310",
+    ]
+    for date_str in folder_dates:
+        folder_name = f"sub-01_date-{date_str}"
+        os.mkdir(temp_dir / folder_name)
+    yield temp_dir
+    shutil.rmtree(temp_dir)
+
+
+def test_date_range_wildcard(temp_project_dir: Path):
+    """
+    When given a date-range wildcard pattern like "sub-01_20250306@DATETO@20250309",
+    only folders whose embedded date falls between 20250306 and 20250309 (inclusive)
+    should be returned.
+    """
+
+    class Configs:
+        pass
+
+    cfg = Configs()
+    base_folder = temp_project_dir
+    local_or_central = "local"
+    pattern = "sub-01_20250306@DATETO@20250309"
+    result = search_for_wildcards(
+        cfg, base_folder, local_or_central, [pattern]
+    )
+
+    # Extract the dates from the returned folder names.
+    found_dates = set()
+    for folder in result:
+        basename = os.path.basename(folder)
+        m = re.search(r"date-(\d{8})", basename)
+        if m:
+            found_dates.add(m.group(1))
+
+    expected_dates = {"20250306", "20250307", "20250308", "20250309"}
+    assert found_dates == expected_dates
+
+
+def test_simple_wildcard(temp_project_dir: Path):
+    """
+    When given a simple wildcard pattern like "sub-01_@*@",
+    all folders should be returned.
+    """
+
+    class Configs:
+        pass
+
+    cfg = Configs()
+    base_folder = temp_project_dir
+    local_or_central = "local"
+    pattern = "sub-01_@*@"
+    result = search_for_wildcards(
+        cfg, base_folder, local_or_central, [pattern]
+    )
+    # We expect six folders.
+    assert len(result) == 6