diff --git a/src/bids_validator/context.py b/src/bids_validator/context.py index 5d3efea..25144e4 100644 --- a/src/bids_validator/context.py +++ b/src/bids_validator/context.py @@ -4,6 +4,7 @@ import gzip import itertools +import struct from functools import cache import attrs @@ -140,6 +141,45 @@ def load_nifti_header(file: FileTree) -> ctx.NiftiHeader: return nifti_header +def load_gzip_header(file: FileTree) -> ctx.Gzip | None: + """Load gzip header fields. + + Reference: https://www.rfc-editor.org/info/rfc1952/ + """ + try: + with file.path_obj.open('rb') as fobj: + # Constant header fields: ID1, ID2, CM, FLG, MTIME, XFL, OS + magic, flags, timestamp = struct.unpack('512 bytes, so be it. + buffer = fobj.read(502) + except OSError: + return None + + filename = '' + if flags & 0b0_1000: # FNAME + fname, buffer = buffer.split(b'\x00', 1) + filename = fname.decode('latin-1') + + comment = '' + if flags & 0b1_0000: # FCOMMENT + # If the comment extends beyond the buffer, split will return one entry. + # *_ accounts for this by setting `_ = []`. + cmnt, *_ = buffer.split(b'\x00', 1) + comment = cmnt.decode('latin-1') + + return ctx.Gzip(timestamp=timestamp, filename=filename, comment=comment) + + class Subjects: """Collections of subjects in the dataset.""" @@ -452,10 +492,12 @@ def json(self) -> Namespace | None: return None - @property - def gzip(self) -> None: + @cached_property + def gzip(self) -> ctx.Gzip | None: """Parsed contents of gzip header.""" - pass + if self.path.endswith('.gz'): + return load_gzip_header(self.file) + return None @cached_property def nifti_header(self) -> ctx.NiftiHeader | None: diff --git a/tests/test_context.py b/tests/test_context.py index f0c642d..a0b9ebc 100644 --- a/tests/test_context.py +++ b/tests/test_context.py @@ -78,6 +78,7 @@ def test_context(synthetic_dataset: FileTree, schema: Namespace) -> None: 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' + stim = sub01 / 'ses-01' / 'func' / 'sub-01_ses-01_task-nback_run-01_stim.tsv.gz' events = synthetic_dataset / 'task-nback_events.tsv' subject = Subject(context.Sessions(sub01)) @@ -101,12 +102,14 @@ def test_context(synthetic_dataset: FileTree, schema: Namespace) -> None: assert T1w_context.sidecar is not None assert T1w_context.sidecar == {} assert T1w_context.json is None + assert T1w_context.gzip is None bold_context = context.Context(bold, ds, subject) assert bold_context.sidecar is not None assert bold_context.sidecar.to_dict() == {'TaskName': 'N-Back', 'RepetitionTime': 2.5} assert bold_context.json is None + assert bold_context.gzip is None assert bold_context.nifti_header is not None assert bold_context.nifti_header.voxel_sizes == (2.0, 2.0, 2.0, 2.5) @@ -114,14 +117,24 @@ def test_context(synthetic_dataset: FileTree, schema: Namespace) -> None: assert events_context.sidecar == Namespace() assert events_context.json is None + assert events_context.gzip is None assert events_context.nifti_header is None assert isinstance(events_context.columns, Namespace) assert 'onset' in events_context.columns assert len(events_context.columns.onset) == 42 + stim_context = context.Context(stim, ds, subject) + assert stim_context.sidecar is not None + assert set(stim_context.sidecar) == {'SamplingFrequency', 'StartTime', 'Columns'} + assert stim_context.gzip is not None + assert stim_context.gzip.filename == stim.name.removesuffix('.gz') + assert stim_context.gzip.timestamp != 0 + assert stim_context.gzip.comment == '' + assert isinstance(stim_context.columns, Namespace) + assert list(stim_context.columns.keys()) == stim_context.sidecar.Columns + ## Tests for: # associations - # gzip # ome # tiff @@ -216,6 +229,22 @@ def test_load_tsv_gz(synthetic_dataset: FileTree) -> None: # Will need an additional test for the content +def test_load_gzip_header(synthetic_dataset: FileTree) -> None: + tsvgz = ( + synthetic_dataset + / 'sub-01' + / 'ses-01' + / 'func' + / 'sub-01_ses-01_task-nback_run-01_stim.tsv.gz' + ) + + gz_header = context.load_gzip_header(tsvgz) + assert gz_header is not None + assert gz_header.timestamp != 0 + assert gz_header.filename == tsvgz.name.removesuffix('.gz') + assert gz_header.comment == '' + + def test_nifti_mrs_header( mrs_data: Path, schema: Namespace,