Skip to content

Data storage

The canonical storage layer shared by the raw and derivative datasets.

Derivative samples and eye-movement annotations are stored as compressed BIDS TSV.GZ files with JSON sidecars, following the general BIDS Derivatives conventions. BIDS does not yet define a domain-specific derivative schema for detected eye movements, so the additional columns and the processing provenance are documented in those sidecars.

tables

The in-memory table container and the BIDS TSV read and write helpers. The whole tabular pipeline uses Polars.

Polars-native table models and BIDS tabular I/O helpers.

SessionTables dataclass

Canonical in-memory representation of one eye-tracking session.

Source code in pyxations/tables.py 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
@dataclass(slots=True)
class SessionTables:
    """Canonical in-memory representation of one eye-tracking session."""

    samples: pl.DataFrame
    fixations: pl.DataFrame = field(default_factory=empty_frame)
    saccades: pl.DataFrame = field(default_factory=empty_frame)
    blinks: pl.DataFrame = field(default_factory=empty_frame)
    messages: pl.DataFrame = field(default_factory=empty_frame)
    calibration: pl.DataFrame = field(default_factory=empty_frame)
    header: pl.DataFrame = field(default_factory=empty_frame)
    behavioral_events: pl.DataFrame = field(default_factory=empty_frame)
    sampling_frequency: float | None = None
    screen_width: int | None = None
    screen_height: int | None = None

    def __post_init__(self) -> None:
        for name in (
            "samples",
            "fixations",
            "saccades",
            "blinks",
            "messages",
            "calibration",
            "header",
            "behavioral_events",
        ):
            setattr(self, name, as_polars(getattr(self, name), name=name))

        if self.sampling_frequency is not None:
            frequency = float(self.sampling_frequency)
            if not math.isfinite(frequency) or frequency <= 0:
                raise ValueError(
                    "sampling_frequency must be finite and greater than zero"
                )
            self.sampling_frequency = frequency

    def clone(self, **updates: Any) -> SessionTables:
        """Return an independent copy, optionally replacing selected fields.

        Parameters
        ----------
        **updates : object
            Field values to replace in the cloned session container.

        Returns
        -------
        SessionTables
            Independent container whose table fields are cloned.
        """

        values = {
            "samples": self.samples,
            "fixations": self.fixations,
            "saccades": self.saccades,
            "blinks": self.blinks,
            "messages": self.messages,
            "calibration": self.calibration,
            "header": self.header,
            "behavioral_events": self.behavioral_events,
            "sampling_frequency": self.sampling_frequency,
            "screen_width": self.screen_width,
            "screen_height": self.screen_height,
        }
        values.update(updates)
        return SessionTables(**values)

clone(**updates)

Return an independent copy, optionally replacing selected fields.

Parameters:

Name Type Description Default
**updates object

Field values to replace in the cloned session container.

 {}

Returns:

Type Description
SessionTables

Independent container whose table fields are cloned.
Source code in pyxations/tables.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def clone(self, **updates: Any) -> SessionTables:
    """Return an independent copy, optionally replacing selected fields.

    Parameters
    ----------
    **updates : object
        Field values to replace in the cloned session container.

    Returns
    -------
    SessionTables
        Independent container whose table fields are cloned.
    """

    values = {
        "samples": self.samples,
        "fixations": self.fixations,
        "saccades": self.saccades,
        "blinks": self.blinks,
        "messages": self.messages,
        "calibration": self.calibration,
        "header": self.header,
        "behavioral_events": self.behavioral_events,
        "sampling_frequency": self.sampling_frequency,
        "screen_width": self.screen_width,
        "screen_height": self.screen_height,
    }
    values.update(updates)
    return SessionTables(**values)

as_polars(frame, *, name='table')

Clone a Polars table, or create an empty one for None.

Parameters:

Name Type Description Default
frame DataFrame or None

Table to clone. None creates a schema-less empty table.

 required
name str

Human-readable value name used in type errors.

 "table"

Returns:

Type Description
DataFrame

Independent clone of frame or a new empty table.

Raises:

Type Description
TypeError

If frame is neither a Polars DataFrame nor None.
Source code in pyxations/tables.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def as_polars(frame: Any | None, *, name: str = "table") -> pl.DataFrame:
    """Clone a Polars table, or create an empty one for ``None``.

    Parameters
    ----------
    frame : polars.DataFrame or None
        Table to clone. ``None`` creates a schema-less empty table.
    name : str, default "table"
        Human-readable value name used in type errors.

    Returns
    -------
    polars.DataFrame
        Independent clone of ``frame`` or a new empty table.

    Raises
    ------
    TypeError
        If ``frame`` is neither a Polars DataFrame nor ``None``.
    """

    if frame is None:
        return pl.DataFrame()
    if isinstance(frame, pl.DataFrame):
        return frame.clone()
    raise TypeError(f"{name} must be a Polars DataFrame, got {type(frame)!r}.")

empty_frame()

Return a new empty Polars frame for dataclass defaults.

Returns:

Type Description
DataFrame

New schema-less empty table.
Source code in pyxations/tables.py 
19
20
21
22
23
24
25
26
27
28
def empty_frame() -> pl.DataFrame:
    """Return a new empty Polars frame for dataclass defaults.

    Returns
    -------
    polars.DataFrame
        New schema-less empty table.
    """

    return pl.DataFrame()

frame_payload(frame)

Serialize a table into an explicitly column-ordered JSON payload.

Parameters:

Name Type Description Default
frame DataFrame or None

Table to serialize.

 required

Returns:

Type Description
dict

Payload containing ordered Columns and row Records.
Source code in pyxations/tables.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
def frame_payload(frame: Any | None) -> dict[str, Any]:
    """Serialize a table into an explicitly column-ordered JSON payload.

    Parameters
    ----------
    frame : polars.DataFrame or None
        Table to serialize.

    Returns
    -------
    dict
        Payload containing ordered ``Columns`` and row ``Records``.
    """

    table = as_polars(frame)
    records = [
        {column: json_value(value) for column, value in row.items()}
        for row in table.to_dicts()
    ]
    return {"Columns": table.columns, "Records": records}

json_value(value)

Convert scalar or nested table values to JSON-safe values.

Parameters:

Name Type Description Default
value object

Scalar, sequence, mapping, date, or Polars series to normalize.

 required

Returns:

Type Description
object

Recursively normalized value accepted by the standard JSON encoder.
Source code in pyxations/tables.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def json_value(value: Any) -> Any:
    """Convert scalar or nested table values to JSON-safe values.

    Parameters
    ----------
    value : object
        Scalar, sequence, mapping, date, or Polars series to normalize.

    Returns
    -------
    object
        Recursively normalized value accepted by the standard JSON encoder.
    """

    if value is None:
        return None
    if isinstance(value, (np.bool_, bool)):
        return bool(value)
    if isinstance(value, (np.integer,)):
        return int(value)
    if isinstance(value, (np.floating, float)):
        return None if not math.isfinite(float(value)) else float(value)
    if isinstance(value, (datetime, date)):
        return value.isoformat()
    if isinstance(value, Mapping):
        return {str(key): json_value(item) for key, item in value.items()}
    if isinstance(value, pl.Series):
        return [json_value(item) for item in value.to_list()]
    if isinstance(value, (list, tuple)):
        return [json_value(item) for item in value]
    if isinstance(value, str):
        return value
    if isinstance(value, int):
        return value
    try:
        if bool(np.isnan(value)):
            return None
    except (TypeError, ValueError):
        pass
    return str(value)

payload_frame(payload)

Deserialize a canonical table payload.

Parameters:

Name Type Description Default
payload mapping or None

Payload containing optional Columns and Records entries.

 required

Returns:

Type Description
DataFrame

Reconstructed table, preserving the recorded column order.
Source code in pyxations/tables.py 
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
def payload_frame(payload: Mapping[str, Any] | None) -> pl.DataFrame:
    """Deserialize a canonical table payload.

    Parameters
    ----------
    payload : mapping or None
        Payload containing optional ``Columns`` and ``Records`` entries.

    Returns
    -------
    polars.DataFrame
        Reconstructed table, preserving the recorded column order.
    """

    if not payload:
        return pl.DataFrame()
    columns = list(payload.get("Columns", []))
    records = list(payload.get("Records", []))
    if not records:
        return pl.DataFrame({column: [] for column in columns})
    table = pl.DataFrame(records, strict=False)
    missing = [column for column in columns if column not in table.columns]
    if missing:
        table = table.with_columns(pl.lit(None).alias(column) for column in missing)
    return table.select(columns)

read_tsv(path, *, columns=None, has_header, schema_overrides=None)

Read a BIDS TSV using Polars with stable null handling.

Parameters:

Name Type Description Default
path str or Path

Plain or gzip-compressed TSV file.

 required
columns sequence of str

Column names for a headerless table.

 None
has_header bool

Whether the first row contains column names.

 required
schema_overrides mapping

Polars data types to impose on selected columns.

 None

Returns:

Type Description
DataFrame

Parsed table with BIDS n/a values represented as nulls.
Source code in pyxations/tables.py 
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
def read_tsv(
    path: str | Path,
    *,
    columns: Sequence[str] | None = None,
    has_header: bool,
    schema_overrides: Mapping[str, pl.DataType] | None = None,
) -> pl.DataFrame:
    """Read a BIDS TSV using Polars with stable null handling.

    Parameters
    ----------
    path : str or pathlib.Path
        Plain or gzip-compressed TSV file.
    columns : sequence of str, optional
        Column names for a headerless table.
    has_header : bool
        Whether the first row contains column names.
    schema_overrides : mapping, optional
        Polars data types to impose on selected columns.

    Returns
    -------
    polars.DataFrame
        Parsed table with BIDS ``n/a`` values represented as nulls.
    """

    options: dict[str, Any] = {
        "separator": "\t",
        "has_header": has_header,
        "null_values": ["n/a"],
        "infer_schema_length": None,
        "truncate_ragged_lines": False,
    }
    if columns is not None:
        options["new_columns"] = list(columns)
    if schema_overrides is not None:
        options["schema_overrides"] = dict(schema_overrides)
    return pl.read_csv(path, **options)

tabular_frame(frame)

Return a TSV-safe frame with nested objects encoded as JSON strings.

Parameters:

Name Type Description Default
frame DataFrame or None

Table to normalize for BIDS tabular output.

 required

Returns:

Type Description
DataFrame

Clone with non-finite floats nulled and nested values JSON-encoded.
Source code in pyxations/tables.py 
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
def tabular_frame(frame: Any | None) -> pl.DataFrame:
    """Return a TSV-safe frame with nested objects encoded as JSON strings.

    Parameters
    ----------
    frame : polars.DataFrame or None
        Table to normalize for BIDS tabular output.

    Returns
    -------
    polars.DataFrame
        Clone with non-finite floats nulled and nested values JSON-encoded.
    """

    table = as_polars(frame)
    if table.is_empty() or not table.columns:
        return table

    expressions = []
    for column, dtype in table.schema.items():
        if dtype.is_float():
            expressions.append(
                pl.when(pl.col(column).is_finite())
                .then(pl.col(column))
                .otherwise(None)
                .alias(column)
            )
        elif dtype.is_nested() or dtype == pl.Object:
            expressions.append(
                pl.col(column)
                .map_elements(_tabular_json, return_dtype=pl.String)
                .alias(column)
            )
    return table.with_columns(expressions) if expressions else table.clone()

write_tsv(path, frame, *, include_header, compressed)

Write a deterministic BIDS TSV, optionally gzip-compressed.

Parameters:

Name Type Description Default
path str or Path

Destination filename.

 required
frame DataFrame

Table to serialize.

 required
include_header bool

Whether to write column names as the first row.

 required
compressed bool

Whether to create a deterministic gzip stream.

 required

Returns:

Type Description
Path

