Skip to content

Visualization

Plotting utilities for detected eye movements and for raw gaze samples.

Figures are written under the derivative dataset's figures/ directory, in a subdirectory named after the detection algorithm, so results from different detectors do not overwrite each other. That directory is listed in the dataset's .bidsignore, so plotting never invalidates the dataset.

visualization

Scanpaths, fixation-duration and saccade-amplitude distributions, saccade direction, the main sequence, the multipanel summary, and animated gaze. Animations require the optional video extra, installed with pip install 'pyxations[video]'.

Visualization

Plotting utilities for detected eye movements.

Figures are written under derivatives_folder_path in a subdirectory named after the detection algorithm, so results from different detectors stay side by side without overwriting each other. The derivative dataset's .bidsignore excludes figures/, so plotting never invalidates the dataset.

Parameters:

Name Type Description Default
derivatives_folder_path str or Path

Directory where figures are written.

required
events_detection_algorithm str

Name of the detection algorithm, used as the subdirectory name. Must be a bare name, not a path.

required

Raises:

Type Description
ValueError

If events_detection_algorithm is empty or contains path separators.

Source code in pyxations/visualization/visualization.py
  27
  28
  29
  30
  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
  57
  58
  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
 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
 152
 153
 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
 769
 770
 771
 772
 773
 774
 775
 776
 777
 778
 779
 780
 781
 782
 783
 784
 785
 786
 787
 788
 789
 790
 791
 792
 793
 794
 795
 796
 797
 798
 799
 800
 801
 802
 803
 804
 805
 806
 807
 808
 809
 810
 811
 812
 813
 814
 815
 816
 817
 818
 819
 820
 821
 822
 823
 824
 825
 826
 827
 828
 829
 830
 831
 832
 833
 834
 835
 836
 837
 838
 839
 840
 841
 842
 843
 844
 845
 846
 847
 848
 849
 850
 851
 852
 853
 854
 855
 856
 857
 858
 859
 860
 861
 862
 863
 864
 865
 866
 867
 868
 869
 870
 871
 872
 873
 874
 875
 876
 877
 878
 879
 880
 881
 882
 883
 884
 885
 886
 887
 888
 889
 890
 891
 892
 893
 894
 895
 896
 897
 898
 899
 900
 901
 902
 903
 904
 905
 906
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
class Visualization:
    """Plotting utilities for detected eye movements.

    Figures are written under ``derivatives_folder_path`` in a subdirectory
    named after the detection algorithm, so results from different detectors
    stay side by side without overwriting each other. The derivative dataset's
    ``.bidsignore`` excludes ``figures/``, so plotting never invalidates the
    dataset.

    Parameters
    ----------
    derivatives_folder_path : str or pathlib.Path
        Directory where figures are written.
    events_detection_algorithm : str
        Name of the detection algorithm, used as the subdirectory name. Must be
        a bare name, not a path.

    Raises
    ------
    ValueError
        If ``events_detection_algorithm`` is empty or contains path
        separators.
    """

    def __init__(self, derivatives_folder_path, events_detection_algorithm):
        self.derivatives_folder_path = Path(derivatives_folder_path)
        algorithm = str(events_detection_algorithm).strip()
        if not algorithm or Path(algorithm).name != algorithm:
            raise ValueError("events_detection_algorithm must be a non-empty name")
        self.events_detection_folder = Path(algorithm)

    def scanpath(
        self,
        fixations: pl.DataFrame,
        screen_height: int,
        screen_width: int,
        folder_path: str | Path | None = None,
        tmin: int | None = None,
        tmax: int | None = None,
        saccades: pl.DataFrame | None = None,
        samples: pl.DataFrame | None = None,
        phase_data: dict[str, dict] | None = None,
        display: bool = True,
    ):
        """
        Fast scan‑path visualiser.

        • **Vectorised**: no per‑row Python loops
        • **Single pass** phase grouping
        • Uses `BrokenBarHCollection` for fixation spans
        • Each requested PNG is written once

        Parameters
        ----------
        fixations : polars.DataFrame
            Polars DataFrame with at least `tStart`, `duration`, `xAvg`, `yAvg`, `phase`.
        screen_height, screen_width : int
            Stimulus resolution in pixels.
        folder_path : str or pathlib.Path, optional
            Directory where 1 PNG per phase will be stored.  If *None*, nothing is saved.
        tmin, tmax : int, optional
            Time window in **ms**.  If both `None`, the whole trial is plotted.
        saccades : polars.DataFrame, optional
            Polars DataFrame with `tStart`, `phase`, …  (optional).
        samples : polars.DataFrame, optional
            Polars DataFrame with gaze traces (`tSample`, `LX`, `LY`, `RX`, `RY` or
            `X`, `Y`) (optional).
        phase_data : dict, optional
            Per‑phase extras::

                {
                    "search": {
                        "img_paths": [...],
                        "img_plot_coords": [(x1,y1,x2,y2), ...],
                        "bbox": (x1,y1,x2,y2),
                    },
                    ...
                }

        display : bool, default True
            If *False* the figure canvas is never shown (faster for batch jobs).

        Notes
        -----
        One figure is produced per named trial phase. Fixations that fall
        outside every phase are skipped. If no fixation carries a phase name at
        all, which happens when the recording was never segmented, they are all
        drawn as a single unnamed phase and a :class:`UserWarning` is issued.
        """
        if fixations.is_empty():
            return
        required = {"trial_number", "phase", "tStart", "duration", "xAvg", "yAvg"}
        missing = sorted(required - set(fixations.columns))
        if missing:
            raise ValueError(
                "Fixations are missing required columns: " + ", ".join(missing)
            )
        if (tmin is None) != (tmax is None):
            raise ValueError("tmin and tmax must be provided together")
        if folder_path is not None:
            Path(folder_path).mkdir(parents=True, exist_ok=True)

        # ------------- small helpers ------------------------------------------------
        def _make_axes(plot_samples: bool):
            if plot_samples:
                fig, (ax_main, ax_gaze) = plt.subplots(
                    2, 1, height_ratios=(4, 1), figsize=(10, 6), sharex=False
                )
            else:
                fig, ax_main = plt.subplots(figsize=(10, 6))
                ax_gaze = None
            ax_main.set_xlim(0, screen_width)
            ax_main.set_ylim(screen_height, 0)
            return fig, ax_main, ax_gaze

        def _maybe_cache_img(path):
            """Load image from disk with a small LRU cache."""

            # Cache hit: move to the end (most recently used)
            if path in _img_cache:
                img = _img_cache.pop(path)
                _img_cache[path] = img
                return img

            # Cache miss: load image
            img = mpimg.imread(path)

            # Optional: reduce memory if image is float64 in [0, 1]
            if isinstance(img, np.ndarray) and img.dtype == np.float64:
                img = (img * 255).clip(0, 255).astype(np.uint8)

            # Insert into cache
            _img_cache[path] = img

            # If cache too big, drop least recently used item
            if len(_img_cache) > MAX_CACHED_IMAGES:
                _img_cache.popitem(last=False)  # pops the oldest inserted item

            return img

        # ---------------------------------------------------------------------------
        plot_saccades = saccades is not None
        plot_samples = samples is not None
        _img_cache = OrderedDict()

        trial_idx = fixations["trial_number"][0]
        if (
            isinstance(trial_idx, (float, np.floating))
            and float(trial_idx).is_integer()
        ):
            trial_idx = int(trial_idx)

        # ---- time filter ----------------------------------------------------------
        if tmin is not None and tmax is not None:
            fixations = fixations.filter(pl.col("tStart").is_between(tmin, tmax))
            if plot_saccades:
                saccades = saccades.filter(pl.col("tStart").is_between(tmin, tmax))
            if plot_samples:
                samples = samples.filter(pl.col("tSample").is_between(tmin, tmax))

        # Rows outside any named phase carry an empty ``phase``. Dropping them
        # is right for a segmented recording, where they fall between trials.
        # But a recording with no named phase at all -- any format whose source
        # carries no synchronisation messages, such as a plain Tobii or
        # GazePoint export -- would then lose every row and plot nothing, with
        # no clue as to why. Keep those rows as a single unnamed phase instead,
        # and say what happened.
        if fixations.get_column("phase").fill_null("").eq("").all():
            warnings.warn(
                "No named trial phase was found, so every fixation is plotted "
                "as a single unnamed phase. Pass start_msgs and end_msgs to "
                "compute_derivatives_for_dataset to segment the recording into "
                "named phases.",
                UserWarning,
                stacklevel=2,
            )
        else:
            fixations = fixations.filter(pl.col("phase") != "")
            if plot_saccades:
                saccades = saccades.filter(pl.col("phase") != "")
            if plot_samples:
                samples = samples.filter(pl.col("phase") != "")

        # Rows the preprocessing step flagged as bad hold gaze that fell off
        # the screen or was never tracked. Drawing them stretches the scanpath
        # towards coordinates the participant never looked at, and joins them
        # with lines that cross the whole stimulus.
        def _drop_bad(frame: pl.DataFrame) -> pl.DataFrame:
            if frame is None or "bad" not in frame.columns:
                return frame
            return frame.filter(
                ~pl.col("bad").cast(pl.Boolean, strict=False).fill_null(False)
            )

        fixations = _drop_bad(fixations)
        if plot_saccades:
            saccades = _drop_bad(saccades)
        if plot_samples:
            samples = _drop_bad(samples)
        if fixations.is_empty():
            warnings.warn(
                "Every fixation was flagged as bad, so there is nothing to "
                "plot. Check the screen size passed to "
                "compute_derivatives_for_dataset.",
                UserWarning,
                stacklevel=2,
            )
            return

        # ---- split once by phase --------------------------------------------------
        fix_by_phase = fixations.partition_by("phase", as_dict=True)
        sac_by_phase = (
            saccades.partition_by("phase", as_dict=True) if plot_saccades else {}
        )
        samp_by_phase = (
            samples.partition_by("phase", as_dict=True) if plot_samples else {}
        )

        # colour map shared across phases
        cmap = plt.cm.rainbow

        # ---- build & draw ---------------------------------------------------------
        interactive_before = plt.isinteractive()
        if not display:
            plt.ioff()

        for phase, phase_fix in fix_by_phase.items():
            if phase_fix.is_empty():
                continue
            phase_name = phase[0] if isinstance(phase, tuple) else phase

            # ---------- vectors (zero‑copy) -----------------
            fx, fy, fdur = phase_fix.select(["xAvg", "yAvg", "duration"]).to_numpy().T
            n_fix = fx.size
            fix_idx = np.arange(1, n_fix + 1)

            # One colour band per fixation is only possible while the colormap
            # has enough of them. Long recordings hold thousands of fixations,
            # so fall back to a continuous scale instead of raising.
            if n_fix < cmap.N:
                norm = mplcolors.BoundaryNorm(np.arange(1, n_fix + 2), cmap.N)
            else:
                norm = mplcolors.Normalize(vmin=1, vmax=max(n_fix, 2))

            # saccades
            sac_t = (
                sac_by_phase[phase]["tStart"].to_numpy()
                if plot_saccades and phase in sac_by_phase
                else np.empty(0)
            )

            # samples
            if plot_samples and phase in samp_by_phase and samp_by_phase[phase].height:
                samp_phase = samp_by_phase[phase]
                t0 = samp_phase["tSample"][0]
                ts = samp_phase["tSample"].to_numpy() - t0
                get = samp_phase.get_column
                lx = get("LX").to_numpy() if "LX" in samp_phase.columns else None
                ly = get("LY").to_numpy() if "LY" in samp_phase.columns else None
                rx = get("RX").to_numpy() if "RX" in samp_phase.columns else None
                ry = get("RY").to_numpy() if "RY" in samp_phase.columns else None
                gx = get("X").to_numpy() if "X" in samp_phase.columns else None
                gy = get("Y").to_numpy() if "Y" in samp_phase.columns else None
            else:
                t0 = None

            # ---------- figure -----------------------------
            fig, ax_main, ax_gaze = _make_axes(plot_samples and t0 is not None)
            # scatter fixations
            sc = ax_main.scatter(
                fx,
                fy,
                c=fix_idx,
                s=fdur,
                cmap=cmap,
                norm=norm,
                alpha=0.5,
                zorder=2,
            )
            fig.colorbar(
                sc,
                ax=ax_main,
                ticks=[1, n_fix // 2 if n_fix > 2 else n_fix, n_fix],
                fraction=0.046,
                pad=0.04,
            ).set_label("# of fixation")

            # ---------- stimulus imagery / bbox ------------
            if phase_data and phase_name in phase_data:
                pdict = phase_data[phase_name]
                coords = pdict.get("img_plot_coords") or []
                bbox = pdict.get("bbox", None)
                for img_path, box in zip(pdict.get("img_paths", []), coords):
                    ax_main.imshow(
                        _maybe_cache_img(img_path),
                        extent=[box[0], box[2], box[3], box[1]],
                        zorder=0,
                    )
                if bbox is not None:
                    x1, y1, x2, y2 = bbox
                    ax_main.plot(
                        [x1, x2, x2, x1, x1],
                        [y1, y1, y2, y2, y1],
                        color="red",
                        linewidth=1.5,
                        zorder=3,
                    )

            # ---------- gaze traces ------------------------
            if ax_gaze is not None:
                if lx is not None:
                    ax_main.plot(lx, ly, "--", color="C0", zorder=1)
                    ax_gaze.plot(ts, lx, label="Left X")
                    ax_gaze.plot(ts, ly, label="Left Y")
                if rx is not None:
                    ax_main.plot(rx, ry, "--", color="k", zorder=1)
                    ax_gaze.plot(ts, rx, label="Right X")
                    ax_gaze.plot(ts, ry, label="Right Y")
                if gx is not None:
                    ax_main.plot(gx, gy, "--", color="k", zorder=1, alpha=0.6)
                    ax_gaze.plot(ts, gx, label="X")
                    ax_gaze.plot(ts, gy, label="Y")

                # fixation spans
                bars = np.c_[
                    phase_fix["tStart"].to_numpy() - t0,
                    phase_fix["duration"].to_numpy(),
                ]
                height = ax_gaze.get_ylim()[1] - ax_gaze.get_ylim()[0]
                colors = cmap(norm(fix_idx))

                # Draw all bars in one call; no BrokenBarHCollection import needed
                ax_gaze.broken_barh(bars, (0, height), facecolors=colors, alpha=0.4)
                # saccades
                if sac_t.size:
                    ymin, ymax = ax_gaze.get_ylim()
                    ax_gaze.vlines(
                        sac_t - t0,
                        ymin,
                        ymax,
                        colors="red",
                        linestyles="--",
                        linewidth=0.8,
                    )

                # tidy gaze axis
                h, l = ax_gaze.get_legend_handles_labels()
                by_label = {lab: hdl for hdl, lab in zip(h, l)}
                ax_gaze.legend(
                    by_label.values(),
                    by_label.keys(),
                    loc="center left",
                    bbox_to_anchor=(1, 0.5),
                )
                ax_gaze.set_ylabel("Gaze")
                ax_gaze.set_xlabel("Time [ms]")

            fig.tight_layout()

            # ---------- save / show ------------------------
            if folder_path:
                scan_name = f"scanpath_{trial_idx}"
                if tmin is not None and tmax is not None:
                    scan_name += f"_{tmin}_{tmax}"
                out = Path(folder_path) / f"{scan_name}_{phase_name or 'unphased'}.png"
                fig.savefig(out, dpi=150)

            if display:
                plt.show()
            plt.close(fig)

        if not display and interactive_before:
            plt.ion()

    def fix_duration(self, fixations: pl.DataFrame, axs=None):
        """Plot the distribution of fixation durations.

        Parameters
        ----------
        fixations : polars.DataFrame
            Fixation table containing a ``duration`` column, in milliseconds.
        axs : matplotlib.axes.Axes, optional
            Axes to draw on. A new figure is created when omitted.
        """

        ax = axs
        if ax is None:
            _, ax = plt.subplots()

        ax.hist(
            fixations.select(pl.col("duration")).to_numpy().ravel(),
            bins=100,
            edgecolor="black",
            linewidth=1.2,
            density=True,
        )
        ax.set_title("Fixation duration")
        ax.set_xlabel("Time (ms)")
        ax.set_ylabel("Density")

    def sacc_amplitude(self, saccades: pl.DataFrame, axs=None):
        """Plot the distribution of saccade amplitudes.

        Amplitudes are histogrammed over the 0-20 degree range.

        Parameters
        ----------
        saccades : polars.DataFrame
            Saccade table containing an ``ampDeg`` column, in degrees of visual
            angle.
        axs : matplotlib.axes.Axes, optional
            Axes to draw on. A new figure is created when omitted.
        """

        ax = axs
        if ax is None:
            _, ax = plt.subplots()

        saccades_amp = saccades.select(pl.col("ampDeg")).to_numpy().ravel()
        ax.hist(
            saccades_amp,
            bins=100,
            range=(0, 20),
            edgecolor="black",
            linewidth=1.2,
            density=True,
        )
        ax.set_title("Saccades amplitude")
        ax.set_xlabel("Amplitude (deg)")
        ax.set_ylabel("Density")

    def sacc_direction(self, saccades: pl.DataFrame, axs=None, figs=None):
        """Plot saccade directions as a polar histogram.

        Requires the direction columns produced by
        :meth:`~pyxations.PreProcessing.saccades_direction`.

        Parameters
        ----------
        saccades : polars.DataFrame
            Saccade table containing the ``deg`` and ``dir`` columns.
        axs : matplotlib.axes.Axes, optional
            Axes to replace with a polar subplot. A new polar figure is created
            when omitted.
        figs : matplotlib.figure.Figure, optional
            Figure in which the polar subplot is created. Required when ``axs``
            is given, since polar axes cannot be added to existing Cartesian
            ones.

        Raises
        ------
        ValueError
            If the ``deg`` or ``dir`` columns are missing, meaning saccade
            directions were not computed yet.
        """

        ax = axs
        if ax is None:
            plt.figure()
            ax = plt.subplot(polar=True)
        else:
            ax.set_axis_off()
            ax = figs.add_subplot(2, 2, 3, projection="polar")
        if "deg" not in saccades.columns or "dir" not in saccades.columns:
            raise ValueError(
                "Compute saccades direction first by using saccades_direction function from the PreProcessing module."
            )
        if saccades.is_empty():
            ax.set_title("Saccades direction")
            ax.set_yticklabels([])
            return
        # Convert from deg to rad
        saccades_rad = saccades.select(pl.col("deg")).to_numpy().ravel() * np.pi / 180

        n_bins = 24
        ang_hist, bin_edges = np.histogram(saccades_rad, bins=24, density=True)
        bin_centers = [
            np.mean((bin_edges[i], bin_edges[i + 1])) for i in range(len(bin_edges) - 1)
        ]

        bars = ax.bar(
            bin_centers,
            ang_hist,
            width=2 * np.pi / n_bins,
            bottom=0.0,
            alpha=0.4,
            edgecolor="black",
        )
        ax.set_title("Saccades direction")
        ax.set_yticklabels([])

        maximum = np.max(ang_hist)
        for radius, bar in zip(ang_hist, bars):
            bar.set_facecolor(plt.cm.Blues(radius / maximum if maximum else 0))

    def sacc_main_sequence(self, saccades: pl.DataFrame, axs=None, hline=None):
        """Plot the saccadic main sequence: peak velocity against amplitude.

        Drawn as a 2D histogram on logarithmic axes. Saccades with
        non-finite or non-positive amplitude or peak velocity are excluded,
        since they cannot be placed on a log scale.

        Parameters
        ----------
        saccades : polars.DataFrame
            Saccade table containing ``ampDeg`` and ``vPeak`` columns.
        axs : matplotlib.axes.Axes, optional
            Axes to draw on. A new figure is created when omitted.
        hline : float, optional
            Peak-velocity value at which to draw a labelled horizontal
            reference line, useful for marking a detection threshold.
        """

        ax = axs
        if ax is None:
            _, ax = plt.subplots()
        # Logarithmic bins
        XL = np.log10(25)  # Adjusted to fit the xlim
        YL = np.log10(1000)  # Adjusted to fit the ylim

        valid = saccades.filter(
            pl.col("vPeak").cast(pl.Float64, strict=False).is_finite()
            & pl.col("ampDeg").cast(pl.Float64, strict=False).is_finite()
            & (pl.col("vPeak") > 0)
            & (pl.col("ampDeg") > 0)
        )
        saccades_peak_vel = valid.select(pl.col("vPeak")).to_numpy().ravel()
        saccades_amp = valid.select(pl.col("ampDeg")).to_numpy().ravel()

        # Create a 2D histogram with logarithmic bins
        ax.hist2d(
            saccades_amp,
            saccades_peak_vel,
            bins=[np.logspace(-1, XL, 50), np.logspace(0, YL, 50)],
        )

        if hline:
            ax.hlines(
                y=hline,
                xmin=ax.get_xlim()[0],
                xmax=ax.get_xlim()[1],
                colors="grey",
                linestyles="--",
                label=hline,
            )
            ax.legend()
        ax.set_yscale("log")
        ax.set_xscale("log")
        ax.set_title("Main sequence")
        ax.set_xlabel("Amplitude (deg)")
        ax.set_ylabel("Peak velocity (deg)")
        # Set the limits of the axes
        ax.set_xlim(0.1, 25)
        ax.set_ylim(10, 1000)
        ax.set_aspect("equal")

    def plot_multipanel(
        self, fixations: pl.DataFrame, saccades: pl.DataFrame, display: bool = True
    ) -> None:
        """Create and save a diagnostic plot for every non-empty phase.

        Each 2-by-2 figure contains fixation-duration, saccade-amplitude,
        saccade-direction and main-sequence panels and is saved below the
        configured derivatives and detector directories.

        Parameters
        ----------
        fixations : polars.DataFrame
            Fixation events containing ``trial_number`` and ``phase``.
        saccades : polars.DataFrame
            Saccade events containing ``trial_number`` and ``phase``.
        display : bool, default True
            Whether to show each figure interactively in addition to saving it.
        """
        # ── paths & matplotlib style ────────────────────────────────
        folder_path: Path = self.derivatives_folder_path / self.events_detection_folder
        folder_path.mkdir(parents=True, exist_ok=True)
        plt.rcParams.update({"font.size": 12})

        # ── drop practice / invalid trials ─────────────────────────
        fixations = fixations.filter(pl.col("trial_number") != -1)
        saccades = saccades.filter(pl.col("trial_number") != -1)

        # ── collect valid phase labels (skip empty string) ─────────
        phases = (
            fixations.select(pl.col("phase").filter(pl.col("phase") != ""))
            .unique()  # unique values in this Series
            .to_series()
            .to_list()  # plain Python list of strings
        )

        # ── one figure per phase ───────────────────────────────────
        for phase in phases:
            fix_phase = fixations.filter(pl.col("phase") == phase)
            sacc_phase = saccades.filter(pl.col("phase") == phase)

            fig, axs = plt.subplots(2, 2, figsize=(12, 7))

            self.fix_duration(fix_phase, axs=axs[0, 0])
            self.sacc_main_sequence(sacc_phase, axs=axs[1, 1])
            self.sacc_direction(sacc_phase, axs=axs[1, 0], figs=fig)
            self.sacc_amplitude(sacc_phase, axs=axs[0, 1])

            fig.tight_layout()
            plt.savefig(folder_path / f"multipanel_{phase}.png")
            if display:
                plt.show()
            plt.close()

    def plot_animation(
        self,
        samples: pl.DataFrame,
        screen_height: int,
        screen_width: int,
        video_path: str | Path | None = None,
        background_image_path: str | Path | None = None,
        folder_path: str | Path | None = None,
        tmin: int | None = None,
        tmax: int | None = None,
        seconds_to_show: float | None = None,
        scale_factor: float = 0.5,
        gaze_radius: int = 10,
        gaze_color: tuple = (255, 0, 0),
        fps: float | None = None,
        output_format: str = "matplotlib",
        display: bool = True,
    ):
        """
        Create an animated visualization of eye-tracking data.

        When a video is provided, the animation syncs gaze samples with video frames.
        When no video is provided, gaze points are animated on a grey background or
        a provided background image, using the sample timestamps for timing.

        Parameters
        ----------
        samples : polars.DataFrame
            Polars DataFrame with gaze samples. Must contain 'tSample' and gaze
            position columns ('X', 'Y' or 'LX', 'LY', 'RX', 'RY').
        screen_height, screen_width : int
            Stimulus resolution in pixels.
        video_path : str or pathlib.Path, optional
            Path to a video file. If provided, gaze is overlaid on video frames.
        background_image_path : str or pathlib.Path, optional
            Path to a background image. Only used when video_path is None.
            If both are None, a grey background is used.
        folder_path : str or pathlib.Path, optional
            Directory where the animation will be saved. If None, nothing is saved.
            The file format depends on `output_format`.
        tmin, tmax : int, optional
            Time window in **ms**. If both None, the whole trial is plotted.
        seconds_to_show : float, optional
            Limit the animation to the first N seconds. If None, shows all available data.
        scale_factor : float, default 0.5
            Resolution scaling factor (1.0 = original, 0.5 = half resolution).
        gaze_radius : int, default 10
            Radius of the gaze point circle in pixels (before scaling).
        gaze_color : tuple of int, default (255, 0, 0)
            RGB tuple for gaze point color.
        fps : float, optional
            Frames per second for the animation. If None:
            - With video: uses the video's native FPS
            - Without video: defaults to 60 FPS
        output_format : {"matplotlib", "html", "mp4", "gif"}, default "matplotlib"
            Output format for saved animations:
            - "matplotlib": Show in matplotlib GUI window (default, blocking)
            - "html": Interactive HTML file (works in browsers)
            - "mp4": Video file (requires ffmpeg)
            - "gif": Animated GIF file (requires pillow)
        display : bool, default True
            If True and output_format is "html", returns an HTML object for notebooks.
            If output_format is "matplotlib", this is ignored (always shows window).
            If False, only saves to file (if folder_path is provided).

        Returns
        -------
        IPython.display.HTML or None
            Returns HTML animation if display=True and output_format="html", otherwise None.
        """
        cv2 = _load_cv2()

        import matplotlib as mpl
        from matplotlib.animation import FuncAnimation

        mpl.rcParams["animation.embed_limit"] = 100

        if scale_factor <= 0:
            raise ValueError("scale_factor must be greater than zero")
        if fps is not None and fps <= 0:
            raise ValueError("fps must be greater than zero")
        if (tmin is None) != (tmax is None):
            raise ValueError("tmin and tmax must be provided together")

        # Validate output_format
        valid_formats = ["html", "mp4", "gif", "matplotlib"]
        if output_format not in valid_formats:
            raise ValueError(
                f"output_format must be one of {valid_formats}, got '{output_format}'"
            )

        # ---- Determine gaze columns ----
        if "X" in samples.columns and "Y" in samples.columns:
            x_col, y_col = "X", "Y"
        elif "LX" in samples.columns and "LY" in samples.columns:
            x_col, y_col = "LX", "LY"
        elif "RX" in samples.columns and "RY" in samples.columns:
            x_col, y_col = "RX", "RY"
        else:
            raise ValueError(
                "Samples DataFrame must contain gaze columns (X, Y) or (LX, LY) or (RX, RY)"
            )

        # ---- Time filter ----
        if tmin is not None and tmax is not None:
            samples = samples.filter(pl.col("tSample").is_between(tmin, tmax))

        if samples.is_empty():
            raise ValueError("No samples available after time filtering")

        # ---- Drop NaN gaze values ----
        samples = samples.filter(
            pl.col(x_col).is_not_null() & pl.col(y_col).is_not_null()
        )
        samples = samples.filter(
            pl.col(x_col).cast(pl.Float64, strict=False).is_finite()
            & pl.col(y_col).cast(pl.Float64, strict=False).is_finite()
        )
        if samples.is_empty():
            raise ValueError("No finite gaze samples available")

        # ---- Calculate scaled dimensions ----
        scaled_width = int(screen_width * scale_factor)
        scaled_height = int(screen_height * scale_factor)

        trial_idx = (
            samples["trial_number"][0] if "trial_number" in samples.columns else 0
        )

        # ================= WITH VIDEO =================
        if video_path is not None:
            video_path = Path(video_path)
            if not video_path.exists():
                raise FileNotFoundError(f"Video file not found: {video_path}")

            cap = cv2.VideoCapture(str(video_path))
            video_fps = cap.get(cv2.CAP_PROP_FPS)
            total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
            if fps is None:
                fps = video_fps

            # Calculate time to frame mapping
            t_start = samples["tSample"].min()
            t_end = samples["tSample"].max()

            # Create frame-to-time mapping
            frame_edges = np.linspace(t_start, t_end, total_frames + 1)
            frame_times = ((frame_edges[:-1] + frame_edges[1:]) / 2).astype(int)

            # Build a lookup: frame_index -> list of gaze points
            samples_np = samples.select([x_col, y_col, "tSample"]).to_numpy()
            gaze_by_frame = {i: [] for i in range(total_frames)}

            for x, y, t in samples_np:
                # Find the closest frame
                frame_idx = np.searchsorted(frame_times, t, side="right") - 1
                frame_idx = max(0, min(frame_idx, total_frames - 1))
                gaze_by_frame[frame_idx].append((x, y))

            # Limit frames if seconds_to_show is set
            frames_to_show = total_frames
            if seconds_to_show is not None:
                frames_to_show = min(int(fps * seconds_to_show), total_frames)

            # Reset video
            cap.set(cv2.CAP_PROP_POS_FRAMES, 0)

            # Create figure
            fig, ax = plt.subplots(figsize=(10 * scale_factor, 6 * scale_factor))
            ax.axis("off")

            # Initialize with first frame
            ret, frame = cap.read()
            if not ret:
                cap.release()
                raise RuntimeError("Could not read first frame from video")

            frame_resized = cv2.resize(
                frame, (scaled_width, scaled_height), interpolation=cv2.INTER_AREA
            )
            frame_rgb = cv2.cvtColor(frame_resized, cv2.COLOR_BGR2RGB)
            im = ax.imshow(frame_rgb)

            def update_frame_video(frame_idx):
                ret, frame = cap.read()
                if not ret:
                    return [im]

                frame_resized = cv2.resize(
                    frame, (scaled_width, scaled_height), interpolation=cv2.INTER_AREA
                )
                frame_rgb = cv2.cvtColor(frame_resized, cv2.COLOR_BGR2RGB)

                # Draw gaze points for this frame
                for gx, gy in gaze_by_frame.get(frame_idx, []):
                    scaled_x = int(gx * scale_factor)
                    scaled_y = int(gy * scale_factor)
                    if 0 <= scaled_x < scaled_width and 0 <= scaled_y < scaled_height:
                        radius = max(3, int(gaze_radius * scale_factor))
                        cv2.circle(
                            frame_rgb,
                            (scaled_x, scaled_y),
                            radius=radius,
                            color=gaze_color,
                            thickness=-1,
                        )

                im.set_array(frame_rgb)
                return [im]

            anim = FuncAnimation(
                fig,
                update_frame_video,
                frames=frames_to_show,
                interval=1000 / fps,
                blit=True,
                repeat=True,
            )

        # ================= WITHOUT VIDEO =================
        else:
            if fps is None:
                fps = 60  # Default FPS for sample-based animation

            # Prepare background
            if background_image_path is not None:
                bg_path = Path(background_image_path)
                if not bg_path.exists():
                    raise FileNotFoundError(f"Background image not found: {bg_path}")
                bg_img = mpimg.imread(str(bg_path))
                if bg_img.dtype == np.float64:
                    bg_img = (bg_img * 255).clip(0, 255).astype(np.uint8)
                # Resize background to match screen dimensions then scale
                bg_img = cv2.resize(
                    bg_img, (scaled_width, scaled_height), interpolation=cv2.INTER_AREA
                )
            else:
                # Grey background
                bg_img = np.ones((scaled_height, scaled_width, 3), dtype=np.uint8) * 128

            # Get time range
            t_start = samples["tSample"].min()
            t_end = samples["tSample"].max()
            trial_duration = t_end - t_start

            # Limit duration if seconds_to_show is set
            if seconds_to_show is not None:
                t_end = min(t_end, t_start + int(seconds_to_show * 1000))
                samples = samples.filter(pl.col("tSample") <= t_end)
                trial_duration = t_end - t_start

            # Calculate total frames based on duration and fps
            total_frames = int((trial_duration / 1000) * fps)
            total_frames = max(total_frames, 1)

            # Create time bins for each animation frame
            frame_times = np.linspace(t_start, t_end, total_frames + 1)

            # Build gaze lookup by frame
            samples_np = samples.select([x_col, y_col, "tSample"]).to_numpy()
            gaze_by_frame = {i: [] for i in range(total_frames)}

            for x, y, t in samples_np:
                frame_idx = np.searchsorted(frame_times, t, side="right") - 1
                frame_idx = max(0, min(frame_idx, total_frames - 1))
                gaze_by_frame[frame_idx].append((x, y))

            # Create figure
            fig, ax = plt.subplots(figsize=(10 * scale_factor, 6 * scale_factor))
            ax.axis("off")

            # Initialize with background
            im = ax.imshow(bg_img.copy())

            def update_frame_no_video(frame_idx):
                # Start with fresh background copy
                frame_rgb = bg_img.copy()

                # Draw gaze points for this frame
                for gx, gy in gaze_by_frame.get(frame_idx, []):
                    scaled_x = int(gx * scale_factor)
                    scaled_y = int(gy * scale_factor)
                    if 0 <= scaled_x < scaled_width and 0 <= scaled_y < scaled_height:
                        radius = max(3, int(gaze_radius * scale_factor))
                        cv2.circle(
                            frame_rgb,
                            (scaled_x, scaled_y),
                            radius=radius,
                            color=gaze_color,
                            thickness=-1,
                        )

                im.set_array(frame_rgb)
                return [im]

            anim = FuncAnimation(
                fig,
                update_frame_no_video,
                frames=total_frames,
                interval=1000 / fps,
                blit=True,
                repeat=True,
            )

        # ================= SAVE / DISPLAY =================
        result = None
        trial_idx_val = trial_idx
        if (
            isinstance(trial_idx_val, (float, np.floating))
            and float(trial_idx_val).is_integer()
        ):
            trial_idx_val = int(trial_idx_val)

        # Build output filename
        anim_name = f"animation_{trial_idx_val}"
        if tmin is not None and tmax is not None:
            anim_name += f"_{tmin}_{tmax}"

        # Handle different output formats
        if output_format == "matplotlib":
            # Show in matplotlib GUI window (blocking)
            plt.show()
            # Cleanup video capture if used
            if video_path is not None:
                cap.release()
            return None

        elif output_format == "mp4":
            if folder_path:
                folder_path = Path(folder_path)
                folder_path.mkdir(parents=True, exist_ok=True)
                out_path = folder_path / f"{anim_name}.mp4"
                try:
                    anim.save(str(out_path), writer="ffmpeg", fps=fps)
                    print(f"Animation saved to: {out_path}")
                except (OSError, RuntimeError, ValueError) as e:
                    raise RuntimeError(
                        f"Failed to save MP4. Make sure ffmpeg is installed. Error: {e}"
                    )
            plt.close(fig)

        elif output_format == "gif":
            if folder_path:
                folder_path = Path(folder_path)
                folder_path.mkdir(parents=True, exist_ok=True)
                out_path = folder_path / f"{anim_name}.gif"
                try:
                    anim.save(str(out_path), writer="pillow", fps=fps)
                    print(f"Animation saved to: {out_path}")
                except (OSError, RuntimeError, ValueError) as e:
                    raise RuntimeError(
                        f"Failed to save GIF. Make sure pillow is installed. Error: {e}"
                    )
            plt.close(fig)

        else:  # html (default)
            if folder_path:
                folder_path = Path(folder_path)
                folder_path.mkdir(parents=True, exist_ok=True)
                out_path = folder_path / f"{anim_name}.html"
                with open(out_path, "w") as f:
                    f.write(anim.to_jshtml())
                print(f"Animation saved to: {out_path}")

            if display:
                try:
                    from IPython.display import HTML

                    plt.close(fig)
                    result = HTML(anim.to_jshtml())
                except ImportError:
                    print(
                        "IPython not available. Use output_format='matplotlib' for GUI display."
                    )
                    plt.close(fig)
            else:
                plt.close(fig)

        # Cleanup video capture if used
        if video_path is not None:
            cap.release()

        return result

fix_duration(fixations, axs=None)

Plot the distribution of fixation durations.

Parameters:

Name Type Description Default
fixations DataFrame

Fixation table containing a duration column, in milliseconds.

required
axs Axes

Axes to draw on. A new figure is created when omitted.

None
Source code in pyxations/visualization/visualization.py
def fix_duration(self, fixations: pl.DataFrame, axs=None):
    """Plot the distribution of fixation durations.

    Parameters
    ----------
    fixations : polars.DataFrame
        Fixation table containing a ``duration`` column, in milliseconds.
    axs : matplotlib.axes.Axes, optional
        Axes to draw on. A new figure is created when omitted.
    """

    ax = axs
    if ax is None:
        _, ax = plt.subplots()

    ax.hist(
        fixations.select(pl.col("duration")).to_numpy().ravel(),
        bins=100,
        edgecolor="black",
        linewidth=1.2,
        density=True,
    )
    ax.set_title("Fixation duration")
    ax.set_xlabel("Time (ms)")
    ax.set_ylabel("Density")

plot_animation(samples, screen_height, screen_width, video_path=None, background_image_path=None, folder_path=None, tmin=None, tmax=None, seconds_to_show=None, scale_factor=0.5, gaze_radius=10, gaze_color=(255, 0, 0), fps=None, output_format='matplotlib', display=True)

Create an animated visualization of eye-tracking data.

When a video is provided, the animation syncs gaze samples with video frames. When no video is provided, gaze points are animated on a grey background or a provided background image, using the sample timestamps for timing.

Parameters:

Name Type Description Default
samples DataFrame

Polars DataFrame with gaze samples. Must contain 'tSample' and gaze position columns ('X', 'Y' or 'LX', 'LY', 'RX', 'RY').

required
screen_height int

Stimulus resolution in pixels.

required
screen_width int

Stimulus resolution in pixels.

required
video_path str or Path

Path to a video file. If provided, gaze is overlaid on video frames.

None
background_image_path str or Path

Path to a background image. Only used when video_path is None. If both are None, a grey background is used.

None
folder_path str or Path

Directory where the animation will be saved. If None, nothing is saved. The file format depends on output_format.

None
tmin int

Time window in ms. If both None, the whole trial is plotted.

None
tmax int

Time window in ms. If both None, the whole trial is plotted.

None
seconds_to_show float

Limit the animation to the first N seconds. If None, shows all available data.

None
scale_factor float

Resolution scaling factor (1.0 = original, 0.5 = half resolution).

0.5
gaze_radius int

Radius of the gaze point circle in pixels (before scaling).

10
gaze_color tuple of int

RGB tuple for gaze point color.

(255, 0, 0)
fps float

Frames per second for the animation. If None: - With video: uses the video's native FPS - Without video: defaults to 60 FPS

None
output_format (matplotlib, html, mp4, gif)

Output format for saved animations: - "matplotlib": Show in matplotlib GUI window (default, blocking) - "html": Interactive HTML file (works in browsers) - "mp4": Video file (requires ffmpeg) - "gif": Animated GIF file (requires pillow)

"matplotlib"
display bool

If True and output_format is "html", returns an HTML object for notebooks. If output_format is "matplotlib", this is ignored (always shows window). If False, only saves to file (if folder_path is provided).

True

Returns:

Type Description
HTML or None

Returns HTML animation if display=True and output_format="html", otherwise None.

Source code in pyxations/visualization/visualization.py
 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
 769
 770
 771
 772
 773
 774
 775
 776
 777
 778
 779
 780
 781
 782
 783
 784
 785
 786
 787
 788
 789
 790
 791
 792
 793
 794
 795
 796
 797
 798
 799
 800
 801
 802
 803
 804
 805
 806
 807
 808
 809
 810
 811
 812
 813
 814
 815
 816
 817
 818
 819
 820
 821
 822
 823
 824
 825
 826
 827
 828
 829
 830
 831
 832
 833
 834
 835
 836
 837
 838
 839
 840
 841
 842
 843
 844
 845
 846
 847
 848
 849
 850
 851
 852
 853
 854
 855
 856
 857
 858
 859
 860
 861
 862
 863
 864
 865
 866
 867
 868
 869
 870
 871
 872
 873
 874
 875
 876
 877
 878
 879
 880
 881
 882
 883
 884
 885
 886
 887
 888
 889
 890
 891
 892
 893
 894
 895
 896
 897
 898
 899
 900
 901
 902
 903
 904
 905
 906
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
def plot_animation(
    self,
    samples: pl.DataFrame,
    screen_height: int,
    screen_width: int,
    video_path: str | Path | None = None,
    background_image_path: str | Path | None = None,
    folder_path: str | Path | None = None,
    tmin: int | None = None,
    tmax: int | None = None,
    seconds_to_show: float | None = None,
    scale_factor: float = 0.5,
    gaze_radius: int = 10,
    gaze_color: tuple = (255, 0, 0),
    fps: float | None = None,
    output_format: str = "matplotlib",
    display: bool = True,
):
    """
    Create an animated visualization of eye-tracking data.

    When a video is provided, the animation syncs gaze samples with video frames.
    When no video is provided, gaze points are animated on a grey background or
    a provided background image, using the sample timestamps for timing.

    Parameters
    ----------
    samples : polars.DataFrame
        Polars DataFrame with gaze samples. Must contain 'tSample' and gaze
        position columns ('X', 'Y' or 'LX', 'LY', 'RX', 'RY').
    screen_height, screen_width : int
        Stimulus resolution in pixels.
    video_path : str or pathlib.Path, optional
        Path to a video file. If provided, gaze is overlaid on video frames.
    background_image_path : str or pathlib.Path, optional
        Path to a background image. Only used when video_path is None.
        If both are None, a grey background is used.
    folder_path : str or pathlib.Path, optional
        Directory where the animation will be saved. If None, nothing is saved.
        The file format depends on `output_format`.
    tmin, tmax : int, optional
        Time window in **ms**. If both None, the whole trial is plotted.
    seconds_to_show : float, optional
        Limit the animation to the first N seconds. If None, shows all available data.
    scale_factor : float, default 0.5
        Resolution scaling factor (1.0 = original, 0.5 = half resolution).
    gaze_radius : int, default 10
        Radius of the gaze point circle in pixels (before scaling).
    gaze_color : tuple of int, default (255, 0, 0)
        RGB tuple for gaze point color.
    fps : float, optional
        Frames per second for the animation. If None:
        - With video: uses the video's native FPS
        - Without video: defaults to 60 FPS
    output_format : {"matplotlib", "html", "mp4", "gif"}, default "matplotlib"
        Output format for saved animations:
        - "matplotlib": Show in matplotlib GUI window (default, blocking)
        - "html": Interactive HTML file (works in browsers)
        - "mp4": Video file (requires ffmpeg)
        - "gif": Animated GIF file (requires pillow)
    display : bool, default True
        If True and output_format is "html", returns an HTML object for notebooks.
        If output_format is "matplotlib", this is ignored (always shows window).
        If False, only saves to file (if folder_path is provided).

    Returns
    -------
    IPython.display.HTML or None
        Returns HTML animation if display=True and output_format="html", otherwise None.
    """
    cv2 = _load_cv2()

    import matplotlib as mpl
    from matplotlib.animation import FuncAnimation

    mpl.rcParams["animation.embed_limit"] = 100

    if scale_factor <= 0:
        raise ValueError("scale_factor must be greater than zero")
    if fps is not None and fps <= 0:
        raise ValueError("fps must be greater than zero")
    if (tmin is None) != (tmax is None):
        raise ValueError("tmin and tmax must be provided together")

    # Validate output_format
    valid_formats = ["html", "mp4", "gif", "matplotlib"]
    if output_format not in valid_formats:
        raise ValueError(
            f"output_format must be one of {valid_formats}, got '{output_format}'"
        )

    # ---- Determine gaze columns ----
    if "X" in samples.columns and "Y" in samples.columns:
        x_col, y_col = "X", "Y"
    elif "LX" in samples.columns and "LY" in samples.columns:
        x_col, y_col = "LX", "LY"
    elif "RX" in samples.columns and "RY" in samples.columns:
        x_col, y_col = "RX", "RY"
    else:
        raise ValueError(
            "Samples DataFrame must contain gaze columns (X, Y) or (LX, LY) or (RX, RY)"
        )

    # ---- Time filter ----
    if tmin is not None and tmax is not None:
        samples = samples.filter(pl.col("tSample").is_between(tmin, tmax))

    if samples.is_empty():
        raise ValueError("No samples available after time filtering")

    # ---- Drop NaN gaze values ----
    samples = samples.filter(
        pl.col(x_col).is_not_null() & pl.col(y_col).is_not_null()
    )
    samples = samples.filter(
        pl.col(x_col).cast(pl.Float64, strict=False).is_finite()
        & pl.col(y_col).cast(pl.Float64, strict=False).is_finite()
    )
    if samples.is_empty():
        raise ValueError("No finite gaze samples available")

    # ---- Calculate scaled dimensions ----
    scaled_width = int(screen_width * scale_factor)
    scaled_height = int(screen_height * scale_factor)

    trial_idx = (
        samples["trial_number"][0] if "trial_number" in samples.columns else 0
    )

    # ================= WITH VIDEO =================
    if video_path is not None:
        video_path = Path(video_path)
        if not video_path.exists():
            raise FileNotFoundError(f"Video file not found: {video_path}")

        cap = cv2.VideoCapture(str(video_path))
        video_fps = cap.get(cv2.CAP_PROP_FPS)
        total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
        if fps is None:
            fps = video_fps

        # Calculate time to frame mapping
        t_start = samples["tSample"].min()
        t_end = samples["tSample"].max()

        # Create frame-to-time mapping
        frame_edges = np.linspace(t_start, t_end, total_frames + 1)
        frame_times = ((frame_edges[:-1] + frame_edges[1:]) / 2).astype(int)

        # Build a lookup: frame_index -> list of gaze points
        samples_np = samples.select([x_col, y_col, "tSample"]).to_numpy()
        gaze_by_frame = {i: [] for i in range(total_frames)}

        for x, y, t in samples_np:
            # Find the closest frame
            frame_idx = np.searchsorted(frame_times, t, side="right") - 1
            frame_idx = max(0, min(frame_idx, total_frames - 1))
            gaze_by_frame[frame_idx].append((x, y))

        # Limit frames if seconds_to_show is set
        frames_to_show = total_frames
        if seconds_to_show is not None:
            frames_to_show = min(int(fps * seconds_to_show), total_frames)

        # Reset video
        cap.set(cv2.CAP_PROP_POS_FRAMES, 0)

        # Create figure
        fig, ax = plt.subplots(figsize=(10 * scale_factor, 6 * scale_factor))
        ax.axis("off")

        # Initialize with first frame
        ret, frame = cap.read()
        if not ret:
            cap.release()
            raise RuntimeError("Could not read first frame from video")

        frame_resized = cv2.resize(
            frame, (scaled_width, scaled_height), interpolation=cv2.INTER_AREA
        )
        frame_rgb = cv2.cvtColor(frame_resized, cv2.COLOR_BGR2RGB)
        im = ax.imshow(frame_rgb)

        def update_frame_video(frame_idx):
            ret, frame = cap.read()
            if not ret:
                return [im]

            frame_resized = cv2.resize(
                frame, (scaled_width, scaled_height), interpolation=cv2.INTER_AREA
            )
            frame_rgb = cv2.cvtColor(frame_resized, cv2.COLOR_BGR2RGB)

            # Draw gaze points for this frame
            for gx, gy in gaze_by_frame.get(frame_idx, []):
                scaled_x = int(gx * scale_factor)
                scaled_y = int(gy * scale_factor)
                if 0 <= scaled_x < scaled_width and 0 <= scaled_y < scaled_height:
                    radius = max(3, int(gaze_radius * scale_factor))
                    cv2.circle(
                        frame_rgb,
                        (scaled_x, scaled_y),
                        radius=radius,
                        color=gaze_color,
                        thickness=-1,
                    )

            im.set_array(frame_rgb)
            return [im]

        anim = FuncAnimation(
            fig,
            update_frame_video,
            frames=frames_to_show,
            interval=1000 / fps,
            blit=True,
            repeat=True,
        )

    # ================= WITHOUT VIDEO =================
    else:
        if fps is None:
            fps = 60  # Default FPS for sample-based animation

        # Prepare background
        if background_image_path is not None:
            bg_path = Path(background_image_path)
            if not bg_path.exists():
                raise FileNotFoundError(f"Background image not found: {bg_path}")
            bg_img = mpimg.imread(str(bg_path))
            if bg_img.dtype == np.float64:
                bg_img = (bg_img * 255).clip(0, 255).astype(np.uint8)
            # Resize background to match screen dimensions then scale
            bg_img = cv2.resize(
                bg_img, (scaled_width, scaled_height), interpolation=cv2.INTER_AREA
            )
        else:
            # Grey background
            bg_img = np.ones((scaled_height, scaled_width, 3), dtype=np.uint8) * 128

        # Get time range
        t_start = samples["tSample"].min()
        t_end = samples["tSample"].max()
        trial_duration = t_end - t_start

        # Limit duration if seconds_to_show is set
        if seconds_to_show is not None:
            t_end = min(t_end, t_start + int(seconds_to_show * 1000))
            samples = samples.filter(pl.col("tSample") <= t_end)
            trial_duration = t_end - t_start

        # Calculate total frames based on duration and fps
        total_frames = int((trial_duration / 1000) * fps)
        total_frames = max(total_frames, 1)

        # Create time bins for each animation frame
        frame_times = np.linspace(t_start, t_end, total_frames + 1)

        # Build gaze lookup by frame
        samples_np = samples.select([x_col, y_col, "tSample"]).to_numpy()
        gaze_by_frame = {i: [] for i in range(total_frames)}

        for x, y, t in samples_np:
            frame_idx = np.searchsorted(frame_times, t, side="right") - 1
            frame_idx = max(0, min(frame_idx, total_frames - 1))
            gaze_by_frame[frame_idx].append((x, y))

        # Create figure
        fig, ax = plt.subplots(figsize=(10 * scale_factor, 6 * scale_factor))
        ax.axis("off")

        # Initialize with background
        im = ax.imshow(bg_img.copy())

        def update_frame_no_video(frame_idx):
            # Start with fresh background copy
            frame_rgb = bg_img.copy()

            # Draw gaze points for this frame
            for gx, gy in gaze_by_frame.get(frame_idx, []):
                scaled_x = int(gx * scale_factor)
                scaled_y = int(gy * scale_factor)
                if 0 <= scaled_x < scaled_width and 0 <= scaled_y < scaled_height:
                    radius = max(3, int(gaze_radius * scale_factor))
                    cv2.circle(
                        frame_rgb,
                        (scaled_x, scaled_y),
                        radius=radius,
                        color=gaze_color,
                        thickness=-1,
                    )

            im.set_array(frame_rgb)
            return [im]

        anim = FuncAnimation(
            fig,
            update_frame_no_video,
            frames=total_frames,
            interval=1000 / fps,
            blit=True,
            repeat=True,
        )

    # ================= SAVE / DISPLAY =================
    result = None
    trial_idx_val = trial_idx
    if (
        isinstance(trial_idx_val, (float, np.floating))
        and float(trial_idx_val).is_integer()
    ):
        trial_idx_val = int(trial_idx_val)

    # Build output filename
    anim_name = f"animation_{trial_idx_val}"
    if tmin is not None and tmax is not None:
        anim_name += f"_{tmin}_{tmax}"

    # Handle different output formats
    if output_format == "matplotlib":
        # Show in matplotlib GUI window (blocking)
        plt.show()
        # Cleanup video capture if used
        if video_path is not None:
            cap.release()
        return None

    elif output_format == "mp4":
        if folder_path:
            folder_path = Path(folder_path)
            folder_path.mkdir(parents=True, exist_ok=True)
            out_path = folder_path / f"{anim_name}.mp4"
            try:
                anim.save(str(out_path), writer="ffmpeg", fps=fps)
                print(f"Animation saved to: {out_path}")
            except (OSError, RuntimeError, ValueError) as e:
                raise RuntimeError(
                    f"Failed to save MP4. Make sure ffmpeg is installed. Error: {e}"
                )
        plt.close(fig)

    elif output_format == "gif":
        if folder_path:
            folder_path = Path(folder_path)
            folder_path.mkdir(parents=True, exist_ok=True)
            out_path = folder_path / f"{anim_name}.gif"
            try:
                anim.save(str(out_path), writer="pillow", fps=fps)
                print(f"Animation saved to: {out_path}")
            except (OSError, RuntimeError, ValueError) as e:
                raise RuntimeError(
                    f"Failed to save GIF. Make sure pillow is installed. Error: {e}"
                )
        plt.close(fig)

    else:  # html (default)
        if folder_path:
            folder_path = Path(folder_path)
            folder_path.mkdir(parents=True, exist_ok=True)
            out_path = folder_path / f"{anim_name}.html"
            with open(out_path, "w") as f:
                f.write(anim.to_jshtml())
            print(f"Animation saved to: {out_path}")

        if display:
            try:
                from IPython.display import HTML

                plt.close(fig)
                result = HTML(anim.to_jshtml())
            except ImportError:
                print(
                    "IPython not available. Use output_format='matplotlib' for GUI display."
                )
                plt.close(fig)
        else:
            plt.close(fig)

    # Cleanup video capture if used
    if video_path is not None:
        cap.release()

    return result

plot_multipanel(fixations, saccades, display=True)

Create and save a diagnostic plot for every non-empty phase.

Each 2-by-2 figure contains fixation-duration, saccade-amplitude, saccade-direction and main-sequence panels and is saved below the configured derivatives and detector directories.

Parameters:

Name Type Description Default
fixations DataFrame

Fixation events containing trial_number and phase.

required
saccades DataFrame

Saccade events containing trial_number and phase.

required
display bool

Whether to show each figure interactively in addition to saving it.

True
Source code in pyxations/visualization/visualization.py
def plot_multipanel(
    self, fixations: pl.DataFrame, saccades: pl.DataFrame, display: bool = True
) -> None:
    """Create and save a diagnostic plot for every non-empty phase.

    Each 2-by-2 figure contains fixation-duration, saccade-amplitude,
    saccade-direction and main-sequence panels and is saved below the
    configured derivatives and detector directories.

    Parameters
    ----------
    fixations : polars.DataFrame
        Fixation events containing ``trial_number`` and ``phase``.
    saccades : polars.DataFrame
        Saccade events containing ``trial_number`` and ``phase``.
    display : bool, default True
        Whether to show each figure interactively in addition to saving it.
    """
    # ── paths & matplotlib style ────────────────────────────────
    folder_path: Path = self.derivatives_folder_path / self.events_detection_folder
    folder_path.mkdir(parents=True, exist_ok=True)
    plt.rcParams.update({"font.size": 12})

    # ── drop practice / invalid trials ─────────────────────────
    fixations = fixations.filter(pl.col("trial_number") != -1)
    saccades = saccades.filter(pl.col("trial_number") != -1)

    # ── collect valid phase labels (skip empty string) ─────────
    phases = (
        fixations.select(pl.col("phase").filter(pl.col("phase") != ""))
        .unique()  # unique values in this Series
        .to_series()
        .to_list()  # plain Python list of strings
    )

    # ── one figure per phase ───────────────────────────────────
    for phase in phases:
        fix_phase = fixations.filter(pl.col("phase") == phase)
        sacc_phase = saccades.filter(pl.col("phase") == phase)

        fig, axs = plt.subplots(2, 2, figsize=(12, 7))

        self.fix_duration(fix_phase, axs=axs[0, 0])
        self.sacc_main_sequence(sacc_phase, axs=axs[1, 1])
        self.sacc_direction(sacc_phase, axs=axs[1, 0], figs=fig)
        self.sacc_amplitude(sacc_phase, axs=axs[0, 1])

        fig.tight_layout()
        plt.savefig(folder_path / f"multipanel_{phase}.png")
        if display:
            plt.show()
        plt.close()

sacc_amplitude(saccades, axs=None)

Plot the distribution of saccade amplitudes.

Amplitudes are histogrammed over the 0-20 degree range.

Parameters:

Name Type Description Default
saccades DataFrame

Saccade table containing an ampDeg column, in degrees of visual angle.

required
axs Axes

Axes to draw on. A new figure is created when omitted.

None
Source code in pyxations/visualization/visualization.py
def sacc_amplitude(self, saccades: pl.DataFrame, axs=None):
    """Plot the distribution of saccade amplitudes.

    Amplitudes are histogrammed over the 0-20 degree range.

    Parameters
    ----------
    saccades : polars.DataFrame
        Saccade table containing an ``ampDeg`` column, in degrees of visual
        angle.
    axs : matplotlib.axes.Axes, optional
        Axes to draw on. A new figure is created when omitted.
    """

    ax = axs
    if ax is None:
        _, ax = plt.subplots()

    saccades_amp = saccades.select(pl.col("ampDeg")).to_numpy().ravel()
    ax.hist(
        saccades_amp,
        bins=100,
        range=(0, 20),
        edgecolor="black",
        linewidth=1.2,
        density=True,
    )
    ax.set_title("Saccades amplitude")
    ax.set_xlabel("Amplitude (deg)")
    ax.set_ylabel("Density")

sacc_direction(saccades, axs=None, figs=None)

Plot saccade directions as a polar histogram.

Requires the direction columns produced by :meth:~pyxations.PreProcessing.saccades_direction.

Parameters:

Name Type Description Default
saccades DataFrame

Saccade table containing the deg and dir columns.

required
axs Axes

Axes to replace with a polar subplot. A new polar figure is created when omitted.

None
figs Figure

Figure in which the polar subplot is created. Required when axs is given, since polar axes cannot be added to existing Cartesian ones.

None

Raises:

Type Description
ValueError

If the deg or dir columns are missing, meaning saccade directions were not computed yet.

Source code in pyxations/visualization/visualization.py
def sacc_direction(self, saccades: pl.DataFrame, axs=None, figs=None):
    """Plot saccade directions as a polar histogram.

    Requires the direction columns produced by
    :meth:`~pyxations.PreProcessing.saccades_direction`.

    Parameters
    ----------
    saccades : polars.DataFrame
        Saccade table containing the ``deg`` and ``dir`` columns.
    axs : matplotlib.axes.Axes, optional
        Axes to replace with a polar subplot. A new polar figure is created
        when omitted.
    figs : matplotlib.figure.Figure, optional
        Figure in which the polar subplot is created. Required when ``axs``
        is given, since polar axes cannot be added to existing Cartesian
        ones.

    Raises
    ------
    ValueError
        If the ``deg`` or ``dir`` columns are missing, meaning saccade
        directions were not computed yet.
    """

    ax = axs
    if ax is None:
        plt.figure()
        ax = plt.subplot(polar=True)
    else:
        ax.set_axis_off()
        ax = figs.add_subplot(2, 2, 3, projection="polar")
    if "deg" not in saccades.columns or "dir" not in saccades.columns:
        raise ValueError(
            "Compute saccades direction first by using saccades_direction function from the PreProcessing module."
        )
    if saccades.is_empty():
        ax.set_title("Saccades direction")
        ax.set_yticklabels([])
        return
    # Convert from deg to rad
    saccades_rad = saccades.select(pl.col("deg")).to_numpy().ravel() * np.pi / 180

    n_bins = 24
    ang_hist, bin_edges = np.histogram(saccades_rad, bins=24, density=True)
    bin_centers = [
        np.mean((bin_edges[i], bin_edges[i + 1])) for i in range(len(bin_edges) - 1)
    ]

    bars = ax.bar(
        bin_centers,
        ang_hist,
        width=2 * np.pi / n_bins,
        bottom=0.0,
        alpha=0.4,
        edgecolor="black",
    )
    ax.set_title("Saccades direction")
    ax.set_yticklabels([])

    maximum = np.max(ang_hist)
    for radius, bar in zip(ang_hist, bars):
        bar.set_facecolor(plt.cm.Blues(radius / maximum if maximum else 0))

sacc_main_sequence(saccades, axs=None, hline=None)

Plot the saccadic main sequence: peak velocity against amplitude.

Drawn as a 2D histogram on logarithmic axes. Saccades with non-finite or non-positive amplitude or peak velocity are excluded, since they cannot be placed on a log scale.

Parameters:

Name Type Description Default
saccades DataFrame

Saccade table containing ampDeg and vPeak columns.

required
axs Axes

Axes to draw on. A new figure is created when omitted.

None
hline float

Peak-velocity value at which to draw a labelled horizontal reference line, useful for marking a detection threshold.

None
Source code in pyxations/visualization/visualization.py
def sacc_main_sequence(self, saccades: pl.DataFrame, axs=None, hline=None):
    """Plot the saccadic main sequence: peak velocity against amplitude.

    Drawn as a 2D histogram on logarithmic axes. Saccades with
    non-finite or non-positive amplitude or peak velocity are excluded,
    since they cannot be placed on a log scale.

    Parameters
    ----------
    saccades : polars.DataFrame
        Saccade table containing ``ampDeg`` and ``vPeak`` columns.
    axs : matplotlib.axes.Axes, optional
        Axes to draw on. A new figure is created when omitted.
    hline : float, optional
        Peak-velocity value at which to draw a labelled horizontal
        reference line, useful for marking a detection threshold.
    """

    ax = axs
    if ax is None:
        _, ax = plt.subplots()
    # Logarithmic bins
    XL = np.log10(25)  # Adjusted to fit the xlim
    YL = np.log10(1000)  # Adjusted to fit the ylim

    valid = saccades.filter(
        pl.col("vPeak").cast(pl.Float64, strict=False).is_finite()
        & pl.col("ampDeg").cast(pl.Float64, strict=False).is_finite()
        & (pl.col("vPeak") > 0)
        & (pl.col("ampDeg") > 0)
    )
    saccades_peak_vel = valid.select(pl.col("vPeak")).to_numpy().ravel()
    saccades_amp = valid.select(pl.col("ampDeg")).to_numpy().ravel()

    # Create a 2D histogram with logarithmic bins
    ax.hist2d(
        saccades_amp,
        saccades_peak_vel,
        bins=[np.logspace(-1, XL, 50), np.logspace(0, YL, 50)],
    )

    if hline:
        ax.hlines(
            y=hline,
            xmin=ax.get_xlim()[0],
            xmax=ax.get_xlim()[1],
            colors="grey",
            linestyles="--",
            label=hline,
        )
        ax.legend()
    ax.set_yscale("log")
    ax.set_xscale("log")
    ax.set_title("Main sequence")
    ax.set_xlabel("Amplitude (deg)")
    ax.set_ylabel("Peak velocity (deg)")
    # Set the limits of the axes
    ax.set_xlim(0.1, 25)
    ax.set_ylim(10, 1000)
    ax.set_aspect("equal")

scanpath(fixations, screen_height, screen_width, folder_path=None, tmin=None, tmax=None, saccades=None, samples=None, phase_data=None, display=True)

Fast scan‑path visualiser.

• Vectorised: no per‑row Python loops • Single pass phase grouping • Uses BrokenBarHCollection for fixation spans • Each requested PNG is written once

Parameters:

Name Type Description Default
fixations DataFrame

Polars DataFrame with at least tStart, duration, xAvg, yAvg, phase.

required
screen_height int

Stimulus resolution in pixels.

required
screen_width int

Stimulus resolution in pixels.

required
folder_path str or Path

Directory where 1 PNG per phase will be stored. If None, nothing is saved.

None
tmin int

Time window in ms. If both None, the whole trial is plotted.

None
tmax int

Time window in ms. If both None, the whole trial is plotted.

None
saccades DataFrame

Polars DataFrame with tStart, phase, … (optional).

None
samples DataFrame

Polars DataFrame with gaze traces (tSample, LX, LY, RX, RY or X, Y) (optional).

None
phase_data dict

Per‑phase extras::

{
    "search": {
        "img_paths": [...],
        "img_plot_coords": [(x1,y1,x2,y2), ...],
        "bbox": (x1,y1,x2,y2),
    },
    ...
}
None
display bool

If False the figure canvas is never shown (faster for batch jobs).

True
Notes

One figure is produced per named trial phase. Fixations that fall outside every phase are skipped. If no fixation carries a phase name at all, which happens when the recording was never segmented, they are all drawn as a single unnamed phase and a :class:UserWarning is issued.

Source code in pyxations/visualization/visualization.py
def scanpath(
    self,
    fixations: pl.DataFrame,
    screen_height: int,
    screen_width: int,
    folder_path: str | Path | None = None,
    tmin: int | None = None,
    tmax: int | None = None,
    saccades: pl.DataFrame | None = None,
    samples: pl.DataFrame | None = None,
    phase_data: dict[str, dict] | None = None,
    display: bool = True,
):
    """
    Fast scan‑path visualiser.

    • **Vectorised**: no per‑row Python loops
    • **Single pass** phase grouping
    • Uses `BrokenBarHCollection` for fixation spans
    • Each requested PNG is written once

    Parameters
    ----------
    fixations : polars.DataFrame
        Polars DataFrame with at least `tStart`, `duration`, `xAvg`, `yAvg`, `phase`.
    screen_height, screen_width : int
        Stimulus resolution in pixels.
    folder_path : str or pathlib.Path, optional
        Directory where 1 PNG per phase will be stored.  If *None*, nothing is saved.
    tmin, tmax : int, optional
        Time window in **ms**.  If both `None`, the whole trial is plotted.
    saccades : polars.DataFrame, optional
        Polars DataFrame with `tStart`, `phase`, …  (optional).
    samples : polars.DataFrame, optional
        Polars DataFrame with gaze traces (`tSample`, `LX`, `LY`, `RX`, `RY` or
        `X`, `Y`) (optional).
    phase_data : dict, optional
        Per‑phase extras::

            {
                "search": {
                    "img_paths": [...],
                    "img_plot_coords": [(x1,y1,x2,y2), ...],
                    "bbox": (x1,y1,x2,y2),
                },
                ...
            }

    display : bool, default True
        If *False* the figure canvas is never shown (faster for batch jobs).

    Notes
    -----
    One figure is produced per named trial phase. Fixations that fall
    outside every phase are skipped. If no fixation carries a phase name at
    all, which happens when the recording was never segmented, they are all
    drawn as a single unnamed phase and a :class:`UserWarning` is issued.
    """
    if fixations.is_empty():
        return
    required = {"trial_number", "phase", "tStart", "duration", "xAvg", "yAvg"}
    missing = sorted(required - set(fixations.columns))
    if missing:
        raise ValueError(
            "Fixations are missing required columns: " + ", ".join(missing)
        )
    if (tmin is None) != (tmax is None):
        raise ValueError("tmin and tmax must be provided together")
    if folder_path is not None:
        Path(folder_path).mkdir(parents=True, exist_ok=True)

    # ------------- small helpers ------------------------------------------------
    def _make_axes(plot_samples: bool):
        if plot_samples:
            fig, (ax_main, ax_gaze) = plt.subplots(
                2, 1, height_ratios=(4, 1), figsize=(10, 6), sharex=False
            )
        else:
            fig, ax_main = plt.subplots(figsize=(10, 6))
            ax_gaze = None
        ax_main.set_xlim(0, screen_width)
        ax_main.set_ylim(screen_height, 0)
        return fig, ax_main, ax_gaze

    def _maybe_cache_img(path):
        """Load image from disk with a small LRU cache."""

        # Cache hit: move to the end (most recently used)
        if path in _img_cache:
            img = _img_cache.pop(path)
            _img_cache[path] = img
            return img

        # Cache miss: load image
        img = mpimg.imread(path)

        # Optional: reduce memory if image is float64 in [0, 1]
        if isinstance(img, np.ndarray) and img.dtype == np.float64:
            img = (img * 255).clip(0, 255).astype(np.uint8)

        # Insert into cache
        _img_cache[path] = img

        # If cache too big, drop least recently used item
        if len(_img_cache) > MAX_CACHED_IMAGES:
            _img_cache.popitem(last=False)  # pops the oldest inserted item

        return img

    # ---------------------------------------------------------------------------
    plot_saccades = saccades is not None
    plot_samples = samples is not None
    _img_cache = OrderedDict()

    trial_idx = fixations["trial_number"][0]
    if (
        isinstance(trial_idx, (float, np.floating))
        and float(trial_idx).is_integer()
    ):
        trial_idx = int(trial_idx)

    # ---- time filter ----------------------------------------------------------
    if tmin is not None and tmax is not None:
        fixations = fixations.filter(pl.col("tStart").is_between(tmin, tmax))
        if plot_saccades:
            saccades = saccades.filter(pl.col("tStart").is_between(tmin, tmax))
        if plot_samples:
            samples = samples.filter(pl.col("tSample").is_between(tmin, tmax))

    # Rows outside any named phase carry an empty ``phase``. Dropping them
    # is right for a segmented recording, where they fall between trials.
    # But a recording with no named phase at all -- any format whose source
    # carries no synchronisation messages, such as a plain Tobii or
    # GazePoint export -- would then lose every row and plot nothing, with
    # no clue as to why. Keep those rows as a single unnamed phase instead,
    # and say what happened.
    if fixations.get_column("phase").fill_null("").eq("").all():
        warnings.warn(
            "No named trial phase was found, so every fixation is plotted "
            "as a single unnamed phase. Pass start_msgs and end_msgs to "
            "compute_derivatives_for_dataset to segment the recording into "
            "named phases.",
            UserWarning,
            stacklevel=2,
        )
    else:
        fixations = fixations.filter(pl.col("phase") != "")
        if plot_saccades:
            saccades = saccades.filter(pl.col("phase") != "")
        if plot_samples:
            samples = samples.filter(pl.col("phase") != "")

    # Rows the preprocessing step flagged as bad hold gaze that fell off
    # the screen or was never tracked. Drawing them stretches the scanpath
    # towards coordinates the participant never looked at, and joins them
    # with lines that cross the whole stimulus.
    def _drop_bad(frame: pl.DataFrame) -> pl.DataFrame:
        if frame is None or "bad" not in frame.columns:
            return frame
        return frame.filter(
            ~pl.col("bad").cast(pl.Boolean, strict=False).fill_null(False)
        )

    fixations = _drop_bad(fixations)
    if plot_saccades:
        saccades = _drop_bad(saccades)
    if plot_samples:
        samples = _drop_bad(samples)
    if fixations.is_empty():
        warnings.warn(
            "Every fixation was flagged as bad, so there is nothing to "
            "plot. Check the screen size passed to "
            "compute_derivatives_for_dataset.",
            UserWarning,
            stacklevel=2,
        )
        return

    # ---- split once by phase --------------------------------------------------
    fix_by_phase = fixations.partition_by("phase", as_dict=True)
    sac_by_phase = (
        saccades.partition_by("phase", as_dict=True) if plot_saccades else {}
    )
    samp_by_phase = (
        samples.partition_by("phase", as_dict=True) if plot_samples else {}
    )

    # colour map shared across phases
    cmap = plt.cm.rainbow

    # ---- build & draw ---------------------------------------------------------
    interactive_before = plt.isinteractive()
    if not display:
        plt.ioff()

    for phase, phase_fix in fix_by_phase.items():
        if phase_fix.is_empty():
            continue
        phase_name = phase[0] if isinstance(phase, tuple) else phase

        # ---------- vectors (zero‑copy) -----------------
        fx, fy, fdur = phase_fix.select(["xAvg", "yAvg", "duration"]).to_numpy().T
        n_fix = fx.size
        fix_idx = np.arange(1, n_fix + 1)

        # One colour band per fixation is only possible while the colormap
        # has enough of them. Long recordings hold thousands of fixations,
        # so fall back to a continuous scale instead of raising.
        if n_fix < cmap.N:
            norm = mplcolors.BoundaryNorm(np.arange(1, n_fix + 2), cmap.N)
        else:
            norm = mplcolors.Normalize(vmin=1, vmax=max(n_fix, 2))

        # saccades
        sac_t = (
            sac_by_phase[phase]["tStart"].to_numpy()
            if plot_saccades and phase in sac_by_phase
            else np.empty(0)
        )

        # samples
        if plot_samples and phase in samp_by_phase and samp_by_phase[phase].height:
            samp_phase = samp_by_phase[phase]
            t0 = samp_phase["tSample"][0]
            ts = samp_phase["tSample"].to_numpy() - t0
            get = samp_phase.get_column
            lx = get("LX").to_numpy() if "LX" in samp_phase.columns else None
            ly = get("LY").to_numpy() if "LY" in samp_phase.columns else None
            rx = get("RX").to_numpy() if "RX" in samp_phase.columns else None
            ry = get("RY").to_numpy() if "RY" in samp_phase.columns else None
            gx = get("X").to_numpy() if "X" in samp_phase.columns else None
            gy = get("Y").to_numpy() if "Y" in samp_phase.columns else None
        else:
            t0 = None

        # ---------- figure -----------------------------
        fig, ax_main, ax_gaze = _make_axes(plot_samples and t0 is not None)
        # scatter fixations
        sc = ax_main.scatter(
            fx,
            fy,
            c=fix_idx,
            s=fdur,
            cmap=cmap,
            norm=norm,
            alpha=0.5,
            zorder=2,
        )
        fig.colorbar(
            sc,
            ax=ax_main,
            ticks=[1, n_fix // 2 if n_fix > 2 else n_fix, n_fix],
            fraction=0.046,
            pad=0.04,
        ).set_label("# of fixation")

        # ---------- stimulus imagery / bbox ------------
        if phase_data and phase_name in phase_data:
            pdict = phase_data[phase_name]
            coords = pdict.get("img_plot_coords") or []
            bbox = pdict.get("bbox", None)
            for img_path, box in zip(pdict.get("img_paths", []), coords):
                ax_main.imshow(
                    _maybe_cache_img(img_path),
                    extent=[box[0], box[2], box[3], box[1]],
                    zorder=0,
                )
            if bbox is not None:
                x1, y1, x2, y2 = bbox
                ax_main.plot(
                    [x1, x2, x2, x1, x1],
                    [y1, y1, y2, y2, y1],
                    color="red",
                    linewidth=1.5,
                    zorder=3,
                )

        # ---------- gaze traces ------------------------
        if ax_gaze is not None:
            if lx is not None:
                ax_main.plot(lx, ly, "--", color="C0", zorder=1)
                ax_gaze.plot(ts, lx, label="Left X")
                ax_gaze.plot(ts, ly, label="Left Y")
            if rx is not None:
                ax_main.plot(rx, ry, "--", color="k", zorder=1)
                ax_gaze.plot(ts, rx, label="Right X")
                ax_gaze.plot(ts, ry, label="Right Y")
            if gx is not None:
                ax_main.plot(gx, gy, "--", color="k", zorder=1, alpha=0.6)
                ax_gaze.plot(ts, gx, label="X")
                ax_gaze.plot(ts, gy, label="Y")

            # fixation spans
            bars = np.c_[
                phase_fix["tStart"].to_numpy() - t0,
                phase_fix["duration"].to_numpy(),
            ]
            height = ax_gaze.get_ylim()[1] - ax_gaze.get_ylim()[0]
            colors = cmap(norm(fix_idx))

            # Draw all bars in one call; no BrokenBarHCollection import needed
            ax_gaze.broken_barh(bars, (0, height), facecolors=colors, alpha=0.4)
            # saccades
            if sac_t.size:
                ymin, ymax = ax_gaze.get_ylim()
                ax_gaze.vlines(
                    sac_t - t0,
                    ymin,
                    ymax,
                    colors="red",
                    linestyles="--",
                    linewidth=0.8,
                )

            # tidy gaze axis
            h, l = ax_gaze.get_legend_handles_labels()
            by_label = {lab: hdl for hdl, lab in zip(h, l)}
            ax_gaze.legend(
                by_label.values(),
                by_label.keys(),
                loc="center left",
                bbox_to_anchor=(1, 0.5),
            )
            ax_gaze.set_ylabel("Gaze")
            ax_gaze.set_xlabel("Time [ms]")

        fig.tight_layout()

        # ---------- save / show ------------------------
        if folder_path:
            scan_name = f"scanpath_{trial_idx}"
            if tmin is not None and tmax is not None:
                scan_name += f"_{tmin}_{tmax}"
            out = Path(folder_path) / f"{scan_name}_{phase_name or 'unphased'}.png"
            fig.savefig(out, dpi=150)

        if display:
            plt.show()
        plt.close(fig)

    if not display and interactive_before:
        plt.ion()

samples

Plotting and animating sample-level gaze directly, without requiring detected events.

Visualizations for sample-level gaze data.

SampleVisualization

Plot and animate sample-level gaze coordinates.

Parameters:

Name Type Description Default
samples_df DataFrame

Polars dataframe containing gaze-coordinate columns. X and Y are required by both plotting methods; tSample is additionally required by :meth:plot.

required
screen_width float

Screen width in pixels.

1366
screen_height float

Screen height in pixels.

768
Source code in pyxations/visualization/samples.py
class SampleVisualization:
    """Plot and animate sample-level gaze coordinates.

    Parameters
    ----------
    samples_df
        Polars dataframe containing gaze-coordinate columns. ``X`` and ``Y``
        are required by both plotting methods; ``tSample`` is additionally
        required by :meth:`plot`.
    screen_width
        Screen width in pixels.
    screen_height
        Screen height in pixels.
    """

    def __init__(
        self,
        samples_df: pl.DataFrame,
        screen_width: float = 1366,
        screen_height: float = 768,
    ) -> None:
        if not isinstance(samples_df, pl.DataFrame):
            raise TypeError(
                "SampleVisualization requires a polars.DataFrame; "
                f"received {type(samples_df).__name__}."
            )
        if not np.isfinite(screen_width) or screen_width <= 0:
            raise ValueError("screen_width must be a positive finite number.")
        if not np.isfinite(screen_height) or screen_height <= 0:
            raise ValueError("screen_height must be a positive finite number.")

        self.samples = samples_df
        self.screen_width = float(screen_width)
        self.screen_height = float(screen_height)

    def _numeric_column(self, name: str) -> np.ndarray:
        """Return one dataframe column as a one-dimensional float array."""
        if name not in self.samples.columns:
            raise ValueError(f"Sample dataframe is missing required column {name!r}.")

        try:
            values = self.samples.get_column(name).cast(pl.Float64, strict=True)
        except Exception as exc:
            raise TypeError(
                f"Sample column {name!r} must contain numeric values."
            ) from exc

        array = np.asarray(values.to_numpy(), dtype=float)
        if array.ndim != 1:
            raise ValueError(f"Sample column {name!r} must be one-dimensional.")
        return array

    def _gaze_arrays(self, *, in_percent: bool) -> tuple[np.ndarray, np.ndarray]:
        """Return gaze coordinates in pixels as NumPy arrays."""
        x = self._numeric_column("X")
        y = self._numeric_column("Y")

        if in_percent:
            x = x * self.screen_width
            y = y * self.screen_height

        return x, y

    @staticmethod
    def _require_samples(x: np.ndarray, y: np.ndarray) -> None:
        if x.size == 0 or y.size == 0:
            raise ValueError("At least one gaze sample is required.")

    def plot(
        self,
        display: bool = True,
        scanpath_file_name: str | Path = "scanpath",
        in_percent: bool = False,
    ) -> None:
        """Save a scanpath and gaze-over-time plot as a PNG image.

        Parameters
        ----------
        display : bool, default True
            Whether to show the figure interactively after saving it.
        scanpath_file_name : str or pathlib.Path, default "scanpath"
            Output path without the ``.png`` suffix.
        in_percent : bool, default False
            Whether gaze coordinates are fractions of the screen dimensions.

        Raises
        ------
        ValueError
            If the sample table contains no usable gaze samples.
        """
        x, y = self._gaze_arrays(in_percent=in_percent)
        self._require_samples(x, y)
        timestamps = self._numeric_column("tSample")

        fig, axs = plt.subplots(
            nrows=2,
            ncols=1,
            height_ratios=(4, 1),
            figsize=(10, 6),
        )
        ax_main = axs[0]
        ax_gaze = axs[1]

        ax_main.set_xlim(0, self.screen_width)
        ax_main.set_ylim(0, self.screen_height)
        ax_main.plot(x, y, "--", color="C0", zorder=1)

        ax_gaze.plot(timestamps, x, label="X")
        ax_gaze.plot(timestamps, y, label="Y")
        ax_gaze.legend(loc="center left", bbox_to_anchor=(1, 0.5))
        ax_gaze.set_ylabel("Gaze")
        ax_gaze.set_xlabel("Time [ms]")

        plt.tight_layout()
        file_path = Path(f"{scanpath_file_name}.png")
        fig.savefig(file_path)
        if display:
            plt.show()
        plt.close(fig)

    def animate(
        self,
        display: bool = True,
        in_percent: bool = False,
        out_file: str | Path = "output.gif",
    ) -> None:
        """Save an animated gaze trace as a GIF image.

        Parameters
        ----------
        display : bool, default True
            Whether to show the animation interactively before saving it.
        in_percent : bool, default False
            Whether gaze coordinates are fractions of the screen dimensions.
        out_file : str or pathlib.Path, default "output.gif"
            Destination GIF path.

        Raises
        ------
        ValueError
            If the sample table contains no usable gaze samples.
        """
        x, y = self._gaze_arrays(in_percent=in_percent)
        self._require_samples(x, y)

        fig, ax = plt.subplots()
        ax.set_xlim(0, self.screen_width)
        ax.set_ylim(0, self.screen_height)

        scat = ax.scatter(x[0], y[0], c="b", s=5, label="a")
        line = ax.plot(x[0], y[0], label="b")[0]
        ax.legend()

        def update(frame: int):
            # ``frame`` is a sample count, not a zero-based sample index.
            x_frame = x[:frame]
            y_frame = y[:frame]
            scat.set_offsets(np.column_stack((x_frame, y_frame)))
            line.set_xdata(x_frame)
            line.set_ydata(y_frame)
            return scat, line

        gaze_animation = animation.FuncAnimation(
            fig=fig,
            func=update,
            frames=range(1, len(x) + 1),
            interval=1,
        )
        if display:
            plt.show()

        gaze_animation.save(filename=Path(out_file), writer="pillow")
        plt.close(fig)

animate(display=True, in_percent=False, out_file='output.gif')

Save an animated gaze trace as a GIF image.

Parameters:

Name Type Description Default
display bool

Whether to show the animation interactively before saving it.

True
in_percent bool

Whether gaze coordinates are fractions of the screen dimensions.

False
out_file str or Path

Destination GIF path.

"output.gif"

Raises:

Type Description
ValueError

If the sample table contains no usable gaze samples.

Source code in pyxations/visualization/samples.py
def animate(
    self,
    display: bool = True,
    in_percent: bool = False,
    out_file: str | Path = "output.gif",
) -> None:
    """Save an animated gaze trace as a GIF image.

    Parameters
    ----------
    display : bool, default True
        Whether to show the animation interactively before saving it.
    in_percent : bool, default False
        Whether gaze coordinates are fractions of the screen dimensions.
    out_file : str or pathlib.Path, default "output.gif"
        Destination GIF path.

    Raises
    ------
    ValueError
        If the sample table contains no usable gaze samples.
    """
    x, y = self._gaze_arrays(in_percent=in_percent)
    self._require_samples(x, y)

    fig, ax = plt.subplots()
    ax.set_xlim(0, self.screen_width)
    ax.set_ylim(0, self.screen_height)

    scat = ax.scatter(x[0], y[0], c="b", s=5, label="a")
    line = ax.plot(x[0], y[0], label="b")[0]
    ax.legend()

    def update(frame: int):
        # ``frame`` is a sample count, not a zero-based sample index.
        x_frame = x[:frame]
        y_frame = y[:frame]
        scat.set_offsets(np.column_stack((x_frame, y_frame)))
        line.set_xdata(x_frame)
        line.set_ydata(y_frame)
        return scat, line

    gaze_animation = animation.FuncAnimation(
        fig=fig,
        func=update,
        frames=range(1, len(x) + 1),
        interval=1,
    )
    if display:
        plt.show()

    gaze_animation.save(filename=Path(out_file), writer="pillow")
    plt.close(fig)

plot(display=True, scanpath_file_name='scanpath', in_percent=False)

Save a scanpath and gaze-over-time plot as a PNG image.

Parameters:

Name Type Description Default
display bool

Whether to show the figure interactively after saving it.

True
scanpath_file_name str or Path

Output path without the .png suffix.

"scanpath"
in_percent bool

Whether gaze coordinates are fractions of the screen dimensions.

False

Raises:

Type Description
ValueError

If the sample table contains no usable gaze samples.

Source code in pyxations/visualization/samples.py
def plot(
    self,
    display: bool = True,
    scanpath_file_name: str | Path = "scanpath",
    in_percent: bool = False,
) -> None:
    """Save a scanpath and gaze-over-time plot as a PNG image.

    Parameters
    ----------
    display : bool, default True
        Whether to show the figure interactively after saving it.
    scanpath_file_name : str or pathlib.Path, default "scanpath"
        Output path without the ``.png`` suffix.
    in_percent : bool, default False
        Whether gaze coordinates are fractions of the screen dimensions.

    Raises
    ------
    ValueError
        If the sample table contains no usable gaze samples.
    """
    x, y = self._gaze_arrays(in_percent=in_percent)
    self._require_samples(x, y)
    timestamps = self._numeric_column("tSample")

    fig, axs = plt.subplots(
        nrows=2,
        ncols=1,
        height_ratios=(4, 1),
        figsize=(10, 6),
    )
    ax_main = axs[0]
    ax_gaze = axs[1]

    ax_main.set_xlim(0, self.screen_width)
    ax_main.set_ylim(0, self.screen_height)
    ax_main.plot(x, y, "--", color="C0", zorder=1)

    ax_gaze.plot(timestamps, x, label="X")
    ax_gaze.plot(timestamps, y, label="Y")
    ax_gaze.legend(loc="center left", bbox_to_anchor=(1, 0.5))
    ax_gaze.set_ylabel("Gaze")
    ax_gaze.set_xlabel("Time [ms]")

    plt.tight_layout()
    file_path = Path(f"{scanpath_file_name}.png")
    fig.savefig(file_path)
    if display:
        plt.show()
    plt.close(fig)