feat: Access gzip header through context

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('<HxBIxx', fobj.read(10))
+            if magic != 0x8B1F:
+                return None
+
+            if flags & 0b0_0100:  # FEXTRA
+                size = int.from_bytes(fobj.read(2), 'little')
+                fobj.seek(size, 1)  # Skip extra field
+
+            # Fetch enough header to include filename and at least the start of comment.
+            # Buffer should be 1-4KiB. Use 512 bytes (-10 already read) to avoid potentially
+            # depleting the buffer and triggering another IO call, if possible.
+            # If extra fields were present and >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:

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,27 +102,39 @@ 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)
 
     events_context = context.Context(events, ds, subject=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,