revision

filter function for bluesky folders
new function names
more documentation

Author: ddkohler

WrightTools/kit/_bluesky.py

@@ -1,5 +1,5 @@
 """
-Helpers specific to deal with data structures from the Wright Group's Bluesky deployment
+Helpers and containers for data structures in the Wright Group's Bluesky deployment
 https://github.com/wright-group/bluesky-in-a-box/
 """
 
@@ -8,7 +8,7 @@
 import datetime
 import pathlib
 import logging
-from typing import NamedTuple
+from typing import NamedTuple, Generator, Iterable
 
 from .._open import open as wt5_open
 
@@ -111,56 +111,76 @@ def folder(self):
         )
 
 
-def match_identifier(dir: pathlib.Path, **bluesky_identifier) -> list[BlueskyFolder]:
+def filter_bluesky(items:Iterable[pathlib.Path], **bluesky_identifiers) -> Generator[pathlib.Path, None, None]:
     """
-    walk a directory to find bluesky folder names that match the specified identifiers
+    Filter an iterator of folder names for bluesky folder pattern that match specified values.
 
     Parameters
     ----------
 
-    dir: path-like
-        the directory to iterate through
-
+    items: pathlikes
+        potential paths of bluesky folders
+        
     kwargs
     ------
 
     bluesky_identifiers
         keys corresponding to FolderInfo properties (e.g. date, plan).
 
-    Returns
+    Yields
     -------
-    matches: list of BlueskyFolder objects
-        BlueskyFolders corresponding to full matches with the bluesky_identifiers
+
+    pathlib.Path: 
+        bluesky folders corresponding to full matches with the bluesky_identifiers
+
+    Examples
+    --------
+    ```
+    # match within a directory
+    spooky_folders = [
+        info for info in filter_bluesky(
+            data_folder.iterdir(), 
+            date="2025-10-31"
+        )
+    ]
+    ```
     """
-    for key in bluesky_identifier.keys():
+    for key in bluesky_identifiers.keys():
         assert key in FolderInfo._fields
 
-    keep = []
+    for item in map(pathlib.Path, items):
+        if (info := parse_folder_name(item.name)) is not None:
+            idict = info._asdict()
+            if all(idict[k] == v for k,v in bluesky_identifiers.items()):
+                yield item
 
-    for info in map(
-        lambda item: parse_folder_name(item.name),
-        filter(
-            lambda item: item.is_dir() and re.fullmatch(__folder_seed__, item.name), dir.iterdir()
-        ),
-    ):
-        idict = info._asdict()
-        if all(idict[k] == bluesky_identifier[k] for k in bluesky_identifier.keys()):
-            keep.append(BlueskyFolder(dir / info.folder))
 
-    return keep
+def bluesky_paths(dir: pathlib.Path, **bluesky_identifiers) -> list[pathlib.Path]:
+    """
+    walk a directory to find bluesky folder names that match the specified identifiers.
 
+    Parameters
+    ----------
 
-def parse_folder_contents(folder: str) -> FolderInfo | None:
-    """
-    (Coming soon!)
-    Parse key identifiers from a bluesky folder based on the data inside.
+    dir: path-like
+        the directory to iterate through
 
-    See also
-    --------
-    parse_folder_name:
-        Use the folder name, rather than data inside the folder.
+    kwargs
+    ------
+
+    bluesky_identifiers
+        keys corresponding to FolderInfo properties (e.g. date, plan).
+
+    Returns
+    -------
+    matches: list of BlueskyFolder objects
+        BlueskyFolders corresponding to full matches with the bluesky_identifiers
     """
-    raise NotImplementedError
+
+    return sorted([
+        dir / info.folder
+        for info in filter_bluesky(dir.iterdir(), **bluesky_identifiers)
+    ])
 
 
 def parse_folder_name(folder: str) -> FolderInfo | None:
@@ -178,14 +198,14 @@ def parse_folder_name(folder: str) -> FolderInfo | None:
         if the name is parsed, returns a FolderInfo object.
         otherwise, returns None.
     """
+    out = None
     if ((uid_match := re.fullmatch(r"(?P<uid>\w{8})", folder.split()[-1])) is not None) and (
         (datetime_match := __datetime_seed__.match(folder)) is not None
     ):
         matchdict = uid_match.groupdict() | datetime_match.groupdict()
         matchdict["name"] = " ".join(folder.split()[3:-1])
-        return _to_object(matchdict)
-    else:
-        return None
+        out = _to_object(matchdict)
+    return out
 
 
 def _to_object(mdict: dict) -> FolderInfo: