Data

Package

---

data

psyphy.data

submodule for handling psychophysical experiment data.

Includes: - dataset: ResponseData, TrialBatch, loaders - transforms: color/model space conversions - io: save/load datasets - published: loaders for published third-party datasets (downloaded on request)

Classes:

Name	Description
ResponseData	Python-friendly incremental trial log.
TrialBatch	Container for a proposed batch of trials.
TrialData	Batched trial data for compute.

ResponseData

ResponseData()

Python-friendly incremental trial log.

This container is convenient for adaptive trial placement and I/O (e.g., CSV), but it is not a compute-efficient representation for JAX.

Use :class:TrialData for model fitting and likelihood evaluation.

Methods:

Name	Description
add_batch	Append responses for a batch of trials.
add_trial	append a single trial.
copy	Create a deep copy of this dataset.
from_arrays	Construct ResponseData from arrays.
from_trial_data	Build a ResponseData log from a :class:TrialData batch.
merge	Merge another dataset into this one (in-place).
tail	Return last n trials as a new ResponseData.
to_numpy	Return stimuli and responses as NumPy arrays.
to_trial_data	Convert this log into the canonical JAX batch (:class:TrialData).

Attributes:

Name	Type	Description
contexts	list[ndarray]
responses	list[ndarray]
stim_shape	tuple | None
stimuli	list[ndarray]
trials	list[tuple[Any, ...]]	Return list of (stim1, stim2, ... , response) tuples.

Source code in src/psyphy/data/dataset.py

def __init__(self) -> None:
    self.stimuli: list[np.ndarray] = []
    self.responses: list[np.ndarray] = []
    self.stim_shape: tuple | None = None  # set on first add_trial call
    self.contexts: list[np.ndarray] = []

contexts

contexts: list[ndarray] = []

responses

responses: list[ndarray] = []

stim_shape

stim_shape: tuple | None = None

stimuli

stimuli: list[ndarray] = []

trials

trials: list[tuple[Any, ...]]

Return list of (stim1, stim2, ... , response) tuples. Does NOT include context information.

Returns:

Type	Description
list[tuple]	Each element is tuple representing all stimuli and the associated response for a given trial.

add_batch

add_batch(responses: list[Any], trial_batch: TrialBatch, contexts: list[Any] | None = None) -> None

Append responses for a batch of trials.

Parameters:

Name	Type	Description	Default
responses	List[Any]	Responses corresponding to each stimulus group in the trial batch.	required
trial_batch	TrialBatch	The batch of proposed trials.	required

Source code in src/psyphy/data/dataset.py

def add_batch(
    self,
    responses: list[Any],
    trial_batch: TrialBatch,
    contexts: list[Any] | None = None,
) -> None:
    """
    Append responses for a batch of trials.

    Parameters
    ----------
    responses : List[Any]
        Responses corresponding to each stimulus group in the trial batch.
    trial_batch : TrialBatch
        The batch of proposed trials.
    """
    if contexts is None:
        for input, resp in zip(trial_batch.stimuli, responses):
            self.add_trial(input, resp)
    else:
        for input, resp, context in zip(trial_batch.stimuli, responses, contexts):
            self.add_trial(input, resp, context)

add_trial

add_trial(input: tuple[Any, ...], resp: Any, context: Any = None) -> None

append a single trial.

Parameters:

Name	Type	Description	Default
input	tuple(Any, ...)	Group of presented stimuli each represented in any format (numpy array, list, etc.) Input must contain appropriate number of stimuli of appropriate dimension.	required
resp	Any	Subject response	required

Source code in src/psyphy/data/dataset.py

def add_trial(self, input: tuple[Any, ...], resp: Any, context: Any = None) -> None:
    """
    append a single trial.

    Parameters
    ----------
    input : tuple(Any, ...)
        Group of presented stimuli each represented in any format (numpy array,
        list, etc.)
        Input must contain appropriate number of stimuli of appropriate dimension.
    resp : Any
        Subject response
    """
    input_arr = np.atleast_2d(np.asarray(input))  # (K, d) — 1D input treated as K=1
    resp_arr = np.atleast_1d(np.asarray(resp))  # (R,)   — scalar treated as R=1
    if self.stimuli:
        if self.stim_shape != input_arr.shape:
            raise ValueError(
                f"stimuli must have consistent shape (K, d). Expected {self.stim_shape}, but received {input_arr.shape}"
            )
    else:
        self.stim_shape = input_arr.shape

    if context is None:
        if self.contexts:
            raise ValueError(
                "Context cannot be omitted if it was included in previous trials."
                "This ResponseData instance expected context but received none."
            )
    else:
        if self.contexts or self.stimuli == []:
            self.contexts.append(np.asarray(context))
        else:
            raise ValueError(
                "Context cannot be accepted if it was excluded from prior trials."
                f"This ResponseData instance expected no context, but received {context}"
            )
    self.stimuli.append(input_arr)
    self.responses.append(resp_arr)

copy

copy() -> ResponseData

Create a deep copy of this dataset.

Returns:

Type	Description
ResponseData	New dataset with copied data

Source code in src/psyphy/data/dataset.py

def copy(self) -> ResponseData:
    """
    Create a deep copy of this dataset.

    Returns
    -------
    ResponseData
        New dataset with copied data
    """
    new_data = ResponseData()
    new_data.stimuli = list(self.stimuli)
    new_data.responses = list(self.responses)
    if self.contexts is not None:
        new_data.contexts = list(self.contexts)
    return new_data

from_arrays

from_arrays(X: ndarray | ndarray, y: ndarray | ndarray, c: ndarray | ndarray | None = None) -> ResponseData

Construct ResponseData from arrays.

Parameters:

Name	Type	Description	Default
X	(array, shape(n_trials, n_stimuli, input_dim) or (n_trials, input_dim))	Stimuli. If 3D, second axis is input stumili. For OddityTask, this is (ref, comparison)	required
y	(array, shape(n_trials, response_dim))	Responses	required
c	optional array, shape (n_trials, context_dim)	Context	None

Returns:

Type	Description
ResponseData	Data container

OddityTask Example

From paired stimuli

X = jnp.array([[[0, 0], [1, 0]], [[1, 1], [2, 1]]])

X is formed from refs = [[0, 0], [1, 1]], comparisons = [[1, 0], [2, 1]]

y = jnp.array([1, 0]) data = ResponseData.from_arrays(X, y)

Source code in src/psyphy/data/dataset.py

@classmethod
def from_arrays(
    cls,
    X: jnp.ndarray | np.ndarray,
    y: jnp.ndarray | np.ndarray,
    c: jnp.ndarray | np.ndarray | None = None,
) -> ResponseData:
    """
    Construct ResponseData from arrays.

    Parameters
    ----------
    X : array, shape (n_trials, n_stimuli, input_dim) or (n_trials, input_dim)
        Stimuli. If 3D, second axis is input stumili. For OddityTask, this is
        (ref, comparison)
    y : array, shape (n_trials, response_dim)
        Responses
    c : optional array, shape (n_trials, context_dim)
        Context

    Returns
    -------
    ResponseData
        Data container

    OddityTask Example
    --------
    >>> # From paired stimuli
    >>> X = jnp.array([[[0, 0], [1, 0]], [[1, 1], [2, 1]]])
    >>> # X is formed from refs = [[0, 0], [1, 1]], comparisons = [[1, 0], [2, 1]]
    >>> y = jnp.array([1, 0])
    >>> data = ResponseData.from_arrays(X, y)
    """
    data = cls()

    X = np.asarray(X)
    y = np.asarray(y)

    if X.ndim == 2:
        # reshape to ensure appropriate conversion to stimuli groups
        dims = X.shape
        new_dims = (dims[0], 1, dims[1])
        X = np.reshape(X, new_dims)
    elif X.ndim != 3:
        raise ValueError(
            "X must be shape (n_trials, n_stimuli, input_dim) or \
            (n_trials, input_dim)."
        )
    if y.shape[0] != X.shape[0]:
        raise ValueError("X and y must contain the same n_trials.")
    if c is not None and c.shape[0] != X.shape[0]:
        raise ValueError("c must contain same n_trials as X.")

    # X is (n_trials, K, d) — split into per-trial tuples of K stimulus rows
    stimuli = []
    for plane in X:
        stimuli.append(tuple(plane))

    if c is not None:
        for stim, response, context in zip(stimuli, y, c):
            data.add_trial(stim, response, context)
    else:
        for stim, response in zip(stimuli, y):
            data.add_trial(stim, response)

    return data

from_trial_data

from_trial_data(data: TrialData) -> ResponseData

Build a ResponseData log from a :class:TrialData batch.

Source code in src/psyphy/data/dataset.py