Destination path after the file has been written.
Source code in pyxations/tables.py 
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
def write_tsv(
    path: str | Path,
    frame: Any,
    *,
    include_header: bool,
    compressed: bool,
) -> Path:
    """Write a deterministic BIDS TSV, optionally gzip-compressed.

    Parameters
    ----------
    path : str or pathlib.Path
        Destination filename.
    frame : polars.DataFrame
        Table to serialize.
    include_header : bool
        Whether to write column names as the first row.
    compressed : bool
        Whether to create a deterministic gzip stream.

    Returns
    -------
    pathlib.Path
        Destination path after the file has been written.
    """

    destination = Path(path)
    destination.parent.mkdir(parents=True, exist_ok=True)
    table = tabular_frame(frame)

    if compressed:
        # Compress into memory rather than onto the open file. Polars writes
        # straight to the file descriptor whenever the object it is given
        # exposes fileno(), which a wrapper chain over a real file does, so
        # writing through GzipFile that way silently bypasses compression and
        # produces a corrupt archive. A BytesIO has no fileno(), so the text
        # actually goes through the compressor.
        buffer = io.BytesIO()
        with (
            gzip.GzipFile(
                filename="", fileobj=buffer, mode="wb", mtime=0
            ) as gzip_stream,
            io.TextIOWrapper(gzip_stream, encoding="utf-8", newline="") as text_stream,
        ):
            table.write_csv(
                text_stream,
                separator="\t",
                include_header=include_header,
                null_value="n/a",
            )
        destination.write_bytes(buffer.getvalue())
    else:
        table.write_csv(
            destination,
            separator="\t",
            include_header=include_header,
            null_value="n/a",
        )
    return destination

export

The derivative reader and writer, including the initialization of a linked derivative dataset.

Polars-native BIDS Derivatives storage for processed eye-tracking data.

BIDSDerivativeExport

Write and read a :class:SessionTables BIDS derivative session.

