Develop Context class

pyproject.toml

@@ -24,6 +24,7 @@ requires-python = ">=3.9"
 dependencies = [
     "attrs >=24.1",
     "bidsschematools @ git+https://github.com/bids-standard/bids-specification.git@refs/pull/2133/head#subdirectory=tools/schemacode",
+    "orjson>=3.11.3",
     "universal_pathlib >=0.2",
 ]
 

src/bids_validator/__main__.py

@@ -8,15 +8,25 @@
     raise SystemExit(1) from None
 
 import sys
+from collections.abc import Iterator
 from typing import Annotated
 
+from bidsschematools.schema import load_schema
+from bidsschematools.types import Namespace
+from bidsschematools.types.context import Subject
+
 from bids_validator import BIDSValidator
+from bids_validator.context import Context, Dataset, Sessions
 from bids_validator.types.files import FileTree
 
 app = typer.Typer()
 
 
-def walk(tree: FileTree):
+def is_subject_dir(tree):
+    return tree.name.startswith('sub-')
+
+
+def walk(tree: FileTree, dataset: Dataset, subject: Subject = None) -> Iterator[Context]:
     """Iterate over children of a FileTree and check if they are a directory or file.
 
     If it's a directory then run again recursively, if it's a file file check the file name is
@@ -26,34 +36,39 @@ def walk(tree: FileTree):
     ----------
     tree : FileTree
         FileTree object to iterate over
+    dataset: Dataset
+        Object containing properties for entire dataset
+    subject: Subject
+        object containing subject and session info
 
     """
+    if subject is None and is_subject_dir(tree):
+        subject = Subject(Sessions(tree))
+
     for child in tree.children.values():
         if child.is_dir:
-            yield from walk(child)
+            yield from walk(child, dataset, subject)
         else:
-            yield child
+            yield Context(child, dataset, subject)
 
 
-def validate(tree: FileTree):
+def validate(tree: FileTree, schema: Namespace):
     """Check if the file path is BIDS compliant.
 
     Parameters
     ----------
     tree : FileTree
         Full FileTree object to iterate over and check
+    schema : Namespace
+        Schema object to validate dataset against
 
     """
     validator = BIDSValidator()
+    dataset = Dataset(tree, schema)
 
-    for file in walk(tree):
-        # The output of the FileTree.relative_path method always drops the initial for the path
-        # which makes it fail the validator.is_bids check. THis may be a Windows specific thing.
-        # This line adds it back.
-        path = f'/{file.relative_path}'
-
-        if not validator.is_bids(path):
-            print(f'{path} is not a valid bids filename')
+    for file in walk(tree, dataset):
+        if not validator.is_bids(file.path):
+            print(f'{file.path} is not a valid bids filename')
 
 
 def show_version():