@classmethod
def from_trial_data(cls, data: TrialData) -> ResponseData:
    """Build a ResponseData log from a :class:`TrialData` batch."""
    stimuli = np.asarray(data.stimuli)
    ys = np.asarray(data.responses)
    out = cls()
    if data.context is not None:
        cs = np.asarray(data.context)
        for s, y, c in zip(stimuli, ys, cs):
            out.add_trial(s, y, c)
    else:
        for s, y in zip(stimuli, ys):
            out.add_trial(s, y)
    return out

merge

merge(other: ResponseData) -> None

Merge another dataset into this one (in-place).

Parameters:

Name	Type	Description	Default
other	ResponseData	Dataset to merge	required

Source code in src/psyphy/data/dataset.py

def merge(self, other: ResponseData) -> None:
    """
    Merge another dataset into this one (in-place).

    Parameters
    ----------
    other : ResponseData
        Dataset to merge
    """
    no_empty = self.stimuli and other.stimuli

    if no_empty and self.stimuli[0].shape != other.stimuli[0].shape:
        raise ValueError(
            "Cannot merge ResponseData instances with inconsistent input shapes."
            f"Received input shapes of {self.stimuli[0].shape} and {other.stimuli[0].shape}"
        )
    if no_empty and self.responses[0].shape != other.responses[0].shape:
        raise ValueError(
            "Cannot merge ResponseData instances with inconsistent response shapes."
            f"Received response shapes of {self.responses[0].shape} and {other.responses[0].shape}"
        )

    self.stimuli.extend(other.stimuli)
    self.responses.extend(other.responses)
    both_contexts = self.contexts and other.contexts

    if self.contexts == [] and other.contexts == []:
        pass
    elif both_contexts and self.contexts[0].shape == other.contexts[0].shape:
        self.contexts.extend(other.contexts)
    else:
        raise ValueError(
            "Cannot merge ResponseData instances with inconsistent context."
        )

tail

tail(n: int) -> ResponseData

Return last n trials as a new ResponseData.

Parameters:

Name	Type	Description	Default
n	int	Number of trials to keep	required

Returns:

Type	Description
ResponseData	New dataset with last n trials

Source code in src/psyphy/data/dataset.py

def tail(self, n: int) -> ResponseData:
    """
    Return last n trials as a new ResponseData.

    Parameters
    ----------
    n : int
        Number of trials to keep

    Returns
    -------
    ResponseData
        New dataset with last n trials
    """
    new_data = ResponseData()
    new_data.stimuli = self.stimuli[-n:]
    new_data.responses = self.responses[-n:]
    if self.contexts is not None:
        new_data.contexts = self.contexts[-n:]
    return new_data

to_numpy

to_numpy() -> tuple[ndarray, ndarray]

Return stimuli and responses as NumPy arrays. Will NOT include contexts by default. Output always fixed length of 2.

Source code in src/psyphy/data/dataset.py

def to_numpy(self) -> tuple[np.ndarray, np.ndarray]:
    """Return stimuli and responses as NumPy arrays.
    Will NOT include contexts by default. Output always fixed length of 2.
    """
    return (
        np.asarray(self.stimuli),  # shape = (N, K, d)
        np.asarray(self.responses),
    )

to_trial_data

to_trial_data() -> TrialData

Convert this log into the canonical JAX batch (:class:TrialData).

Source code in src/psyphy/data/dataset.py

def to_trial_data(self) -> TrialData:
    """Convert this log into the canonical JAX batch (:class:`TrialData`)."""
    stimuli, responses = self.to_numpy()
    if self.contexts:
        context = np.asarray(self.contexts)
        return TrialData(
            stimuli=jnp.asarray(stimuli),
            responses=jnp.asarray(responses),
            context=jnp.asarray(context),
        )
    else:
        return TrialData(
            stimuli=jnp.asarray(stimuli), responses=jnp.asarray(responses)
        )

TrialBatch

TrialBatch(stimuli: list[tuple[Any, ...]])

Container for a proposed batch of trials. Does NOT include context or responses.

Attributes:

Name	Type	Description
stimuli	List[Tuple[Any, ...]]	Each trial is a tuple of all presented stimuli (stim1, stim2, ...). For OddityTask this is (reference, comparison)

Methods:

Name	Description
from_stimuli	Construct a TrialBatch from a list of stimuli (stim1, stim2, ...) groups.

Source code in src/psyphy/data/dataset.py

def __init__(self, stimuli: list[tuple[Any, ...]]) -> None:
    self.stimuli = list(stimuli)

stimuli

stimuli = list(stimuli)

from_stimuli

from_stimuli(groups: list[tuple[Any, ...]]) -> TrialBatch

Construct a TrialBatch from a list of stimuli (stim1, stim2, ...) groups.

Source code in src/psyphy/data/dataset.py

@classmethod
def from_stimuli(cls, groups: list[tuple[Any, ...]]) -> TrialBatch:
    """
    Construct a TrialBatch from a list of stimuli (stim1, stim2, ...) groups.
    """
    return cls(groups)

TrialData

TrialData(stimuli: ndarray, responses: ndarray, context: ndarray | None = None, stimulus_names: tuple[str, ...] = ())

Batched trial data for compute.

This is the canonical, compute-efficient representation of observed trials.

Shapes

stimuli : (N, K, d) responses : (N, R) context : optional (N, C)

Dimension key

N : number of trials (batch dimension) K : number of stimuli per trial (e.g. K=2 for a two-alternative task; K=2 for the oddity task, where the reference is presented twice but only the unique mean is stored — the duplication is encoded in the task likelihood, not here) d : dimensionality of each stimulus coordinate R : number of response channels (R=1 for binary; R=2 for e.g. (choice, RT)) C : number of context channels (observer-state covariates that condition the likelihood but are not part of the stimulus space, e.g. fatigue level)

Notes

• You can also think of this as a more generic ML-style dataset X with shape (N, K, d) plus y with shape (N, R).
• This is intended to be JAX-friendly (PyTree of arrays) so likelihood and inference code can be JIT-compiled without touching Python containers.
• Context is optional. No current inbuilt uses.

Methods:

Name	Description
stimulus	Return stimuli[:, k, :] for the slot named name.

Attributes:

Name	Type	Description
context	ndarray | None
num_trials	int	Number of trials (N).
responses	ndarray
stimuli	ndarray
stimulus_names	tuple[str, ...]

context

context: ndarray | None = None

num_trials

num_trials: int

Number of trials (N).

responses

responses: ndarray

stimuli

stimuli: ndarray

stimulus_names

stimulus_names: tuple[str, ...] = ()

stimulus

stimulus(name: str) -> ndarray

Return stimuli[:, k, :] for the slot named name.

Parameters:

Name	Type	Description	Default
name	str	Must match one of the entries in stimulus_names.	required

Returns:

Type	Description
(ndarray, shape(N, d))	Stimulus coordinates for all trials at the named slot.

Source code in src/psyphy/data/dataset.py

def stimulus(self, name: str) -> jnp.ndarray:
    """Return stimuli[:, k, :] for the slot named `name`.

    Parameters
    ----------
    name : str
        Must match one of the entries in ``stimulus_names``.

    Returns
    -------
    jnp.ndarray, shape (N, d)
        Stimulus coordinates for all trials at the named slot.
    """
    if not self.stimulus_names:
        raise ValueError(
            "stimulus_names is empty — set it at construction time to use "
            "named access, e.g. stimulus_names=('ref', 'comp')."
        )
    if name not in self.stimulus_names:
        raise ValueError(
            f"unknown stimulus name '{name}'. "
            f"Available names: {self.stimulus_names}."
        )
    idx = self.stimulus_names.index(name)  # resolved in Python, not JAX-traced
    return self.stimuli[:, idx, :]

---

Data Containers for Response Data and Proposed Next Trials

---

dataset

dataset.py

Core data containers for psyphy.

defines: - ResponseData: container for psychophysical trial data - TrialBatch: container for a proposed batch of trials

Notes

• Data is stored in standard NumPy (mutable!) arrays or Python lists.
• Use numpy for I/O and analysis.
• Convert to jax.numpy (jnp) (immutable!) arrays only when passing into WPPM or inference engines that require JAX/Optax.

Classes:

Name	Description
ResponseData	Python-friendly incremental trial log.
TrialBatch	Container for a proposed batch of trials.
TrialData	Batched trial data for compute.

ResponseData

ResponseData()

Python-friendly incremental trial log.

This container is convenient for adaptive trial placement and I/O (e.g., CSV), but it is not a compute-efficient representation for JAX.

Use :class:TrialData for model fitting and likelihood evaluation.

Methods:

Name	Description
add_batch	Append responses for a batch of trials.
add_trial	append a single trial.
copy	Create a deep copy of this dataset.
from_arrays	Construct ResponseData from arrays.
from_trial_data	Build a ResponseData log from a :class:TrialData batch.
merge	Merge another dataset into this one (in-place).
tail	Return last n trials as a new ResponseData.
to_numpy	Return stimuli and responses as NumPy arrays.
to_trial_data	Convert this log into the canonical JAX batch (:class:TrialData).

Attributes:

Name	Type	Description
contexts	list[ndarray]
responses	list[ndarray]
stim_shape	tuple | None
stimuli	list[ndarray]
trials	list[tuple[Any, ...]]	Return list of (stim1, stim2, ... , response) tuples.

Source code in src/psyphy/data/dataset.py

def __init__(self) -> None:
    self.stimuli: list[np.ndarray] = []
    self.responses: list[np.ndarray] = []
    self.stim_shape: tuple | None = None  # set on first add_trial call
    self.contexts: list[np.ndarray] = []

contexts

contexts: list[ndarray] = []

responses

responses: list[ndarray] = []

stim_shape

stim_shape: tuple | None = None

stimuli

stimuli: list[ndarray] = []

trials

trials: list[tuple[Any, ...]]

Return list of (stim1, stim2, ... , response) tuples. Does NOT include context information.

Returns:

Type	Description
list[tuple]	Each element is tuple representing all stimuli and the associated response for a given trial.

add_batch

add_batch(responses: list[Any], trial_batch: TrialBatch, contexts: list[Any] | None = None) -> None

Append responses for a batch of trials.

Parameters:

Name	Type	Description	Default
responses	List[Any]	Responses corresponding to each stimulus group in the trial batch.	required
trial_batch	TrialBatch	The batch of proposed trials.	required

Source code in src/psyphy/data/dataset.py

def add_batch(
    self,
    responses: list[Any],
    trial_batch: TrialBatch,
    contexts: list[Any] | None = None,
) -> None:
    """
    Append responses for a batch of trials.

    Parameters
    ----------
    responses : List[Any]
        Responses corresponding to each stimulus group in the trial batch.
    trial_batch : TrialBatch
        The batch of proposed trials.
    """
    if contexts is None:
        for input, resp in zip(trial_batch.stimuli, responses):
            self.add_trial(input, resp)
    else:
        for input, resp, context in zip(trial_batch.stimuli, responses, contexts):
            self.add_trial(input, resp, context)

add_trial

add_trial(input: tuple[Any, ...], resp: Any, context: Any = None) -> None

append a single trial.

Parameters:

Name	Type	Description	Default
input	tuple(Any, ...)	Group of presented stimuli each represented in any format (numpy array, list, etc.) Input must contain appropriate number of stimuli of appropriate dimension.	required
resp	Any	Subject response	required

Source code in src/psyphy/data/dataset.py

def add_trial(self, input: tuple[Any, ...], resp: Any, context: Any = None) -> None:
    """
    append a single trial.

    Parameters
    ----------
    input : tuple(Any, ...)
        Group of presented stimuli each represented in any format (numpy array,
        list, etc.)
        Input must contain appropriate number of stimuli of appropriate dimension.
    resp : Any
        Subject response
    """
    input_arr = np.atleast_2d(np.asarray(input))  # (K, d) — 1D input treated as K=1
    resp_arr = np.atleast_1d(np.asarray(resp))  # (R,)   — scalar treated as R=1
    if self.stimuli:
        if self.stim_shape != input_arr.shape:
            raise ValueError(
                f"stimuli must have consistent shape (K, d). Expected {self.stim_shape}, but received {input_arr.shape}"
            )
    else:
        self.stim_shape = input_arr.shape

    if context is None:
        if self.contexts:
            raise ValueError(
                "Context cannot be omitted if it was included in previous trials."
                "This ResponseData instance expected context but received none."
            )
    else:
        if self.contexts or self.stimuli == []:
            self.contexts.append(np.asarray(context))
        else:
            raise ValueError(
                "Context cannot be accepted if it was excluded from prior trials."
                f"This ResponseData instance expected no context, but received {context}"
            )
    self.stimuli.append(input_arr)
    self.responses.append(resp_arr)

copy

copy() -> ResponseData

Create a deep copy of this dataset.

Returns:

Type	Description
ResponseData	New dataset with copied data

Source code in src/psyphy/data/dataset.py

def copy(self) -> ResponseData:
    """
    Create a deep copy of this dataset.

    Returns
    -------
    ResponseData
        New dataset with copied data
    """
    new_data = ResponseData()
    new_data.stimuli = list(self.stimuli)
    new_data.responses = list(self.responses)
    if self.contexts is not None:
        new_data.contexts = list(self.contexts)
    return new_data

from_arrays

from_arrays(X: ndarray | ndarray, y: ndarray | ndarray, c: ndarray | ndarray | None = None) -> ResponseData

Construct ResponseData from arrays.

Parameters:

Name	Type	Description	Default
X	(array, shape(n_trials, n_stimuli, input_dim) or (n_trials, input_dim))	Stimuli. If 3D, second axis is input stumili. For OddityTask, this is (ref, comparison)	required
y	(array, shape(n_trials, response_dim))	Responses	required
c	optional array, shape (n_trials, context_dim)	Context	None

Returns:

Type	Description
ResponseData	Data container

OddityTask Example

From paired stimuli

X = jnp.array([[[0, 0], [1, 0]], [[1, 1], [2, 1]]])

X is formed from refs = [[0, 0], [1, 1]], comparisons = [[1, 0], [2, 1]]

y = jnp.array([1, 0]) data = ResponseData.from_arrays(X, y)

Source code in src/psyphy/data/dataset.py

@classmethod
def from_arrays(
    cls,
    X: jnp.ndarray | np.ndarray,
    y: jnp.ndarray | np.ndarray,
    c: jnp.ndarray | np.ndarray | None = None,
) -> ResponseData:
    """
    Construct ResponseData from arrays.

    Parameters
    ----------
    X : array, shape (n_trials, n_stimuli, input_dim) or (n_trials, input_dim)
        Stimuli. If 3D, second axis is input stumili. For OddityTask, this is
        (ref, comparison)
    y : array, shape (n_trials, response_dim)
        Responses
    c : optional array, shape (n_trials, context_dim)
        Context

    Returns
    -------
    ResponseData
        Data container

    OddityTask Example
    --------
    >>> # From paired stimuli
    >>> X = jnp.array([[[0, 0], [1, 0]], [[1, 1], [2, 1]]])
    >>> # X is formed from refs = [[0, 0], [1, 1]], comparisons = [[1, 0], [2, 1]]
    >>> y = jnp.array([1, 0])
    >>> data = ResponseData.from_arrays(X, y)
    """
    data = cls()

    X = np.asarray(X)
    y = np.asarray(y)

    if X.ndim == 2:
        # reshape to ensure appropriate conversion to stimuli groups
        dims = X.shape
        new_dims = (dims[0], 1, dims[1])
        X = np.reshape(X, new_dims)
    elif X.ndim != 3:
        raise ValueError(
            "X must be shape (n_trials, n_stimuli, input_dim) or \
            (n_trials, input_dim)."
        )
    if y.shape[0] != X.shape[0]:
        raise ValueError("X and y must contain the same n_trials.")
    if c is not None and c.shape[0] != X.shape[0]:
        raise ValueError("c must contain same n_trials as X.")

    # X is (n_trials, K, d) — split into per-trial tuples of K stimulus rows
    stimuli = []
    for plane in X:
        stimuli.append(tuple(plane))

    if c is not None:
        for stim, response, context in zip(stimuli, y, c):
            data.add_trial(stim, response, context)
    else:
        for stim, response in zip(stimuli, y):
            data.add_trial(stim, response)

    return data

from_trial_data

from_trial_data(data: TrialData) -> ResponseData

Build a ResponseData log from a :class:TrialData batch.

Source code in src/psyphy/data/dataset.py

@classmethod
def from_trial_data(cls, data: TrialData) -> ResponseData:
    """Build a ResponseData log from a :class:`TrialData` batch."""
    stimuli = np.asarray(data.stimuli)
    ys = np.asarray(data.responses)
    out = cls()
    if data.context is not None:
        cs = np.asarray(data.context)
        for s, y, c in zip(stimuli, ys, cs):
            out.add_trial(s, y, c)
    else:
        for s, y in zip(stimuli, ys):
            out.add_trial(s, y)
    return out

merge

merge(other: ResponseData) -> None

Merge another dataset into this one (in-place).

Parameters:

Name	Type	Description	Default
other	ResponseData	Dataset to merge	required

Source code in src/psyphy/data/dataset.py