Source code in pyxations/export/bids.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
class BIDSDerivativeExport:
    """Write and read a :class:`SessionTables` BIDS derivative session."""

    @staticmethod
    def _roots(session_path: Path) -> tuple[Path, Path]:
        derivative_root = session_path.parents[1]
        suffix = "_derivatives"
        raw_name = derivative_root.name.removesuffix(suffix)
        return derivative_root, derivative_root.with_name(raw_name)

    @staticmethod
    def _raw_sidecars(session_path: Path) -> list[tuple[Path, dict]]:
        _, raw_root = BIDSDerivativeExport._roots(session_path)
        folder = raw_root / session_path.parent.name / session_path.name / "beh"
        values = []
        for path in sorted(folder.glob("*_physio.json")):
            try:
                values.append((path, json.loads(path.read_text(encoding="utf-8"))))
            except (OSError, json.JSONDecodeError):
                continue
        return values

    @staticmethod
    def _source_prefix(session_path: Path, sidecars) -> str:
        if sidecars:
            stem = sidecars[0][0].name.removesuffix("_physio.json")
            return re.sub(r"_recording-[A-Za-z0-9]+$", "", stem)
        return f"{session_path.parent.name}_{session_path.name}_task-eyetracking"

    @staticmethod
    def _sample_columns(
        frame: pl.DataFrame,
    ) -> tuple[str, str, str | None, str]:
        columns = set(frame.columns)
        if {"X", "Y"}.issubset(columns):
            eye_values = (
                frame.get_column("eye")
                .drop_nulls()
                .cast(pl.String)
                .str.to_uppercase()
                .unique(maintain_order=True)
                .to_list()
                if "eye" in columns
                else []
            )
            recorded_eye = (
                "left"
                if eye_values == ["L"]
                else "right"
                if eye_values == ["R"]
                else "cyclopean"
            )
            return (
                "X",
                "Y",
                "Pupil" if "Pupil" in columns else None,
                recorded_eye,
            )
        candidates = [
            (
                "Gaze2d_Left.x",
                "Gaze2d_Left.y",
                "PupilDiam_Left",
                "left",
            ),
            ("LX", "LY", "LPupil", "left"),
            (
                "Gaze2d_Right.x",
                "Gaze2d_Right.y",
                "PupilDiam_Right",
                "right",
            ),
            ("RX", "RY", "RPupil", "right"),
        ]
        for x_column, y_column, pupil_column, eye in candidates:
            if {x_column, y_column}.issubset(columns):
                return (
                    x_column,
                    y_column,
                    pupil_column if pupil_column in columns else None,
                    eye,
                )
        raise ValueError(
            "Processed samples do not contain a supported gaze-coordinate pair"
        )

    @staticmethod
    def _time_column(frame: pl.DataFrame) -> str:
        for column in ("t_acum", "tSample", "timestamp"):
            if column in frame.columns:
                return column
        raise ValueError("Processed samples do not contain a timestamp column")

    @staticmethod
    def _source_metadata(sidecars, recorded_eye: str) -> dict:
        for _, metadata in sidecars:
            if metadata.get("RecordedEye") == recorded_eye:
                return metadata
        return sidecars[0][1] if sidecars else {}

    @staticmethod
    def _time_scale(frame: pl.DataFrame, time_column: str, metadata: dict) -> float:
        if time_column == "t_acum":
            return 1_000.0
        if {"TIMETICK", "BPOGX"}.intersection(frame.columns):
            return 1.0
        if {"Recording timestamp", "Gaze2d_Left.x"}.intersection(frame.columns):
            return 1_000_000.0
        return _unit_scale(metadata.get("timestamp", {}).get("Units"))

    @staticmethod
    def _read_auxiliary_json(session_path: Path, filename: str):
        path = session_path / filename
        if not path.is_file():
            return None
        try:
            return json.loads(path.read_text(encoding="utf-8"))
        finally:
            path.unlink(missing_ok=True)

    def write_session(
        self,
        session_path: str | Path,
        tables: SessionTables,
        *,
        detection_algorithm: str,
    ) -> tuple[Path, Path | None]:
        """Write one processed sample stream and its event annotations.

        Parameters
        ----------
        session_path : str or pathlib.Path
            Derivative session directory.
        tables : SessionTables
            Processed samples, events, behavior, and metadata to persist.
        detection_algorithm : str
            Detector label incorporated into the BIDS recording entity.

        Returns
        -------
        tuple of pathlib.Path and pathlib.Path or None
            Sample path and optional physiological-event path.

        Raises
        ------
        ValueError
            If samples lack required coordinates or valid timestamps.
        """

        session_path = Path(session_path)
        sample_frame = tables.samples
        sidecars = self._raw_sidecars(session_path)
        prefix = self._source_prefix(session_path, sidecars)
        label = bids_label(detection_algorithm.lower(), fallback="pyxations")
        base = f"{prefix}_recording-eye1{label}"
        destination = session_path / "beh"

        x_column, y_column, pupil_column, recorded_eye = self._sample_columns(
            sample_frame
        )
        time_column = self._time_column(sample_frame)
        source_metadata = self._source_metadata(sidecars, recorded_eye)
        time_scale = self._time_scale(sample_frame, time_column, source_metadata)
        raw_time = _numeric_series(sample_frame, time_column)
        finite_time = _finite_values(raw_time)
        if finite_time.is_empty():
            raise ValueError("Processed samples contain no valid timestamps")
        time_origin = float(finite_time[0])

        canonical_columns = {
            "timestamp": time_column,
            "x_coordinate": x_column,
            "y_coordinate": y_column,
        }
        if pupil_column:
            canonical_columns["pupil_size"] = pupil_column
        auxiliary_columns = [
            column
            for column in sample_frame.columns
            if column not in canonical_columns.values()
        ]
        original_to_bids, bids_to_original = _column_mapping(auxiliary_columns)
        expressions = [
            (
                (pl.col(time_column).cast(pl.Float64, strict=False) - time_origin)
                / time_scale
            ).alias("timestamp"),
            pl.col(x_column).cast(pl.Float64, strict=False).alias("x_coordinate"),
            pl.col(y_column).cast(pl.Float64, strict=False).alias("y_coordinate"),
        ]
        if pupil_column:
            expressions.append(
                pl.col(pupil_column).cast(pl.Float64, strict=False).alias("pupil_size")
            )
        expressions.extend(
            pl.col(original).alias(bids_column)
            for original, bids_column in original_to_bids.items()
        )
        standardized = (
            sample_frame.select(expressions)
            .filter(pl.col("timestamp").is_not_null() & pl.col("timestamp").is_finite())
            .sort("timestamp", maintain_order=True)
        )

        sampling_frequency = source_metadata.get("SamplingFrequency")
        if sampling_frequency is None and "Rate_recorded" in sample_frame.columns:
            rates = _finite_values(_numeric_series(sample_frame, "Rate_recorded"))
            sampling_frequency = float(rates.median()) if not rates.is_empty() else None
        sampling_frequency = float(
            sampling_frequency or _infer_frequency(standardized.get_column("timestamp"))
        )

        coordinate = source_metadata.get("x_coordinate", {})
        coordinate_unit = coordinate.get("Units", "arbitrary")
        coordinate_system = source_metadata.get("SampleCoordinateSystem", "custom")
        coordinate_description = source_metadata.get(
            "SampleCoordinateSystemDescription",
            "Coordinate system retained from the processed source recording.",
        )
        metadata = {
            "SamplingFrequency": sampling_frequency,
            "StartTime": 0.0,
            "Columns": standardized.columns,
            "PhysioType": "eyetrack",
            "RecordedEye": recorded_eye,
            "SampleCoordinateSystem": coordinate_system,
            "SampleCoordinateSystemDescription": coordinate_description,
            "Description": (
                "Eye-tracking samples processed by Pyxations using "
                f"{detection_algorithm}."
            ),
            "timestamp": {
                "Description": "Time elapsed since the first processed sample.",
                "Units": "s",
                "Origin": "First sample in the source recording",
            },
            "x_coordinate": {
                "Description": "Processed horizontal gaze coordinate.",
                "Units": coordinate_unit,
            },
            "y_coordinate": {
                "Description": "Processed vertical gaze coordinate.",
                "Units": coordinate_unit,
            },
            "PyxationsColumnMap": bids_to_original,
            "PyxationsCanonicalColumnMap": canonical_columns,
            "PyxationsSampleColumns": sample_frame.columns,
            "PyxationsSampleSchema": {
                column: str(dtype) for column, dtype in sample_frame.schema.items()
            },
            "PyxationsTimeOrigin": time_origin,
            "PyxationsTimeScale": time_scale,
            "PyxationsDetectionAlgorithm": detection_algorithm,
            "PyxationsCalibration": frame_payload(tables.calibration),
            "PyxationsHeader": frame_payload(tables.header),
            "PyxationsPreprocessingRecipe": self._read_auxiliary_json(
                session_path, "preprocessing_recipe.json"
            ),
            "PyxationsPreprocessingProvenance": self._read_auxiliary_json(
                session_path, "preprocessing_provenance.json"
            ),
            "PyxationsBehavioralEvents": frame_payload(tables.behavioral_events),
        }
        if "pupil_size" in standardized.columns:
            pupil_metadata = source_metadata.get("pupil_size", {})
            metadata["pupil_size"] = {
                "Description": pupil_metadata.get(
                    "Description",
                    "Processed pupil diameter or area as reported by the "
                    "source tracker; consult the source metadata for type.",
                ),
                "Units": pupil_metadata.get("Units", "arbitrary"),
            }
        for bids_column, original in bids_to_original.items():
            metadata[bids_column] = {
                "Description": (
                    f"Pyxations analysis column; original name: {original}."
                )
            }

        physio_path = destination / f"{base}_physio.tsv.gz"
        _write_json(destination / f"{base}_physio.json", metadata)
        write_tsv(
            physio_path,
            standardized,
            include_header=False,
            compressed=True,
        )

        event_path, _ = self._write_events(
            destination=destination,
            base=base,
            tables={
                "fix": tables.fixations,
                "sacc": tables.saccades,
                "blink": tables.blinks,
                "msg": tables.messages,
            },
            sample_time_origin=time_origin,
            sample_time_scale=time_scale,
            sample_duration=float(standardized.get_column("timestamp").max()),
        )
        return physio_path, event_path

    @staticmethod
    def _event_scale(
        onset: pl.Series,
        *,
        sample_time_origin: float,
        sample_time_scale: float,
        sample_duration: float,
        table_name: str,
    ) -> float:
        if table_name == "msg":
            return sample_time_scale
        candidates = []
        numeric = onset.cast(pl.Float64, strict=False)
        for scale in dict.fromkeys((sample_time_scale, 1_000.0, 1.0)):
            seconds = (numeric - sample_time_origin) / scale
            valid = _finite_values(seconds)
            score = (
                float(
                    (
                        (valid >= -1.0) & (valid <= max(sample_duration + 1.0, 1.0))
                    ).mean()
                )
                if not valid.is_empty()
                else 0.0
            )
            candidates.append((score, scale))
        return max(candidates, key=lambda item: item[0])[1]

    def _write_events(
        self,
        *,
        destination: Path,
        base: str,
        tables: Mapping[str, pl.DataFrame],
        sample_time_origin: float,
        sample_time_scale: float,
        sample_duration: float,
    ) -> tuple[Path | None, Path | None]:
        all_columns = [column for frame in tables.values() for column in frame.columns]
        original_to_bids, bids_to_original = _column_mapping(dict.fromkeys(all_columns))
        table_columns = {name: frame.columns for name, frame in tables.items()}
        event_names = {
            "fix": "fixation",
            "sacc": "saccade",
            "blink": "blink",
            "msg": "message",
        }

        prepared = []
        for table_name, original_frame in tables.items():
            if original_frame.is_empty():
                continue
            frame = original_frame
            onset_column = next(
                (
                    column
                    for column in ("tStart", "timestamp", "tSample", "tEnd")
                    if column in frame.columns
                ),
                None,
            )
            if onset_column is None:
                continue
            scale = self._event_scale(
                frame.get_column(onset_column),
                sample_time_origin=sample_time_origin,
                sample_time_scale=sample_time_scale,
                sample_duration=sample_duration,
                table_name=table_name,
            )
            onset = (
                (
                    pl.col(onset_column).cast(pl.Float64, strict=False)
                    - sample_time_origin
                )
                / scale
            ).alias("onset")
            if "duration" in frame.columns:
                duration = (
                    pl.col("duration").cast(pl.Float64, strict=False).fill_null(0.0)
                    / scale
                ).clip(lower_bound=0.0)
            elif "tEnd" in frame.columns and onset_column != "tEnd":
                duration = (
                    (
                        (
                            pl.col("tEnd").cast(pl.Float64, strict=False)
                            - pl.col(onset_column).cast(pl.Float64, strict=False)
                        )
                        / scale
                    )
                    .fill_null(0.0)
                    .clip(lower_bound=0.0)
                )
            else:
                duration = pl.lit(0.0)

            prepared.append(
                frame.select(
                    onset,
                    duration.alias("duration"),
                    pl.lit(event_names[table_name]).alias("trial_type"),
                    pl.lit(table_name).alias("pyxations_table"),
                    *[
                        pl.col(original).alias(original_to_bids[original])
                        for original in frame.columns
                    ],
                )
            )

        if not prepared:
            return None, None
        events = (
            pl.concat(prepared, how="diagonal_relaxed")
            .filter(pl.col("onset").is_not_null())
            .sort("onset", maintain_order=True)
        )
        if events.is_empty():
            return None, None

        metadata = {
            "Columns": events.columns,
            "Description": (
                "Fixations, saccades, blinks, and messages identified or "
                "retained by the Pyxations processing pipeline."
            ),
            "OnsetSource": "timestamp",
            "onset": {
                "Description": (
                    "Onset in seconds on the timeline of the associated "
                    "processed eye-tracking recording."
                ),
                "Units": "s",
            },
            "duration": {
                "Description": "Event duration.",
                "Units": "s",
            },
            "trial_type": {
                "Description": "Type of eye-movement or message event.",
                "Levels": {
                    "fixation": "Fixation event.",
                    "saccade": "Saccade event.",
                    "blink": "Blink event.",
                    "message": "Message retained from the source recording.",
                },
            },
            "pyxations_table": {
                "Description": (
                    "Original Pyxations table used to reconstruct the "
                    "analysis DataFrame."
                )
            },
            "PyxationsColumnMap": bids_to_original,
            "PyxationsTableColumns": table_columns,
        }
        for bids_column, original in bids_to_original.items():
            metadata[bids_column] = {
                "Description": (f"Pyxations event column; original name: {original}.")
            }

        event_path = destination / f"{base}_physioevents.tsv.gz"
        event_json = destination / f"{base}_physioevents.json"
        write_tsv(
            event_path,
            events,
            include_header=False,
            compressed=True,
        )
        _write_json(event_json, metadata)
        return event_path, event_json

    @staticmethod
    def _read_table(path: Path, metadata: Mapping) -> pl.DataFrame:
        return read_tsv(
            path,
            columns=list(metadata["Columns"]),
            has_header=False,
        )

    def read_session(
        self, session_path: str | Path, detection_algorithm: str
    ) -> SessionTables:
        """Load BIDS derivatives into the canonical session table model.

        Parameters
        ----------
        session_path : str or pathlib.Path
            Derivative session directory containing ``beh``.
        detection_algorithm : str
            Detector label used when the derivative files were written.

        Returns
        -------
        SessionTables
            Reconstructed samples, events, behavior, and metadata.

        Raises
        ------
        FileNotFoundError
            If no matching derivative physiological recording exists.
        """

        session_path = Path(session_path)
        label = bids_label(detection_algorithm.lower(), fallback="pyxations")
        physio_files = sorted(
            (session_path / "beh").glob(f"*_recording-eye1{label}_physio.tsv.gz")
        )
        if not physio_files:
            raise FileNotFoundError(
                f"No BIDS derivatives for {detection_algorithm} in {session_path}"
            )
        physio_path = physio_files[0]
        physio_metadata = json.loads(
            physio_path.with_suffix("").with_suffix(".json").read_text(encoding="utf-8")
        )
        samples_bids = self._read_table(physio_path, physio_metadata)
        sample_mapping = physio_metadata.get("PyxationsColumnMap", {})
        auxiliary_columns = [
            column for column in sample_mapping if column in samples_bids.columns
        ]
        canonical_mapping = physio_metadata.get("PyxationsCanonicalColumnMap", {})
        sample_schema = physio_metadata.get("PyxationsSampleSchema", {})
        time_origin = float(physio_metadata.get("PyxationsTimeOrigin", 0.0))
        time_scale = float(physio_metadata.get("PyxationsTimeScale", 1.0))

        sample_expressions = []
        for bids_column, original in canonical_mapping.items():
            if bids_column not in samples_bids.columns:
                continue
            expression = pl.col(bids_column)
            if bids_column == "timestamp":
                expression = expression * time_scale + time_origin
                if str(sample_schema.get(original, "")).startswith(("Int", "UInt")):
                    expression = expression.round()
            dtype = getattr(pl, str(sample_schema.get(original, "")), None)
            if dtype is not None:
                expression = expression.cast(dtype, strict=False)
            sample_expressions.append(expression.alias(original))
        sample_expressions.extend(
            pl.col(column).alias(sample_mapping[column]) for column in auxiliary_columns
        )
        samples = samples_bids.select(sample_expressions)
        sample_columns = list(
            physio_metadata.get("PyxationsSampleColumns", samples.columns)
        )
        missing = [column for column in sample_columns if column not in samples.columns]
        if missing:
            samples = samples.with_columns(
                pl.lit(None).alias(column) for column in missing
            )
        samples = samples.select(sample_columns)

        output = {
            "fix": pl.DataFrame(),
            "sacc": pl.DataFrame(),
            "blink": pl.DataFrame(),
            "msg": pl.DataFrame(),
        }
        event_path = physio_path.with_name(
            physio_path.name.replace("_physio.tsv.gz", "_physioevents.tsv.gz")
        )
        if event_path.is_file():
            event_metadata = json.loads(
                event_path.with_suffix("")
                .with_suffix(".json")
                .read_text(encoding="utf-8")
            )
            events = self._read_table(event_path, event_metadata)
            event_mapping = event_metadata.get("PyxationsColumnMap", {})
            table_columns = event_metadata.get("PyxationsTableColumns", {})
            for table_name in output:
                original_columns = list(table_columns.get(table_name, []))
                rows = events.filter(pl.col("pyxations_table") == table_name)
                bids_columns = [
                    bids_column
                    for bids_column, original in event_mapping.items()
                    if original in original_columns and bids_column in rows.columns
                ]
                if rows.is_empty():
                    output[table_name] = pl.DataFrame(
                        {column: [] for column in original_columns}
                    )
                    continue
                reconstructed = rows.select(bids_columns).rename(
                    {column: event_mapping[column] for column in bids_columns}
                )
                missing = [
                    column
                    for column in original_columns
                    if column not in reconstructed.columns
                ]
                if missing:
                    reconstructed = reconstructed.with_columns(
                        pl.lit(None).alias(column) for column in missing
                    )
                output[table_name] = reconstructed.select(original_columns)

        return SessionTables(
            samples=samples,
            fixations=output["fix"],
            saccades=output["sacc"],
            blinks=output["blink"],
            messages=output["msg"],
            calibration=payload_frame(physio_metadata.get("PyxationsCalibration")),
            header=payload_frame(physio_metadata.get("PyxationsHeader")),
            behavioral_events=payload_frame(
                physio_metadata.get("PyxationsBehavioralEvents")
            ),
            sampling_frequency=float(physio_metadata["SamplingFrequency"]),
        )

