239def batch_index_dataset(
240    roots: Sequence[str | PathT],
241    max_workers: int | None = 0,
242    executor_cls: type[ProcessPoolExecutor | ThreadPoolExecutor] = ProcessPoolExecutor,
243    *,
244    show_progress: bool = False,
245    schema: SchemaSpec = None,
246) -> Generator[pa.Table, None, None]:
247    """Index a batch of BIDS datasets.
248
249    Args:
250        roots: List of BIDS dataset root directories.
251        max_workers: Number of indexing processes to run in parallel. Setting
252            `max_workers=0` (the default) uses the main process only. Setting
253            `max_workers=None` starts as many workers as there are available CPUs. See
254            `concurrent.futures.ProcessPoolExecutor` for details.
255        executor_cls: Executor class to use for parallel indexing.
256        show_progress: Show progress bar.
257        schema: Optional `SchemaSpec`. `None` uses the default BIDS schema.
258
259    Yields:
260        An Arrow table index for each BIDS dataset.
261    """
262    func = partial(_batch_index_func, schema=schema)
263    file_count = 0
264    for dataset, table in (
265        pbar := tqdm(
266            _pmap(func, roots, max_workers, executor_cls=executor_cls),
267            total=len(roots) if isinstance(roots, Sequence) else None,
268            disable=show_progress not in {True, "dataset"},
269        )
270    ):
271        file_count += len(table)
272        pbar.set_postfix({"ds": dataset, "N": _hfmt(file_count)}, refresh=False)
273        yield table

42def cloudpathlib_is_available() -> bool:
43    """Check if cloudpathlib is available."""
44    return _CLOUDPATHLIB_AVAILABLE

132def find_bids_datasets(
133    root: str | PathT,
134    exclude: str | list[str] | None = None,
135    maxdepth: int | None = None,
136) -> Generator[PathT, None, None]:
137    """Find all BIDS datasets under a root directory.
138
139    Args:
140        root: Root path to begin search.
141        exclude: Glob pattern or list of patterns matching sub-directory names to
142            exclude from the search.
143        maxdepth: Maximum depth to search.
144
145    Yields:
146        Root paths of all BIDS datasets under `root`.
147    """
148    root = as_path(root)
149
150    if isinstance(exclude, str):
151        exclude = [exclude]
152    elif exclude is None:
153        exclude = []
154    exclude_patterns = [re.compile(fnmatch.translate(pat)) for pat in exclude]
155
156    entry_count = 1
157    ds_count = 0
158
159    if _is_bids_dataset(root):
160        ds_count += 1
161        yield root
162
163    # Tuple of path, depth
164    stack = [(root, 0)]
165
166    while stack:
167        top, depth = stack.pop()
168
169        inside_bids = _is_bids_dataset(top)
170        depth += 1
171
172        for entry in top.iterdir():
173            entry_count += 1
174
175            if any(re.fullmatch(pat, entry.name) for pat in exclude_patterns):
176                continue
177
178            if _is_bids_dataset(entry):
179                ds_count += 1
180                yield entry
181
182            # Checks if we should descend into this directory.
183            # Check not reached final depth.
184            descend = maxdepth is None or depth < maxdepth
185            # Heuristic checks whether the filename looks like a (visible) directory.
186            descend = descend and not (entry.suffix or entry.name.startswith("."))
187            # Only descend into specific subdirectories of BIDS directories.
188            descend = descend and (
189                not inside_bids or entry.name in _BIDS_NESTED_PARENT_DIRNAMES
190            )
191            # Finally, check if actually a directory (which is slow so we want to
192            # short-circuit as much as possible).
193            if descend and entry.is_dir():
194                stack.append((entry, depth))

184def format_bids_path(entities: dict[str, Any], int_format: str = "%d") -> Path:
185    """Construct a formatted BIDS path from entities dict.
186
187    Args:
188        entities: dict mapping BIDS keys to values.
189        int_format: format string for integer (index) BIDS values.
190
191    Returns:
192        A formatted `Path` instance.
193    """
194    special = {"datatype", "suffix", "ext"}
195
196    # Formatted key-value entities.
197    entities_fmt = []
198    for name, value in entities.items():
199        if name not in special:
200            if isinstance(value, int):
201                value = int_format % value
202            entities_fmt.append(f"{name}-{value}")
203    name = "_".join(entities_fmt)
204
205    # Append suffix and extension.
206    if suffix := entities.get("suffix"):
207        name += f"_{suffix}"
208    if ext := entities.get("ext"):
209        name += ext
210
211    # Prepend parent directories.
212    path = Path(name)
213    if datatype := entities.get("datatype"):
214        path = datatype / path
215    if ses := entities.get("ses"):
216        path = f"ses-{ses}" / path
217    return f"sub-{entities['sub']}" / path

 93def get_arrow_schema(*, schema: SchemaSpec = None) -> pa.Schema:
 94    """Get Arrow schema of the BIDS dataset index."""
 95    adapter = load_bids_schema(schema)
 96    entity_schema = entity_arrow_schema(adapter)
 97    index_fields = {
 98        name: pa.field(name, cfg["dtype"], metadata=cfg["metadata"])
 99        for name, cfg in _INDEX_ARROW_FIELDS.items()
100    }
101    fields = [
102        index_fields["dataset"],
103        *entity_schema,
104        index_fields["extra_entities"],
105        index_fields["root"],
106        index_fields["path"],
107    ]
108    metadata = {
109        **entity_schema.metadata,
110        b"bids2table_version": version.encode(),
111    }
112    return pa.schema(fields, metadata=metadata)