def merge(self, other: ResponseData) -> None:
    """
    Merge another dataset into this one (in-place).

    Parameters
    ----------
    other : ResponseData
        Dataset to merge
    """
    no_empty = self.stimuli and other.stimuli

    if no_empty and self.stimuli[0].shape != other.stimuli[0].shape:
        raise ValueError(
            "Cannot merge ResponseData instances with inconsistent input shapes."
            f"Received input shapes of {self.stimuli[0].shape} and {other.stimuli[0].shape}"
        )
    if no_empty and self.responses[0].shape != other.responses[0].shape:
        raise ValueError(
            "Cannot merge ResponseData instances with inconsistent response shapes."
            f"Received response shapes of {self.responses[0].shape} and {other.responses[0].shape}"
        )

    self.stimuli.extend(other.stimuli)
    self.responses.extend(other.responses)
    both_contexts = self.contexts and other.contexts

    if self.contexts == [] and other.contexts == []:
        pass
    elif both_contexts and self.contexts[0].shape == other.contexts[0].shape:
        self.contexts.extend(other.contexts)
    else:
        raise ValueError(
            "Cannot merge ResponseData instances with inconsistent context."
        )

tail

tail(n: int) -> ResponseData

Return last n trials as a new ResponseData.

Parameters:

Name	Type	Description	Default
n	int	Number of trials to keep	required

Returns:

Type	Description
ResponseData	New dataset with last n trials

Source code in src/psyphy/data/dataset.py

def tail(self, n: int) -> ResponseData:
    """
    Return last n trials as a new ResponseData.

    Parameters
    ----------
    n : int
        Number of trials to keep

    Returns
    -------
    ResponseData
        New dataset with last n trials
    """
    new_data = ResponseData()
    new_data.stimuli = self.stimuli[-n:]
    new_data.responses = self.responses[-n:]
    if self.contexts is not None:
        new_data.contexts = self.contexts[-n:]
    return new_data

to_numpy

to_numpy() -> tuple[ndarray, ndarray]

Return stimuli and responses as NumPy arrays. Will NOT include contexts by default. Output always fixed length of 2.

Source code in src/psyphy/data/dataset.py

def to_numpy(self) -> tuple[np.ndarray, np.ndarray]:
    """Return stimuli and responses as NumPy arrays.
    Will NOT include contexts by default. Output always fixed length of 2.
    """
    return (
        np.asarray(self.stimuli),  # shape = (N, K, d)
        np.asarray(self.responses),
    )

to_trial_data

to_trial_data() -> TrialData

Convert this log into the canonical JAX batch (:class:TrialData).

Source code in src/psyphy/data/dataset.py

def to_trial_data(self) -> TrialData:
    """Convert this log into the canonical JAX batch (:class:`TrialData`)."""
    stimuli, responses = self.to_numpy()
    if self.contexts:
        context = np.asarray(self.contexts)
        return TrialData(
            stimuli=jnp.asarray(stimuli),
            responses=jnp.asarray(responses),
            context=jnp.asarray(context),
        )
    else:
        return TrialData(
            stimuli=jnp.asarray(stimuli), responses=jnp.asarray(responses)
        )

TrialBatch

TrialBatch(stimuli: list[tuple[Any, ...]])

Container for a proposed batch of trials. Does NOT include context or responses.

Attributes:

Name	Type	Description
stimuli	List[Tuple[Any, ...]]	Each trial is a tuple of all presented stimuli (stim1, stim2, ...). For OddityTask this is (reference, comparison)

Methods:

Name	Description
from_stimuli	Construct a TrialBatch from a list of stimuli (stim1, stim2, ...) groups.

Source code in src/psyphy/data/dataset.py

def __init__(self, stimuli: list[tuple[Any, ...]]) -> None:
    self.stimuli = list(stimuli)

stimuli

stimuli = list(stimuli)

from_stimuli

from_stimuli(groups: list[tuple[Any, ...]]) -> TrialBatch

Construct a TrialBatch from a list of stimuli (stim1, stim2, ...) groups.

Source code in src/psyphy/data/dataset.py

@classmethod
def from_stimuli(cls, groups: list[tuple[Any, ...]]) -> TrialBatch:
    """
    Construct a TrialBatch from a list of stimuli (stim1, stim2, ...) groups.
    """
    return cls(groups)

TrialData

TrialData(stimuli: ndarray, responses: ndarray, context: ndarray | None = None, stimulus_names: tuple[str, ...] = ())

Batched trial data for compute.

This is the canonical, compute-efficient representation of observed trials.

Shapes

stimuli : (N, K, d) responses : (N, R) context : optional (N, C)

Dimension key

N : number of trials (batch dimension) K : number of stimuli per trial (e.g. K=2 for a two-alternative task; K=2 for the oddity task, where the reference is presented twice but only the unique mean is stored — the duplication is encoded in the task likelihood, not here) d : dimensionality of each stimulus coordinate R : number of response channels (R=1 for binary; R=2 for e.g. (choice, RT)) C : number of context channels (observer-state covariates that condition the likelihood but are not part of the stimulus space, e.g. fatigue level)

Notes

• You can also think of this as a more generic ML-style dataset X with shape (N, K, d) plus y with shape (N, R).
• This is intended to be JAX-friendly (PyTree of arrays) so likelihood and inference code can be JIT-compiled without touching Python containers.
• Context is optional. No current inbuilt uses.

Methods:

Name	Description
stimulus	Return stimuli[:, k, :] for the slot named name.

Attributes:

Name	Type	Description
context	ndarray | None
num_trials	int	Number of trials (N).
responses	ndarray
stimuli	ndarray
stimulus_names	tuple[str, ...]

context

context: ndarray | None = None

num_trials

num_trials: int

Number of trials (N).

responses

responses: ndarray

stimuli

stimuli: ndarray

stimulus_names

stimulus_names: tuple[str, ...] = ()

stimulus

stimulus(name: str) -> ndarray

Return stimuli[:, k, :] for the slot named name.

Parameters:

Name	Type	Description	Default
name	str	Must match one of the entries in stimulus_names.	required

Returns:

Type	Description
(ndarray, shape(N, d))	Stimulus coordinates for all trials at the named slot.

Source code in src/psyphy/data/dataset.py

def stimulus(self, name: str) -> jnp.ndarray:
    """Return stimuli[:, k, :] for the slot named `name`.

    Parameters
    ----------
    name : str
        Must match one of the entries in ``stimulus_names``.

    Returns
    -------
    jnp.ndarray, shape (N, d)
        Stimulus coordinates for all trials at the named slot.
    """
    if not self.stimulus_names:
        raise ValueError(
            "stimulus_names is empty — set it at construction time to use "
            "named access, e.g. stimulus_names=('ref', 'comp')."
        )
    if name not in self.stimulus_names:
        raise ValueError(
            f"unknown stimulus name '{name}'. "
            f"Available names: {self.stimulus_names}."
        )
    idx = self.stimulus_names.index(name)  # resolved in Python, not JAX-traced
    return self.stimuli[:, idx, :]

---

I/O

---

io

io.py

I/O utilities for saving and loading psyphy data.

Supports: - CSV for human-readable trial logs

Notes

• Data is stored in NumPy arrays (via ResponseData.to_numpy()).
• Convert to jax.numpy when passing into models.

Functions:

Name	Description
load_posterior	Load a Posterior object from pickle.
load_responses_csv	Load ResponseData from a CSV file.
save_posterior	Save a Posterior object to disk using pickle.
save_responses_csv	Save ResponseData to a CSV file. This is completely task agnostic. In the

Attributes:

Name	Type	Description
PathLike

PathLike

PathLike = Union[str, Path]

load_posterior

load_posterior(path: PathLike) -> object

Load a Posterior object from pickle.

Source code in src/psyphy/data/io.py

def load_posterior(path: PathLike) -> object:
    """
    Load a Posterior object from pickle.
    """
    with open(path, "rb") as f:
        return pickle.load(f)

load_responses_csv

load_responses_csv(path: PathLike) -> TrialData

Load ResponseData from a CSV file. Currently catering to OddityTask data format ONLY.

Parameters:

Name	Type	Description	Default
path	str or Path	Must be of expected format for OddityTask.	required

Returns:

Type	Description
ResponseData

Source code in src/psyphy/data/io.py

def load_responses_csv(path: PathLike) -> TrialData:
    """
    Load ResponseData from a CSV file.
    Currently catering to OddityTask data format ONLY.

    Parameters
    ----------
    path : str or Path
        Must be of expected format for OddityTask.

    Returns
    -------
    ResponseData
    """
    data = ResponseData()
    with open(path) as f:
        reader = csv.DictReader(f)
        for row in reader:
            ref = ast.literal_eval(row["ref"])
            probe = ast.literal_eval(row["probe"])
            resp = int(row["response"])
            data.add_trial((ref, probe), resp)
    return data.to_trial_data()