read_session(session_path, detection_algorithm)

Load BIDS derivatives into the canonical session table model.

Parameters:

Name Type Description Default
session_path str or Path

Derivative session directory containing beh.

 required
detection_algorithm str

Detector label used when the derivative files were written.

 required

Returns:

Type Description
SessionTables

Reconstructed samples, events, behavior, and metadata.

Raises:

Type Description
FileNotFoundError

If no matching derivative physiological recording exists.
Source code in pyxations/export/bids.py 
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
def read_session(
    self, session_path: str | Path, detection_algorithm: str
) -> SessionTables:
    """Load BIDS derivatives into the canonical session table model.

    Parameters
    ----------
    session_path : str or pathlib.Path
        Derivative session directory containing ``beh``.
    detection_algorithm : str
        Detector label used when the derivative files were written.

    Returns
    -------
    SessionTables
        Reconstructed samples, events, behavior, and metadata.

    Raises
    ------
    FileNotFoundError
        If no matching derivative physiological recording exists.
    """

    session_path = Path(session_path)
    label = bids_label(detection_algorithm.lower(), fallback="pyxations")
    physio_files = sorted(
        (session_path / "beh").glob(f"*_recording-eye1{label}_physio.tsv.gz")
    )
    if not physio_files:
        raise FileNotFoundError(
            f"No BIDS derivatives for {detection_algorithm} in {session_path}"
        )
    physio_path = physio_files[0]
    physio_metadata = json.loads(
        physio_path.with_suffix("").with_suffix(".json").read_text(encoding="utf-8")
    )
    samples_bids = self._read_table(physio_path, physio_metadata)
    sample_mapping = physio_metadata.get("PyxationsColumnMap", {})
    auxiliary_columns = [
        column for column in sample_mapping if column in samples_bids.columns
    ]
    canonical_mapping = physio_metadata.get("PyxationsCanonicalColumnMap", {})
    sample_schema = physio_metadata.get("PyxationsSampleSchema", {})
    time_origin = float(physio_metadata.get("PyxationsTimeOrigin", 0.0))
    time_scale = float(physio_metadata.get("PyxationsTimeScale", 1.0))

    sample_expressions = []
    for bids_column, original in canonical_mapping.items():
        if bids_column not in samples_bids.columns:
            continue
        expression = pl.col(bids_column)
        if bids_column == "timestamp":
            expression = expression * time_scale + time_origin
            if str(sample_schema.get(original, "")).startswith(("Int", "UInt")):
                expression = expression.round()
        dtype = getattr(pl, str(sample_schema.get(original, "")), None)
        if dtype is not None:
            expression = expression.cast(dtype, strict=False)
        sample_expressions.append(expression.alias(original))
    sample_expressions.extend(
        pl.col(column).alias(sample_mapping[column]) for column in auxiliary_columns
    )
    samples = samples_bids.select(sample_expressions)
    sample_columns = list(
        physio_metadata.get("PyxationsSampleColumns", samples.columns)
    )
    missing = [column for column in sample_columns if column not in samples.columns]
    if missing:
        samples = samples.with_columns(
            pl.lit(None).alias(column) for column in missing
        )
    samples = samples.select(sample_columns)

    output = {
        "fix": pl.DataFrame(),
        "sacc": pl.DataFrame(),
        "blink": pl.DataFrame(),
        "msg": pl.DataFrame(),
    }
    event_path = physio_path.with_name(
        physio_path.name.replace("_physio.tsv.gz", "_physioevents.tsv.gz")
    )
    if event_path.is_file():
        event_metadata = json.loads(
            event_path.with_suffix("")
            .with_suffix(".json")
            .read_text(encoding="utf-8")
        )
        events = self._read_table(event_path, event_metadata)
        event_mapping = event_metadata.get("PyxationsColumnMap", {})
        table_columns = event_metadata.get("PyxationsTableColumns", {})
        for table_name in output:
            original_columns = list(table_columns.get(table_name, []))
            rows = events.filter(pl.col("pyxations_table") == table_name)
            bids_columns = [
                bids_column
                for bids_column, original in event_mapping.items()
                if original in original_columns and bids_column in rows.columns
            ]
            if rows.is_empty():
                output[table_name] = pl.DataFrame(
                    {column: [] for column in original_columns}
                )
                continue
            reconstructed = rows.select(bids_columns).rename(
                {column: event_mapping[column] for column in bids_columns}
            )
            missing = [
                column
                for column in original_columns
                if column not in reconstructed.columns
            ]
            if missing:
                reconstructed = reconstructed.with_columns(
                    pl.lit(None).alias(column) for column in missing
                )
            output[table_name] = reconstructed.select(original_columns)

    return SessionTables(
        samples=samples,
        fixations=output["fix"],
        saccades=output["sacc"],
        blinks=output["blink"],
        messages=output["msg"],
        calibration=payload_frame(physio_metadata.get("PyxationsCalibration")),
        header=payload_frame(physio_metadata.get("PyxationsHeader")),
        behavioral_events=payload_frame(
            physio_metadata.get("PyxationsBehavioralEvents")
        ),
        sampling_frequency=float(physio_metadata["SamplingFrequency"]),
    )

