11"""Initialization for repo_templates datasets."""
22
3+ import glob
4+ import os
35from os import path
6+ from typing import Optional
47
58import tomli
69
7- from cfa .dataops .utils import get_fs_ns_map , remove_ws_and_nonalpha
8-
910from .. import _catalog_ns
1011
1112_here = path .abspath (path .dirname (__file__ ))
1213
1314
15+ def remove_ws_and_nonalpha (s : str ) -> str :
16+ """Remove whitespace and non-alphanumeric characters from a string.
17+
18+ Args:
19+ s (str): The input string.
20+
21+ Returns:
22+ str: The cleaned string with whitespace and non-alphanumeric characters removed.
23+
24+ Example:
25+ >>> remove_ws_and_nonalpha("Hello World! 123.ipynb")
26+ 'hello_world_123_ipynb'
27+ """
28+ s = s .replace (" " , "_" ).replace ("." , "_" ).lower ()
29+ return "" .join (c for c in s if c .isalnum () or c == "_" )
30+
31+
1432def get_dataset_name (p : str ) -> str :
1533 """Extract the dataset name from a given file path.
1634
@@ -25,6 +43,52 @@ def get_dataset_name(p: str) -> str:
2543 return remove_ws_and_nonalpha (config ["properties" ]["name" ])
2644
2745
46+ def get_fs_ns_map (
47+ base_dir : str , file_ext : str , endpoint_func : Optional [callable ] = None
48+ ) -> dict :
49+ """Get a nested dictionary representing the filesystem structure starting from base_dir.
50+
51+ Args:
52+ base_dir (str): The base directory to start the search.
53+ file_ext (str): The file extension to filter files (e.g., 'ipynb').
54+ endpoint_func (Optional[callable]): A function that takes a file path and returns a string
55+ to be used as the key in the nested dictionary. If None, the filename without extension
56+ is used.
57+
58+ Returns:
59+ dict: A nested dictionary representing the filesystem structure.
60+ """
61+ base_dir = os .path .abspath (base_dir )
62+ file_ext = file_ext .lstrip ("." )
63+ fs_paths = glob .glob (
64+ os .path .join (base_dir , "**" , f"*.{ file_ext } " ), recursive = True
65+ )
66+ fs_map = {}
67+ for p_i in fs_paths :
68+ if p_i .startswith (base_dir ):
69+ ns_list = [
70+ remove_ws_and_nonalpha (i )
71+ for i in p_i [len (base_dir ) + 1 :].split (os .sep )
72+ ]
73+ current = fs_map
74+ for part in ns_list [:- 1 ]:
75+ if part not in current :
76+ current [part ] = {}
77+ current = current [part ]
78+ if endpoint_func is not None :
79+ ep = endpoint_func (p_i )
80+ if not isinstance (ep , str ):
81+ raise ValueError (
82+ "endpoint_func must return a string when passed the path of your file."
83+ )
84+ else :
85+ current [ep ] = p_i
86+ else :
87+ current [ns_list [- 1 ][: - (len (file_ext ) + 1 )]] = p_i
88+
89+ return fs_map
90+
91+
2892dataset_ns_map = {
2993 _catalog_ns : get_fs_ns_map (
3094 base_dir = _here , file_ext = "toml" , endpoint_func = get_dataset_name
0 commit comments