save_posterior

save_posterior(posterior: object, path: PathLike) -> None

Save a Posterior object to disk using pickle.

Source code in src/psyphy/data/io.py

def save_posterior(posterior: object, path: PathLike) -> None:
    """
    Save a Posterior object to disk using pickle.
    """
    with open(path, "wb") as f:
        pickle.dump(posterior, f)

save_responses_csv

save_responses_csv(data: TrialData | ResponseData, path: PathLike) -> None

Save ResponseData to a CSV file. This is completely task agnostic. In the current implementation, it will simply create as many stimulus columns as inputs in the data and will label them "stimulus 1", "stimulus 2", etc. The response column will be labeled "response"

Parameters:

Name	Type	Description	Default
data	ResponseData		required
path	str or Path		required

Source code in src/psyphy/data/io.py

def save_responses_csv(data: TrialData | ResponseData, path: PathLike) -> None:
    """
    Save ResponseData to a CSV file. This is completely task agnostic. In the
    current implementation, it will simply create as many stimulus columns as
    inputs in the data and will label them "stimulus 1", "stimulus 2", etc. The
    response column will be labeled "response"

    Parameters
    ----------
    data : ResponseData
    path : str or Path
    """
    if isinstance(data, TrialData):
        inputs, resps = (
            np.asarray(data.stimuli),
            np.asarray(data.responses),
        )
    else:
        inputs, resps = data.to_numpy()
    row_names = []
    for s in range(inputs.shape[1]):
        row_names.append("stimulus " + str(s))
    row_names.append("response")
    with open(path, "w", newline="") as f:
        writer = csv.writer(f)
        writer.writerow(row_names)
        for x, y in zip(inputs, resps):
            row = x.tolist()
            row.append(y.tolist())
            writer.writerow(row)

---

Published Datasets

---

published

psyphy.data.published

Loaders for published third-party psychophysical datasets.

These are datasets released alongside papers, used to validate psyphy against results obtained independently. This is distinct from :mod:psyphy.data.dataset, which provides the containers your own experiment data lives in.

psyphy ships no data. Each submodule exposes a fetch function that downloads on request into a user cache directory (see hong2025.default_data_dir), and loaders that return psyphy objects such as :class:~psyphy.data.dataset.TrialData.

Available datasets

• :mod:~psyphy.data.published.hong2025 — human color discrimination, 8 observers, Wishart process fits. Hong et al. (2026), eLife 14:RP108943.

Examples:

>>> from psyphy.data.published import hong2025
>>> paths = hong2025.fetch(subject=1)
>>> data = hong2025.load_trials(paths["trials"])

---

hong2025

hong2025.py

Loaders for the published human colour-discrimination dataset of Hong et al.

Hong, F., Bouhassira, R., Chow, J., Sanders, C., Shvartsman, M., Guan, P.,
Williams, A. H., & Brainard, D. H. (2026). Comprehensive characterization of
human color discrimination thresholds. eLife, 14:RP108943.
https://doi.org/10.7554/eLife.108943.2

Hong et al 2025 fitted a Wishart Process Psychophysical Model to eight observers and published both the trial-level data and the fitted Chebyshev weights.

Data availability