write_session(session_path, tables, *, detection_algorithm)

Write one processed sample stream and its event annotations.

Parameters:

Name Type Description Default
session_path str or Path

Derivative session directory.

 required
tables SessionTables

Processed samples, events, behavior, and metadata to persist.

 required
detection_algorithm str

Detector label incorporated into the BIDS recording entity.

 required

Returns:

Type Description
tuple of pathlib.Path and pathlib.Path or None

Sample path and optional physiological-event path.

Raises:

Type Description
ValueError

If samples lack required coordinates or valid timestamps.
Source code in pyxations/export/bids.py 
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
def write_session(
    self,
    session_path: str | Path,
    tables: SessionTables,
    *,
    detection_algorithm: str,
) -> tuple[Path, Path | None]:
    """Write one processed sample stream and its event annotations.

    Parameters
    ----------
    session_path : str or pathlib.Path
        Derivative session directory.
    tables : SessionTables
        Processed samples, events, behavior, and metadata to persist.
    detection_algorithm : str
        Detector label incorporated into the BIDS recording entity.

    Returns
    -------
    tuple of pathlib.Path and pathlib.Path or None
        Sample path and optional physiological-event path.

    Raises
    ------
    ValueError
        If samples lack required coordinates or valid timestamps.
    """

    session_path = Path(session_path)
    sample_frame = tables.samples
    sidecars = self._raw_sidecars(session_path)
    prefix = self._source_prefix(session_path, sidecars)
    label = bids_label(detection_algorithm.lower(), fallback="pyxations")
    base = f"{prefix}_recording-eye1{label}"
    destination = session_path / "beh"

    x_column, y_column, pupil_column, recorded_eye = self._sample_columns(
        sample_frame
    )
    time_column = self._time_column(sample_frame)
    source_metadata = self._source_metadata(sidecars, recorded_eye)
    time_scale = self._time_scale(sample_frame, time_column, source_metadata)
    raw_time = _numeric_series(sample_frame, time_column)
    finite_time = _finite_values(raw_time)
    if finite_time.is_empty():
        raise ValueError("Processed samples contain no valid timestamps")
    time_origin = float(finite_time[0])

    canonical_columns = {
        "timestamp": time_column,
        "x_coordinate": x_column,
        "y_coordinate": y_column,
    }
    if pupil_column:
        canonical_columns["pupil_size"] = pupil_column
    auxiliary_columns = [
        column
        for column in sample_frame.columns
        if column not in canonical_columns.values()
    ]
    original_to_bids, bids_to_original = _column_mapping(auxiliary_columns)
    expressions = [
        (
            (pl.col(time_column).cast(pl.Float64, strict=False) - time_origin)
            / time_scale
        ).alias("timestamp"),
        pl.col(x_column).cast(pl.Float64, strict=False).alias("x_coordinate"),
        pl.col(y_column).cast(pl.Float64, strict=False).alias("y_coordinate"),
    ]
    if pupil_column:
        expressions.append(
            pl.col(pupil_column).cast(pl.Float64, strict=False).alias("pupil_size")
        )
    expressions.extend(
        pl.col(original).alias(bids_column)
        for original, bids_column in original_to_bids.items()
    )
    standardized = (
        sample_frame.select(expressions)
        .filter(pl.col("timestamp").is_not_null() & pl.col("timestamp").is_finite())
        .sort("timestamp", maintain_order=True)
    )

    sampling_frequency = source_metadata.get("SamplingFrequency")
    if sampling_frequency is None and "Rate_recorded" in sample_frame.columns:
        rates = _finite_values(_numeric_series(sample_frame, "Rate_recorded"))
        sampling_frequency = float(rates.median()) if not rates.is_empty() else None
    sampling_frequency = float(
        sampling_frequency or _infer_frequency(standardized.get_column("timestamp"))
    )

    coordinate = source_metadata.get("x_coordinate", {})
    coordinate_unit = coordinate.get("Units", "arbitrary")
    coordinate_system = source_metadata.get("SampleCoordinateSystem", "custom")
    coordinate_description = source_metadata.get(
        "SampleCoordinateSystemDescription",
        "Coordinate system retained from the processed source recording.",
    )
    metadata = {
        "SamplingFrequency": sampling_frequency,
        "StartTime": 0.0,
        "Columns": standardized.columns,
        "PhysioType": "eyetrack",
        "RecordedEye": recorded_eye,
        "SampleCoordinateSystem": coordinate_system,
        "SampleCoordinateSystemDescription": coordinate_description,
        "Description": (
            "Eye-tracking samples processed by Pyxations using "
            f"{detection_algorithm}."
        ),
        "timestamp": {
            "Description": "Time elapsed since the first processed sample.",
            "Units": "s",
            "Origin": "First sample in the source recording",
        },
        "x_coordinate": {
            "Description": "Processed horizontal gaze coordinate.",
            "Units": coordinate_unit,
        },
        "y_coordinate": {
            "Description": "Processed vertical gaze coordinate.",
            "Units": coordinate_unit,
        },
        "PyxationsColumnMap": bids_to_original,
        "PyxationsCanonicalColumnMap": canonical_columns,
        "PyxationsSampleColumns": sample_frame.columns,
        "PyxationsSampleSchema": {
            column: str(dtype) for column, dtype in sample_frame.schema.items()
        },
        "PyxationsTimeOrigin": time_origin,
        "PyxationsTimeScale": time_scale,
        "PyxationsDetectionAlgorithm": detection_algorithm,
        "PyxationsCalibration": frame_payload(tables.calibration),
        "PyxationsHeader": frame_payload(tables.header),
        "PyxationsPreprocessingRecipe": self._read_auxiliary_json(
            session_path, "preprocessing_recipe.json"
        ),
        "PyxationsPreprocessingProvenance": self._read_auxiliary_json(
            session_path, "preprocessing_provenance.json"
        ),
        "PyxationsBehavioralEvents": frame_payload(tables.behavioral_events),
    }
    if "pupil_size" in standardized.columns:
        pupil_metadata = source_metadata.get("pupil_size", {})
        metadata["pupil_size"] = {
            "Description": pupil_metadata.get(
                "Description",
                "Processed pupil diameter or area as reported by the "
                "source tracker; consult the source metadata for type.",
            ),
            "Units": pupil_metadata.get("Units", "arbitrary"),
        }
    for bids_column, original in bids_to_original.items():
        metadata[bids_column] = {
            "Description": (
                f"Pyxations analysis column; original name: {original}."
            )
        }

    physio_path = destination / f"{base}_physio.tsv.gz"
    _write_json(destination / f"{base}_physio.json", metadata)
    write_tsv(
        physio_path,
        standardized,
        include_header=False,
        compressed=True,
    )

    event_path, _ = self._write_events(
        destination=destination,
        base=base,
        tables={
            "fix": tables.fixations,
            "sacc": tables.saccades,
            "blink": tables.blinks,
            "msg": tables.messages,
        },
        sample_time_origin=time_origin,
        sample_time_scale=time_scale,
        sample_duration=float(standardized.get_column("timestamp").max()),
    )
    return physio_path, event_path