115def get_column_names(*, schema: SchemaSpec = None) -> type[enum.StrEnum]:
116    """Get an enum of the BIDS index columns."""
117    # TODO: It might be nice if the column names were statically available. One option
118    # would be to generate a static _schema.py module at install time (similar to how
119    # _version.py is generated) which defines the static default schema and column
120    # names.
121    arrow_schema = get_arrow_schema(schema=schema)
122    items = []
123    for f in arrow_schema:
124        name = f.metadata[b"name"].decode()
125        items.append((name, name))
126
127    BIDSColumn = enum.StrEnum("BIDSColumn", items)  # noqa: N806 - class type
128    BIDSColumn.__doc__ = "Enum of BIDS index column names."
129    return BIDSColumn

197def index_dataset(
198    root: str | PathT,
199    include_subjects: str | list[str] | None = None,
200    *,
201    schema: SchemaSpec = None,
202) -> pa.Table:
203    """Index a BIDS dataset.
204
205    Args:
206        root: BIDS dataset root directory.
207        include_subjects: Glob pattern or list of patterns for matching subjects to
208            include in the index.
209        schema: BIDS schema specification to use. If ``None``, uses the bundled
210            default schema.
211
212    Returns:
213        An Arrow table index of the BIDS dataset.
214    """
215    root = as_path(root)
216
217    arrow_schema = get_arrow_schema(schema=schema)
218
219    dataset, _ = _get_bids_dataset(root)
220    if dataset is None:
221        _logger.warning(f"Path {root} is not a valid BIDS dataset directory.")
222        return pa.Table.from_pylist([], schema=arrow_schema)
223
224    subject_dirs = _find_bids_subject_dirs(root, include_subjects)
225    subject_dirs = sorted(subject_dirs, key=lambda p: p.name)
226    if len(subject_dirs) == 0:
227        _logger.warning(f"Path {root} contains no matching subject dirs.")
228        return pa.Table.from_pylist([], schema=arrow_schema)
229
230    tables = []
231    file_count = 0
232    for sub in subject_dirs:
233        _, table = _index_bids_subject_dir(sub, schema=arrow_schema, dataset=dataset)
234        tables.append(table)
235        file_count += len(table)
236    return pa.concat_tables(tables).combine_chunks()

12def load_bids_metadata(path: str | PathT, *, inherit: bool = True) -> dict[str, Any]:
13    """Load the full JSON sidecar metadata for a BIDS file.
14
15    Sidecar files are loaded according to the inheritance principle in top-down order.
16
17    Args:
18        path: BIDS file path
19        inherit: Load the full metadata according to inheritance. Otherwise, load only
20            the first JSON sidecar found in the bottom-up search.
21
22    Returns:
23        A sidecar metadata dictionary.
24    """
25    path = as_path(path)
26    entities = _cache_parse_bids_entities(path)
27    query = dict(entities, ext=".json")
28
29    metadata = {}
30
31    parent = path.parent
32    if inherit:
33        sidecars = reversed(list(_find_bids_parents(parent, query)))
34    else:
35        sidecars = [next(_find_bids_parents(parent, query))]
36
37    for path in sidecars:
38        try:
39            data = _load_json(path)
40            metadata.update(data)
41        except (json.JSONDecodeError, TypeError):
42            continue
43    return metadata

39def parse_bids_entities(path: str | Path) -> dict[str, str]:
40    """Parse entities from BIDS file path.
41
42    Parses all BIDS filename `"{key}-{value}"` entities as well as special entities:
43    datatype, suffix, ext (extension). Does not validate entities or cast to types.
44
45    Args:
46        path: BIDS path to parse.
47
48    Returns:
49        A dict mapping BIDS entity keys to values.
50    """
51    if isinstance(path, str):
52        path = Path(path)
53    entities = {}
54
55    filename = path.name
56    parts = filename.split("_")
57
58    datatype = _parse_bids_datatype(path)
59
60    # Get suffix and extension.
61    suffix_ext = parts.pop()
62    suffix, dot, ext = suffix_ext.partition(".")
63    ext = dot + ext if ext else None
64
65    # Suffix is actually an entity, put back in list.
66    if "-" in suffix:
67        parts.append(suffix)
68        suffix = None
69
70    # Split entities, skipping any that don't contain a '-'.
71    for part in parts:
72        if "-" in part:
73            key, val = part.split("-", maxsplit=1)
74            entities[key] = val
75
76    entities |= {
77        k: v
78        for k, v in zip(
79            ["datatype", "suffix", "ext"], [datatype, suffix, ext], strict=True
80        )
81        if v is not None
82    }
83    return entities

101def validate_bids_entities(
102    entities: dict[str, Any],
103    *,
104    schema: SchemaSpec = None,
105) -> tuple[dict[str, BIDSValue], dict[str, Any]]:
106    """Validate BIDS entities against a BIDS schema.
107
108    Args:
109        entities: dict mapping BIDS keys to unvalidated entities
110        schema: optional `SchemaSpec` (`Namespace | str | PathT | None`).
111            `None` uses the default BIDS schema bundled with bidsschematools.
112
113    Returns:
114        `(valid_entities, extra_entities)` — valid entities cast to the
115        declared type, plus any leftover entries that did not match a
116        known entity or failed validation.
117    """
118    adapter = load_bids_schema(schema)
119    pa_schema = entity_arrow_schema(adapter)
120    return _pyarrow_validate_entities(entities, pa_schema=pa_schema)