@@ -85,6 +100,7 @@ def version_callback(value: bool):
 @app.command()
 def main(
     bids_path: str,
+    schema_path: str = None,
     verbose: Annotated[bool, typer.Option('--verbose', '-v', help='Show verbose output')] = False,
     version: Annotated[
         bool,
@@ -101,7 +117,10 @@ def main(
 
     root_path = FileTree.read_from_filesystem(bids_path)
 
-    validate(root_path)
+    if not schema_path:
+        schema = load_schema()
+
+    validate(root_path, schema)
 
 
 if __name__ == '__main__':

src/bids_validator/context.py

@@ -3,10 +3,10 @@
 from __future__ import annotations
 
 import itertools
-import json
 from functools import cache
 
 import attrs
+import orjson
 from bidsschematools.types import Namespace
 from bidsschematools.types import context as ctx
 from upath import UPath
@@ -46,6 +46,18 @@ class ValidationError(Exception):
     """TODO: Add issue structure."""
 
 
+_DATATYPE_MAP = {}
+
+
+def datatype_to_modality(datatype: str, schema: Namespace) -> str:
+    """Generate a global map for datatype to modality."""
+    global _DATATYPE_MAP
+    if not _DATATYPE_MAP:
+        for mod_name, mod_dtypes in schema.rules.modalities.items():
+            _DATATYPE_MAP |= dict.fromkeys(mod_dtypes['datatypes'], mod_name)
+    return _DATATYPE_MAP[datatype]
+
+
 @cache
 def load_tsv(file: FileTree, *, max_rows=0) -> Namespace:
     """Load TSV contents into a Namespace."""
@@ -60,8 +72,7 @@ def load_tsv(file: FileTree, *, max_rows=0) -> Namespace:
 @cache
 def load_json(file: FileTree) -> dict[str]:
     """Load JSON file contents."""
-    with open(file) as fobj:
-        return json.load(fobj)
+    return orjson.loads(UPath(file).read_bytes())
 
 
 class Subjects:
@@ -130,14 +141,7 @@ def dataset_description(self) -> Namespace:
     @cached_property
     def modalities(self) -> list[str]:
         """List of modalities found in the dataset."""
-        result = set()
-
-        modalities = self.schema.rules.modalities
-        for datatype in self.datatypes:
-            for mod_name, mod_dtypes in modalities.items():
-                if datatype in mod_dtypes.datatypes:
-                    result.add(mod_name)
-
+        result = {datatype_to_modality(datatype, self.schema) for datatype in self.datatypes}
         return list(result)
 
     @cached_property
@@ -202,6 +206,12 @@ def load_sidecar(file: FileTree) -> dict[str, t.Any]:
     # Uses walk back algorithm
     # https://bids-validator.readthedocs.io/en/latest/validation-model/inheritance-principle.html
     # Accumulates all sidecars
+    metadata = {}
+
+    for json in walk_back(file, inherit=True):
+        metadata = load_json(json) | metadata
+
+    return metadata
 
 
 def walk_back(
@@ -298,3 +308,136 @@ def from_file(cls, file: FileTree, schema: Namespace | None = None) -> t.Self:
             suffix=suffix,
             extension=extension,
         )
+
+
+@attrs.define
+class Context:
+    """A context object that creates context for file on access."""
+
+    file: FileTree
+    dataset: Dataset
+    subject: ctx.Subject | None
+    file_parts: FileParts = attrs.field(init=False)
+
+    def __attrs_post_init__(self):
+        self.file_parts = FileParts.from_file(self.file, self.schema)
+
+    @property
+    def schema(self) -> Namespace:
+        """The BIDS specification schema."""
+        return self.dataset.schema
+
+    @property
+    def path(self) -> str:
+        """Path of the current file."""
+        return self.file_parts.path
+
+    @property
+    def entities(self) -> dict[str, str] | None:
+        """Entities parsed from the current filename."""
+        return self.file_parts.entities
+
+    @property
+    def datatype(self) -> str | None:
+        """Datatype of current file, for examples, anat."""
+        return self.file_parts.datatype
+
+    @property
+    def suffix(self) -> str | None:
+        """Suffix of current file."""
+        return self.file_parts.suffix
+
+    @property
+    def extension(self) -> str | None:
+        """Extension of current file including initial dot."""
+        return self.file_parts.extension
+
+    @property
+    def modality(self) -> str | None:
+        """Modality of current file, for examples, MRI."""
+        return datatype_to_modality(self.file_parts.datatype, self.schema)
+
+    @property
+    def size(self) -> int:
+        """Length of the current file in bytes."""
+        return self.file.direntry.stat().st_size
+
+    @property
+    def associations(self) -> ctx.Associations:
+        """Associated files, indexed by suffix, selected according to the inheritance principle."""
+        return ctx.Associations()
+
+    @property
+    def columns(self) -> None:
+        """TSV columns, indexed by column header, values are arrays with column contents."""
+        pass
+
+    @property
+    def json(self) -> Namespace | None:
+        """Contents of the current JSON file."""
+        if self.file_parts.extension == '.json':
+            return Namespace.build(load_json(self.file))
+
+        return None
+
+    @property
+    def gzip(self) -> None:
+        """Parsed contents of gzip header."""
+        pass
+
+    @property
+    def nifti_header(self) -> None:
+        """Parsed contents of NIfTI header referenced elsewhere in schema."""
+        pass
+
+    @property
+    def ome(self) -> None:
+        """Parsed contents of OME-XML header, which may be found in OME-TIFF or OME-ZARR files."""
+        pass
+
+    @property
+    def tiff(self) -> None:
+        """TIFF file format metadata."""
+        pass
+
+    @property
+    def sidecar(self) -> Namespace | None:
+        """Sidecar metadata constructed via the inheritance principle."""
+        sidecar = load_sidecar(self.file)
+
+        if sidecar:
+            return Namespace.build(sidecar)
+
+        return None
+
+
+class Sessions:
+    """Collections of sessions in subject."""
+
+    def __init__(self, tree: FileTree):
+        self._tree = tree
+
+    @cached_property
+    def ses_dirs(self) -> list[str]:
+        """Sessions as determined by ses-* directories."""
+        return [
+            child.name
+            for child in self._tree.children.values()
+            if child.is_dir and child.name.startswith('ses-')
+        ]
+
+    @property
+    def session_id(self) -> list[str] | None:
+        """The session_id column of *_sessions.tsv."""
+        for name, value in self._tree.children.items():
+            if name.endswith('_sessions.tsv'):
+                return self._get_session_id(value)
+        else:
+            return None
+
+    @staticmethod
+    def _get_session_id(phenotype_file: FileTree) -> list[str] | None:
+        columns = load_tsv(phenotype_file)
+        if 'session_id' not in columns:
+            return None
+        return list(columns['session_id'])

tests/test_context.py

@@ -1,4 +1,5 @@
 import pytest
+from bidsschematools.types.context import Subject
 
 from bids_validator import context
 from bids_validator.types.files import FileTree
@@ -52,3 +53,72 @@ def test_walkback(synthetic_dataset, schema):
     sidecars = list(context.walk_back(bold, inherit=True))
     assert len(sidecars) == 1
     assert sidecars[0] is synthetic_dataset / 'task-nback_bold.json'
+
+
+def test_context(synthetic_dataset, schema):
+    sub01 = synthetic_dataset / 'sub-01'
+    T1w = sub01 / 'ses-01' / 'anat' / 'sub-01_ses-01_T1w.nii'
+    bold = sub01 / 'ses-01' / 'func' / 'sub-01_ses-01_task-nback_run-01_bold.nii'
+
+    subject = Subject(context.Sessions(sub01))
+    ds = context.Dataset(synthetic_dataset, schema)
+    T1w_context = context.Context(T1w, ds, subject)
+
+    assert T1w_context.schema is schema
+    assert T1w_context.dataset is ds
+    assert T1w_context.entities == {'sub': '01', 'ses': '01'}
+    assert T1w_context.path == '/sub-01/ses-01/anat/sub-01_ses-01_T1w.nii'
+    assert T1w_context.datatype == 'anat'
+    assert T1w_context.suffix == 'T1w'
+    assert T1w_context.extension == '.nii'
+    assert T1w_context.modality == 'mri'
+    assert T1w_context.size == 352
+    assert isinstance(T1w_context.subject.sessions, context.Sessions)
+    assert sorted(T1w_context.subject.sessions.ses_dirs) == ['ses-01', 'ses-02']
+    assert sorted(T1w_context.subject.sessions.session_id) == ['ses-01', 'ses-02']
+    assert T1w_context.sidecar is None
+    assert T1w_context.json is None
+
+    bold_context = context.Context(bold, ds, subject)
+
+    assert bold_context.sidecar.to_dict() == {'TaskName': 'N-Back', 'RepetitionTime': 2.5}
+    assert bold_context.json is None
+
+    ## Tests for:
+    #  associations
+    #  columns
+    #  gzip
+    #  nifti_header
+    #  ome
+    #  tiff
+
+
+def test_context_json(examples, schema):
+    dataset = FileTree.read_from_filesystem(examples / 'qmri_vfa')
+    file = dataset / 'sub-01' / 'anat' / 'sub-01_flip-1_VFA.json'
+
+    ds = context.Dataset(dataset, schema)
+    file_context = context.Context(file, ds, subject=None)
+
+    assert file_context.json.to_dict() == {'FlipAngle': 3, 'RepetitionTimeExcitation': 0.0150}
+
+
+def test_sidecar_inheritance(examples):
+    """Test to ensure inheritance principle is executed correctly"""
+    dataset = FileTree.read_from_filesystem(examples / 'qmri_mp2rage')
+    file = dataset / 'sub-1' / 'anat' / 'sub-1_inv-2_part-mag_MP2RAGE.nii'
+
+    sidecar = context.load_sidecar(file)
+
+    assert sidecar['FlipAngle'] == 7
+    assert sidecar['InversionTime'] == 2.7
+    assert sidecar['RepetitionTimePreparation'] == 5.5
+
+
+def test_sessions(synthetic_dataset):
+    sub01 = synthetic_dataset / 'sub-01'
+
+    sessions = context.Sessions(sub01)
+
+    assert sorted(sessions.ses_dirs) == ['ses-01', 'ses-02']
+    assert sorted(sessions.session_id) == ['ses-01', 'ses-02']