initialize_bids_derivative(raw_dataset, derivative_dataset)

Create dataset-level metadata for a standalone BIDS Derivatives dataset.

Parameters:

Name Type Description Default
raw_dataset str or Path

Source raw BIDS dataset.

 required
derivative_dataset str or Path

Destination derivative dataset.

 required

Returns:

Type Description
Path

Initialized derivative dataset root.
Source code in pyxations/export/bids.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
def initialize_bids_derivative(
    raw_dataset: str | Path,
    derivative_dataset: str | Path,
) -> Path:
    """Create dataset-level metadata for a standalone BIDS Derivatives dataset.

    Parameters
    ----------
    raw_dataset : str or pathlib.Path
        Source raw BIDS dataset.
    derivative_dataset : str or pathlib.Path
        Destination derivative dataset.

    Returns
    -------
    pathlib.Path
        Initialized derivative dataset root.
    """

    raw_root = Path(raw_dataset)
    derivative_root = Path(derivative_dataset)
    derivative_root.mkdir(parents=True, exist_ok=True)
    _write_json(
        derivative_root / "dataset_description.json",
        {
            "Name": f"{raw_root.name} Pyxations derivatives",
            "BIDSVersion": BIDS_VERSION,
            "DatasetType": "derivative",
            "GeneratedBy": [
                {
                    "Name": "Pyxations",
                    "Version": _package_version(),
                    "Description": (
                        "Eye-movement detection, preprocessing, and trial "
                        "annotation of BIDS eye-tracking recordings."
                    ),
                }
            ],
            "SourceDatasets": [{"URL": f"../{raw_root.name}/"}],
        },
    )
    for filename in ("participants.tsv", "participants.json"):
        source = raw_root / filename
        if source.is_file():
            shutil.copy2(source, derivative_root / filename)
    (derivative_root / "README").write_text(
        "This dataset contains eye-tracking derivatives generated by "
        "Pyxations from the sibling raw BIDS dataset. Processed sample "
        "recordings use the physio suffix, and detected fixations, saccades, "
        "blinks, and retained messages use matching physioevents files. "
        "Human-readable plots may be generated under figures/; those report "
        "artifacts are not part of the standardized BIDS tables.\n",
        encoding="utf-8",
        newline="\n",
    )
    (derivative_root / ".bidsignore").write_text(
        "figures\nfigures/**\n",
        encoding="utf-8",
        newline="\n",
    )
    return derivative_root