Hosted on OSF node k27js (https://osf.io/k27js). psyphy ships no data. :func:fetch downloads on request into a user cache directory. At the time of writing the OSF node carries no explicit license, so please always cite paper!

Typical use

from psyphy.data.published import hong2025 paths = hong2025.fetch(subject=1) # doctest: +SKIP data = hong2025.load_trials(paths["trials"]) # doctest: +SKIP W_org = hong2025.load_reference_W(paths["weights"]) # doctest: +SKIP

Coordinate convention

All stimulus coordinates are in the paper's 2-D "W space" and already lie in the Chebyshev domain [-1, 1]. No normalisation is applied or needed.

See also

docs/examples/wppm/hong2025_reproduction.md for the end-to-end walkthrough.

Functions:

Name	Description
build_paper_model	Build a :class:~psyphy.model.wppm.WPPM configured as in the paper.
default_data_dir	Return the directory :func:fetch downloads into.
fetch	Download one observer's files from OSF node k27js.
fetch_calibration_matrix	Download the 2-D W space -> RGB transformation matrix from OSF.
load_calibration_matrix	Load a 3x3 transformation matrix CSV.
load_reference_W	Load published Chebyshev weights from Bestfit_W_sub{N}.csv.
load_sigma_table	Load a published covariance table.
load_trials	Load trial_data_pooled_by_type_sub{N}.csv into a :class:TrialData.
w2d_to_rgb	Convert 2-D W space coordinates to linear monitor RGB.

Attributes:

Name	Type	Description
CALIBRATION_DATE
PAPER_HYPERPARAMS	Mapping[str, Any]
SUBJECT_INITIALS	Mapping[int, str]

CALIBRATION_DATE

CALIBRATION_DATE = '02242025'

FIT_TRIAL_TYPES

FIT_TRIAL_TYPES: tuple[str, ...] = ('AEPsych',)

OSF_NODE

OSF_NODE = 'k27js'

PAPER_HYPERPARAMS

PAPER_HYPERPARAMS: Mapping[str, Any] = {'basis_degree': 4, 'input_dim': 2, 'extra_dims': 1, 'variance_scale': 0.0003, 'decay_rate': 0.4, 'diag_term': 0.0, 'mc_samples': 2000, 'bandwidth': 0.005, 'learning_rate': 0.0001, 'momentum': 0.2, 'total_steps': 1500, 'n_restarts': 1, 'target_pC': 0.667}

SUBJECT_INITIALS

SUBJECT_INITIALS: Mapping[int, str] = {1: 'CH', 2: 'ME', 4: 'SG', 6: 'DK', 7: 'BH', 8: 'FM', 10: 'HG', 11: 'FW'}

build_paper_model

build_paper_model(*, mc_samples: int | None = None, bandwidth: float | None = None) -> Any

Build a :class:~psyphy.model.wppm.WPPM configured as in the paper.

Every setting comes from :data:PAPER_HYPERPARAMS, itself a transcription of ellipsoids/fit/fit_4d_human.py SECTION 4. Keeping it here rather than in the example script gives the tutorial and the regression test a single source of truth for the paper-to-psyphy mapping.

Parameters:

Name	Type	Description	Default
mc_samples	int	Monte Carlo samples per trial in the oddity likelihood. Defaults to the paper's 2000, which is expensive; lower it for quick runs.	None
bandwidth	float	Logistic smoothing bandwidth. Defaults to the paper's 5e-3.	None

Returns:

Type	Description
WPPM	Ready to fit, or to wrap in a :class:~psyphy.model.covariance_field.WPPMCovarianceField for evaluating covariances from published weights.

Notes

Imported lazily so that psyphy.data does not depend on psyphy.model at import time.

Source code in src/psyphy/data/published/hong2025.py

def build_paper_model(
    *, mc_samples: int | None = None, bandwidth: float | None = None
) -> Any:
    """Build a :class:`~psyphy.model.wppm.WPPM` configured as in the paper.

    Every setting comes from :data:`PAPER_HYPERPARAMS`, itself a transcription
    of ``ellipsoids/fit/fit_4d_human.py`` SECTION 4. Keeping it here rather than
    in the example script gives the tutorial and the regression test a single
    source of truth for the paper-to-psyphy mapping.

    Parameters
    ----------
    mc_samples : int, optional
        Monte Carlo samples per trial in the oddity likelihood. Defaults to the
        paper's 2000, which is expensive; lower it for quick runs.
    bandwidth : float, optional
        Logistic smoothing bandwidth. Defaults to the paper's 5e-3.

    Returns
    -------
    WPPM
        Ready to fit, or to wrap in a
        :class:`~psyphy.model.covariance_field.WPPMCovarianceField` for
        evaluating covariances from published weights.

    Notes
    -----
    Imported lazily so that ``psyphy.data`` does not depend on ``psyphy.model``
    at import time.
    """
    from psyphy.model.likelihood import OddityTask, OddityTaskConfig
    from psyphy.model.noise import GaussianNoise
    from psyphy.model.prior import Prior
    from psyphy.model.wppm import WPPM

    prior = Prior(
        input_dim=PAPER_HYPERPARAMS["input_dim"],
        basis_degree=PAPER_HYPERPARAMS["basis_degree"],
        extra_embedding_dims=PAPER_HYPERPARAMS["extra_dims"],
        variance_scale=PAPER_HYPERPARAMS["variance_scale"],
        decay_rate=PAPER_HYPERPARAMS["decay_rate"],
    )
    task = OddityTask(
        config=OddityTaskConfig(
            num_samples=int(
                mc_samples
                if mc_samples is not None
                else PAPER_HYPERPARAMS["mc_samples"]
            ),
            bandwidth=float(
                bandwidth if bandwidth is not None else PAPER_HYPERPARAMS["bandwidth"]
            ),
        )
    )
    return WPPM(
        input_dim=PAPER_HYPERPARAMS["input_dim"],
        extra_dims=PAPER_HYPERPARAMS["extra_dims"],
        prior=prior,
        likelihood=task,
        noise=GaussianNoise(),
        diag_term=PAPER_HYPERPARAMS["diag_term"],
    )

default_data_dir

default_data_dir() -> Path

Return the directory :func:fetch downloads into.

$PSYPHY_DATA_HOME if set, otherwise ~/.cache/psyphy/hong2025. Deliberately outside the repository so downloaded data is never staged for commit.

Source code in src/psyphy/data/published/hong2025.py

def default_data_dir() -> Path:
    """Return the directory :func:`fetch` downloads into.

    ``$PSYPHY_DATA_HOME`` if set, otherwise ``~/.cache/psyphy/hong2025``.
    Deliberately outside the repository so downloaded data is never staged for
    commit.
    """
    root = os.environ.get("PSYPHY_DATA_HOME")
    base = Path(root) if root else Path.home() / ".cache" / "psyphy"
    return base / "hong2025"

fetch

fetch(subject: int = 1, data_dir: str | Path | None = None, *, noise_ellipses: bool = False, verbose: bool = True) -> dict[str, Path]

Download one observer's files from OSF node k27js.

Parameters:

Name	Type	Description	Default
subject	int	Observer number. See :data:SUBJECT_INITIALS; 1 is observer CH.	1
data_dir	str or Path	Destination root. Defaults to :func:default_data_dir.	None
noise_ellipses	bool	Also fetch Noise_ellipses_sub{N}.csv. This is ~68 MB, and holds the published noise covariances on a 103x103 grid. Needed only to verify psyphy against the published covariance field exactly.	False
verbose	bool	Print progress.	True

Returns:

Type	Description
dict[str, Path]	Keys "trials", "weights", "thres_ellipses", and — when requested — "noise_ellipses".

Notes

Requires network access; nothing else in this module does. Files already present at the expected size are skipped, so repeated calls are cheap.

Source code in src/psyphy/data/published/hong2025.py

def fetch(
    subject: int = 1,
    data_dir: str | Path | None = None,
    *,
    noise_ellipses: bool = False,
    verbose: bool = True,
) -> dict[str, Path]:
    """Download one observer's files from OSF node ``k27js``.

    Parameters
    ----------
    subject : int, default=1
        Observer number. See :data:`SUBJECT_INITIALS`; ``1`` is observer CH.
    data_dir : str or Path, optional
        Destination root. Defaults to :func:`default_data_dir`.
    noise_ellipses : bool, default=False
        Also fetch ``Noise_ellipses_sub{N}.csv``. This is ~68 MB, and holds the
        published noise covariances on a 103x103 grid. Needed only to verify
        psyphy against the published covariance field exactly.
    verbose : bool, default=True
        Print progress.

    Returns
    -------
    dict[str, Path]
        Keys ``"trials"``, ``"weights"``, ``"thres_ellipses"``, and — when
        requested — ``"noise_ellipses"``.

    Notes
    -----
    Requires network access; nothing else in this module does. Files already
    present at the expected size are skipped, so repeated calls are cheap.


    """
    targets = {
        "trials": f"trial_data_pooled_by_type_sub{subject}.csv",
        "weights": f"Bestfit_W_sub{subject}.csv",
        "thres_ellipses": f"Thres_ellipses_sub{subject}.csv",
    }
    if noise_ellipses:
        targets["noise_ellipses"] = f"Noise_ellipses_sub{subject}.csv"

    root_dir = Path(data_dir) if data_dir is not None else default_data_dir()
    dest_dir = root_dir / f"sub{subject}"

    ctx = _ssl_context()
    if verbose:
        initials = SUBJECT_INITIALS.get(subject, "?")
        print(f"OSF {OSF_NODE} sub{subject} ({initials}) -> {dest_dir}")

    root_items = _list_folder(f"{_OSF_API}/nodes/{OSF_NODE}/files/osfstorage/", ctx)
    org_url = _subfolder_url(root_items, _ORGANIZED)
    if not org_url:
        raise RuntimeError(f"'{_ORGANIZED}' not found on OSF node {OSF_NODE}.")

    org_items = _list_folder(org_url, ctx)
    sub_url = _subfolder_url(org_items, f"sub{subject}")
    if not sub_url:
        available = sorted(
            i["attributes"]["name"]
            for i in org_items
            if i.get("attributes", {}).get("kind") == "folder"
        )
        raise ValueError(f"sub{subject} not found on OSF. Available: {available}")

    sub_items = _list_folder(sub_url, ctx)
    paths: dict[str, Path] = {}
    for key, name in targets.items():
        item = _find_file(sub_items, name)
        if item is None:
            raise RuntimeError(f"'{name}' not found under sub{subject} on OSF.")
        dest = dest_dir / name
        _download(item, dest, ctx, verbose=verbose)
        paths[key] = dest
    return paths

fetch_calibration_matrix

fetch_calibration_matrix(data_dir: str | Path | None = None, *, file_date: str = CALIBRATION_DATE, verbose: bool = True) -> Path

Download the 2-D W space -> RGB transformation matrix from OSF.

Separate from :func:fetch because this file is not per-subject and does not live under the same OSF folder: it sits in Calibration and transformation/Transformation matrices/, alongside the monitor colorimetry matrices, and one copy serves every observer.

Parameters:

Name	Type	Description	Default
data_dir	str or Path	Destination root. Defaults to :func:default_data_dir; the file lands in a calibration/ subdirectory of it.	None
file_date	str	Calibration date stamp embedded in the filename.	:data:`CALIBRATION_DATE`
verbose	bool	Print progress.	True

Returns:

Type	Description
Path	Local path to M_2DWToRGB_DELL_<file_date>_copy.csv.

See Also

load_calibration_matrix : read the downloaded file. w2d_to_rgb : apply it.

Source code in src/psyphy/data/published/hong2025.py

def fetch_calibration_matrix(
    data_dir: str | Path | None = None,
    *,
    file_date: str = CALIBRATION_DATE,
    verbose: bool = True,
) -> Path:
    """Download the 2-D W space -> RGB transformation matrix from OSF.

    Separate from :func:`fetch` because this file is not per-subject and does
    not live under the same OSF folder: it sits in
    ``Calibration and transformation/Transformation matrices/``, alongside the
    monitor colorimetry matrices, and one copy serves every observer.

    Parameters
    ----------
    data_dir : str or Path, optional
        Destination root. Defaults to :func:`default_data_dir`; the file lands
        in a ``calibration/`` subdirectory of it.
    file_date : str, default=:data:`CALIBRATION_DATE`
        Calibration date stamp embedded in the filename.
    verbose : bool, default=True
        Print progress.

    Returns
    -------
    Path
        Local path to ``M_2DWToRGB_DELL_<file_date>_copy.csv``.

    See Also
    --------
    load_calibration_matrix : read the downloaded file.
    w2d_to_rgb : apply it.
    """
    root_dir = Path(data_dir) if data_dir is not None else default_data_dir()
    dest_dir = root_dir / "calibration"

    ctx = _ssl_context()
    if verbose:
        print(f"OSF {OSF_NODE} calibration -> {dest_dir}")

    root_items = _list_folder(f"{_OSF_API}/nodes/{OSF_NODE}/files/osfstorage/", ctx)
    cal_url = _subfolder_url(root_items, _CALIBRATION)
    if not cal_url:
        raise RuntimeError(f"'{_CALIBRATION}' not found on OSF node {OSF_NODE}.")

    cal_items = _list_folder(cal_url, ctx)
    matrices_url = _subfolder_url(cal_items, _TRANSFORM_MATRICES)
    if not matrices_url:
        raise RuntimeError(
            f"'{_TRANSFORM_MATRICES}' not found under '{_CALIBRATION}' "
            f"on OSF node {OSF_NODE}."
        )

    matrix_items = _list_folder(matrices_url, ctx)
    pattern = f"M_2DWToRGB*{file_date}*.csv"
    matches = _match_files(matrix_items, pattern)
    if not matches:
        available = sorted(
            str(i["attributes"]["name"])
            for i in matrix_items
            if i.get("attributes", {}).get("kind") == "file"
        )
        raise RuntimeError(
            f"No file matching '{pattern}' under '{_TRANSFORM_MATRICES}'. "
            f"Available: {available}"
        )
    if len(matches) > 1:
        names = sorted(str(m["attributes"]["name"]) for m in matches)
        raise RuntimeError(
            f"'{pattern}' matched more than one file: {names}. "
            "Pass a more specific file_date."
        )

    item = matches[0]
    dest = dest_dir / str(item["attributes"]["name"])
    _download(item, dest, ctx, verbose=verbose)
    return dest

load_calibration_matrix

load_calibration_matrix(path: str | Path) -> ndarray

Load a 3x3 transformation matrix CSV.

Parameters:

Name	Type	Description	Default
path	str or Path	Path to a M_*To*_DELL_*.csv file from :func:fetch_calibration_matrix.	required

Returns:

Type	Description
(ndarray, shape(3, 3))	The matrix, oriented to be applied to column vectors from the left, exactly as stored.

Notes

Header-less, comma-delimited, three rows. The OSF folder's own README documents np.loadtxt as the intended reader.

Source code in src/psyphy/data/published/hong2025.py

def load_calibration_matrix(path: str | Path) -> np.ndarray:
    """Load a 3x3 transformation matrix CSV.

    Parameters
    ----------
    path : str or Path
        Path to a ``M_*To*_DELL_*.csv`` file from
        :func:`fetch_calibration_matrix`.

    Returns
    -------
    np.ndarray, shape (3, 3)
        The matrix, oriented to be applied to **column** vectors from the
        left, exactly as stored.

    Notes
    -----
    Header-less, comma-delimited, three rows. The OSF folder's own README
    documents ``np.loadtxt`` as the intended reader.
    """
    M = np.loadtxt(path, delimiter=",", dtype=np.float64)
    if M.shape != (3, 3):
        raise ValueError(f"{path}: expected a 3x3 matrix, got shape {M.shape}.")
    return M

load_reference_W

load_reference_W(path: str | Path, column: str = 'W_org') -> ndarray

Load published Chebyshev weights from Bestfit_W_sub{N}.csv.

Parameters:

Name	Type	Description	Default
path	str or Path	Path to the best-fit weights CSV (the "weights" entry from :func:fetch).	required
column	str	Which fit to read. "W_org" is the main published fit; the file also carries 120 bootstrap columns named W_btst{b}_rank{r}.	"W_org"

Returns:

Type	Description
ndarray	Shape (5, 5, 2, 3) — exactly psyphy's params["W"] layout for basis_degree=4, input_dim=2, extra_embedding_dims=1, so it can be used directly as {"W": load_reference_W(...)} with no reshaping.

Raises:

Type	Description
ValueError	If the index column is malformed, the requested column is absent, index keys are duplicated, or the table is not dense.

Notes

Enable jax_enable_x64 before calling if you intend to compare against published covariances; in float32 the agreement floors around 1e-7.

Source code in src/psyphy/data/published/hong2025.py

def load_reference_W(path: str | Path, column: str = "W_org") -> jnp.ndarray:
    """Load published Chebyshev weights from ``Bestfit_W_sub{N}.csv``.

    Parameters
    ----------
    path : str or Path
        Path to the best-fit weights CSV (the ``"weights"`` entry from
        :func:`fetch`).
    column : str, default="W_org"
        Which fit to read. ``"W_org"`` is the main published fit; the file also
        carries 120 bootstrap columns named ``W_btst{b}_rank{r}``.

    Returns
    -------
    jnp.ndarray
        Shape (5, 5, 2, 3) — exactly psyphy's ``params["W"]`` layout for
        ``basis_degree=4, input_dim=2, extra_embedding_dims=1``, so it can be
        used directly as ``{"W": load_reference_W(...)}`` with no reshaping.

    Raises
    ------
    ValueError
        If the index column is malformed, the requested column is absent, index
        keys are duplicated, or the table is not dense.

    Notes
    -----
    Enable ``jax_enable_x64`` before calling if you intend to compare against
    published covariances; in float32 the agreement floors around 1e-7.
    """
    with open(path, newline="") as fh:
        reader = csv.DictReader(fh)
        fields = reader.fieldnames or []
        index_col = fields[0] if fields else None
        if index_col != "i,j,k,l":
            raise ValueError(
                f"{path}: expected first column 'i,j,k,l', got {index_col!r}."
            )
        if column not in fields:
            available = [f for f in fields if f != index_col]
            raise ValueError(
                f"{path}: column {column!r} not found. "
                f"{len(available)} available, e.g. {available[:3]}."
            )
        entries = [
            (tuple(int(v) for v in row[index_col].split(",")), float(row[column]))
            for row in reader
        ]

    if not entries:
        raise ValueError(f"{path}: no rows.")

    shape = tuple(max(idx[axis] for idx, _ in entries) + 1 for axis in range(4))
    expected = int(np.prod(shape))
    if len(entries) != expected:
        raise ValueError(
            f"{path}: got {len(entries)} rows but the index range implies shape "
            f"{shape} ({expected} entries) -- the table is not dense."
        )

    # A row count alone is not enough: one duplicated index key plus one missing
    # key preserves both the count and the per-axis maxima, and the missing cell
    # would then hold whatever the allocation happened to contain. Check
    # uniqueness, and fill with NaN so any remaining gap is loud rather than
    # plausible-looking.
    indices = [idx for idx, _ in entries]
    if len(set(indices)) != len(indices):
        raise ValueError(f"{path}: duplicate index keys in column {column!r}.")

    W = np.full(shape, np.nan, dtype=np.float64)
    for idx, value in entries:
        W[idx] = value
    if not np.isfinite(W).all():
        n_bad = int((~np.isfinite(W)).sum())
        raise ValueError(f"{path}: {n_bad} unfilled entries in column {column!r}.")
    return jnp.asarray(W)

load_sigma_table

load_sigma_table(path: str | Path, *, value_column: str | None = None) -> tuple[ndarray, ndarray]

Load a published covariance table.

Handles both Thres_ellipses_sub{N}.csv (49 rows, the 7x7 reference grid, 66.7%-correct threshold covariances) and Noise_ellipses_sub{N}.csv (10 609 rows, a 103x103 grid, model noise covariances). They share a layout: one row per reference location, a first column holding the coordinate as "x,y", then one column per fit holding a 2x2 matrix as "[[a,b],[c,d]]".

Parameters:

Name	Type	Description	Default
path	str or Path	Path to either ellipse CSV.	required
value_column	str	Which fit to read. Defaults to the file's main fit — whichever *_org column is present.	None

Returns:

Name	Type	Description
coords	(ndarray, shape(M, 2))	Reference locations in W space.
Sigmas	(ndarray, shape(M, 2, 2))	Covariance matrix at each location.

Notes

Because every row carries its own coordinate, the row/column ordering ambiguity that affects the published .pkl files — which store bare (7, 7, 2, 2) arrays with no coordinates attached — does not arise here.

Source code in src/psyphy/data/published/hong2025.py

def load_sigma_table(
    path: str | Path, *, value_column: str | None = None
) -> tuple[np.ndarray, np.ndarray]:
    """Load a published covariance table.

    Handles both ``Thres_ellipses_sub{N}.csv`` (49 rows, the 7x7 reference grid,
    66.7%-correct threshold covariances) and ``Noise_ellipses_sub{N}.csv``
    (10 609 rows, a 103x103 grid, model noise covariances). They share a layout:
    one row per reference location, a first column holding the coordinate as
    ``"x,y"``, then one column per fit holding a 2x2 matrix as
    ``"[[a,b],[c,d]]"``.

    Parameters
    ----------
    path : str or Path
        Path to either ellipse CSV.
    value_column : str, optional
        Which fit to read. Defaults to the file's main fit — whichever
        ``*_org`` column is present.

    Returns
    -------
    coords : np.ndarray, shape (M, 2)
        Reference locations in W space.
    Sigmas : np.ndarray, shape (M, 2, 2)
        Covariance matrix at each location.

    Notes
    -----
    Because every row carries its own coordinate, the row/column ordering
    ambiguity that affects the published ``.pkl`` files — which store bare
    (7, 7, 2, 2) arrays with no coordinates attached — does not arise here.
    """
    with open(path, newline="") as fh:
        reader = csv.DictReader(fh)
        fields = reader.fieldnames or []
        coord_col = fields[0] if fields else None
        if coord_col not in ("grid_ref", "grid_ref_fine"):
            raise ValueError(
                f"{path}: expected first column 'grid_ref' or 'grid_ref_fine', "
                f"got {coord_col!r}."
            )
        if value_column is None:
            candidates = [f for f in fields if f.endswith("_org")]
            if not candidates:
                raise ValueError(f"{path}: no '*_org' column found in {fields[:4]}.")
            value_column = candidates[0]
        elif value_column not in fields:
            raise ValueError(f"{path}: column {value_column!r} not found.")

        coords, sigmas = [], []
        for row in reader:
            coords.append(_parse_vec(row[coord_col]))
            sigmas.append(ast.literal_eval(row[value_column]))

    if not coords:
        raise ValueError(f"{path}: no rows.")
    return np.asarray(coords, dtype=np.float64), np.asarray(sigmas, dtype=np.float64)

load_trials

load_trials(path: str | Path, *, trial_types: Sequence[str] | None = FIT_TRIAL_TYPES, max_trials: int | None = None, seed: int = 0) -> TrialData

Load trial_data_pooled_by_type_sub{N}.csv into a :class:TrialData.

Parameters:

Name	Type	Description	Default
path	str or Path	Path to the pooled trial CSV (the "trials" entry from :func:fetch).	required
trial_types	sequence of str	Keep only rows whose TrialType starts with one of these prefixes. Defaults to :data:FIT_TRIAL_TYPES, reproducing the published fit. Pass None to keep every row (AEPsych and MOCS).	FIT_TRIAL_TYPES
max_trials	int	Take a reproducible random subsample of this many trials. Useful for quick runs; None (default) keeps all matching trials.	None
seed	int	Seed for the subsample.	0

Returns:

Type	Description
TrialData	stimuli has shape (N, 2, 2) with slot 0 the reference and slot 1 the comparison; responses has shape (N, 1), 1 meaning a correct oddity identification. stimulus_names is ("ref", "comp").

Notes

The oddity task presents three stimuli (reference, reference, comparison) but only two distinct means, so K=2. The duplication is encoded in the task likelihood, not the data container i.e., the same convention psyphy uses elsewhere, and the same one the paper uses (its data tuple is (y, mref, mprobe)).

Source code in src/psyphy/data/published/hong2025.py

def load_trials(
    path: str | Path,
    *,
    trial_types: Sequence[str] | None = FIT_TRIAL_TYPES,
    max_trials: int | None = None,
    seed: int = 0,
) -> TrialData:
    """Load ``trial_data_pooled_by_type_sub{N}.csv`` into a :class:`TrialData`.

    Parameters
    ----------
    path : str or Path
        Path to the pooled trial CSV (the ``"trials"`` entry from :func:`fetch`).
    trial_types : sequence of str, optional
        Keep only rows whose ``TrialType`` starts with one of these prefixes.
        Defaults to :data:`FIT_TRIAL_TYPES`, reproducing the published fit. Pass
        ``None`` to keep every row (AEPsych and MOCS).
    max_trials : int, optional
        Take a reproducible random subsample of this many trials. Useful for
        quick runs; ``None`` (default) keeps all matching trials.
    seed : int, default=0
        Seed for the subsample.

    Returns
    -------
    TrialData
        ``stimuli`` has shape (N, 2, 2) with slot 0 the reference and slot 1 the
        comparison; ``responses`` has shape (N, 1), 1 meaning a correct oddity
        identification. ``stimulus_names`` is ``("ref", "comp")``.

    Notes
    -----
    The oddity task presents three stimuli (reference, reference, comparison)
    but only two *distinct* means, so K=2. The duplication is encoded in the
    task likelihood, not the data container i.e., the same convention psyphy uses
    elsewhere, and the same one the paper uses (its data tuple is
    ``(y, mref, mprobe)``).
    """
    refs: list[tuple[float, ...]] = []
    comps: list[tuple[float, ...]] = []
    resps: list[int] = []
    prefixes = tuple(trial_types) if trial_types is not None else None

    with open(path, newline="") as fh:
        reader = csv.DictReader(fh)
        missing = {"TrialType", "xref", "x1", "y"} - set(reader.fieldnames or [])
        if missing:
            raise ValueError(
                f"{path} is missing expected column(s) {sorted(missing)}; "
                f"found {reader.fieldnames}."
            )
        for row in reader:
            if prefixes is not None and not row["TrialType"].startswith(prefixes):
                continue
            refs.append(_parse_vec(row["xref"]))
            comps.append(_parse_vec(row["x1"]))
            resps.append(int(row["y"]))

    if not refs:
        raise ValueError(
            f"No trials matched trial_types={trial_types!r} in {path}. "
            'Expected prefixes such as "AEPsych" or "MOCS".'
        )

    stimuli = np.stack([np.asarray(refs), np.asarray(comps)], axis=1)  # (N, 2, 2)
    responses = np.asarray(resps)[:, None]  # (N, 1)

    if max_trials is not None and max_trials < stimuli.shape[0]:
        rng = np.random.default_rng(seed)
        idx = np.sort(rng.choice(stimuli.shape[0], size=max_trials, replace=False))
        stimuli, responses = stimuli[idx], responses[idx]

    return TrialData(
        stimuli=jnp.asarray(stimuli),
        responses=jnp.asarray(responses),
        stimulus_names=("ref", "comp"),
    )

w2d_to_rgb

w2d_to_rgb(coords: ndarray, M: ndarray) -> ndarray

Convert 2-D W space coordinates to linear monitor RGB.

The transform is affine, handled by augmenting each coordinate with a constant 1 so that the matrix's third column supplies the DC offset between W-space origin and monitor gray::

rgb = clip(M @ [w1, w2, 1], 0, 1)

Parameters:

Name	Type	Description	Default
coords	(ndarray, shape(2) or (N, 2))	Stimulus coordinates in the paper's 2-D W space.	required
M	(ndarray, shape(3, 3))	Matrix from :func:load_calibration_matrix.	required

Returns:

Type	Description
(ndarray, shape(3) or (N, 3))	Linear RGB in [0, 1], matching the input's dimensionality.

Notes

These are linear, non-gamma-corrected RGB values, which is what the paper's own plotting code produces and passes to matplotlib. Applying an sRGB gamma would give more perceptually even colours but would no longer match the published figures.

The matrix is monitor-specific -- one Dell display, calibrated on the date in :data:CALIBRATION_DATE. It reproduces this paper's figures; it is not a general W-space-to-RGB conversion, and a different display needs its own calibration.

Values outside the monitor gamut are clipped, so a returned colour on a channel boundary may not be the exact requested chromaticity. Over the paper's own [-0.7, 0.7] reference grid no clipping occurs.

Examples:

>>> M = load_calibration_matrix(fetch_calibration_matrix())
>>> w2d_to_rgb(np.zeros(2), M)
array([0.5, 0.49999396, 0.5])

Source code in src/psyphy/data/published/hong2025.py

def w2d_to_rgb(coords: np.ndarray, M: np.ndarray) -> np.ndarray:
    """Convert 2-D W space coordinates to linear monitor RGB.

    The transform is affine, handled by augmenting each coordinate with a
    constant 1 so that the matrix's third column supplies the DC offset
    between W-space origin and monitor gray::

        rgb = clip(M @ [w1, w2, 1], 0, 1)

    Parameters
    ----------
    coords : np.ndarray, shape (2,) or (N, 2)
        Stimulus coordinates in the paper's 2-D W space.
    M : np.ndarray, shape (3, 3)
        Matrix from :func:`load_calibration_matrix`.

    Returns
    -------
    np.ndarray, shape (3,) or (N, 3)
        Linear RGB in [0, 1], matching the input's dimensionality.

    Notes
    -----
    **These are linear, non-gamma-corrected RGB values**, which is what the
    paper's own plotting code produces and passes to matplotlib. Applying an
    sRGB gamma would give more perceptually even colours but would no longer
    match the published figures.

    **The matrix is monitor-specific** -- one Dell display, calibrated on the
    date in :data:`CALIBRATION_DATE`. It reproduces *this paper's* figures; it
    is not a general W-space-to-RGB conversion, and a different display needs
    its own calibration.

    Values outside the monitor gamut are clipped, so a returned colour on a
    channel boundary may not be the exact requested chromaticity. Over the
    paper's own [-0.7, 0.7] reference grid no clipping occurs.

    Examples
    --------
    >>> M = load_calibration_matrix(fetch_calibration_matrix())  # doctest: +SKIP
    >>> w2d_to_rgb(np.zeros(2), M)  # doctest: +SKIP
    array([0.5, 0.49999396, 0.5])
    """
    xy = np.asarray(coords, dtype=np.float64)
    single = xy.ndim == 1
    if single:
        xy = xy[None, :]
    if xy.ndim != 2 or xy.shape[-1] != 2:
        raise ValueError(
            f"coords must have shape (2,) or (N, 2), got {np.shape(coords)}."
        )

    augmented = np.concatenate([xy, np.ones((xy.shape[0], 1))], axis=1)  # (N, 3)
    rgb = np.clip(augmented @ np.asarray(M, dtype=np.float64).T, 0.0, 1.0)
    return rgb[0] if single else rgb

---