Demo experiment

random_dots/
    experiment.py
    params.py
    remote.py

Experiment module (experiment.py)

"""Simple reaction time random dot motion experiment."""
import pandas as pd
from psychopy import core
from visigoth import AcquireFixation, AcquireTarget, flexible_values
from visigoth.stimuli import RandomDotMotion, Point, Points


def create_stimuli(exp):
    """Initialize the stimulus objects used in the experiment."""
    # Central fixation point
    fix = Point(exp.win,
                exp.p.fix_pos,
                exp.p.fix_radius,
                exp.p.fix_color)

    # Saccade targets
    targets = Points(exp.win,
                     exp.p.target_pos,
                     exp.p.target_radius,
                     exp.p.target_color)

    # Field of moving dots
    dots = RandomDotMotion(exp.win,
                           aperture=exp.p.aperture_size,
                           pos=exp.p.aperture_pos)

    return locals()


def generate_trials(exp):
    """Determine the parameters for each trial."""
    for _ in exp.trial_count(exp.p.n_trials):

        target = flexible_values(range(len(exp.p.dot_dir)))

        t_info = exp.trial_info(

            wait_iti=flexible_values(exp.p.wait_iti),
            wait_dots=flexible_values(exp.p.wait_dots),
            dot_coh=flexible_values(exp.p.dot_coh),
            dot_dir=exp.p.dot_dir[target],
            target=target,

        )

        yield t_info


def run_trial(exp, info):
    """Execute the events on a single trial."""
    # Inter-trial interval
    exp.wait_until(exp.iti_end, iti_duration=info.wait_iti)

    # Wait for trial onset
    res = exp.wait_until(AcquireFixation(exp),
                         timeout=exp.p.wait_fix,
                         draw="fix")

    if res is None:
        info["result"] = "nofix"
        exp.sounds.nofix.play()
        return info

    # Wait before showing the dots
    exp.wait_until(timeout=info.wait_dots, draw=["fix", "targets"])

    # Initialize a clock to get RT
    rt_clock = core.Clock()

    # Draw each frame of the stimulus
    for i in exp.frame_range(seconds=exp.p.wait_resp):

        # Displace the dots with specified coherent motion
        exp.s.dots.update(info.dot_dir, info.dot_coh)

        # Draw the dots on the screen
        exp.draw(["fix", "targets", "dots"])

        if not exp.check_fixation():

            # Use time of first fixation loss as rough estimate of RT
            rt = rt_clock.getTime()

            # Determine which target was chosen, if any
            res = exp.wait_until(AcquireTarget(exp, info.target),
                                 draw="targets")

            # End the dot epoch
            break

    else:
        # Dot stimulus timed out without broken fixation
        res = None

    # Handle the response
    if res is None:
        info["result"] = "fixbreak"
    else:
        info.update(pd.Series(res))

    # Inject the estimated RT into the results structure
    if info.responded:
        info["rt"] = rt

    # Give auditory and visual feedback
    exp.sounds[info.result].play()
    exp.show_feedback("targets", info.result, info.response)
    exp.wait_until(timeout=exp.p.wait_feedback, draw=["targets"])
    exp.s.targets.color = exp.p.target_color

    return info

Parameters module (params.py)

base = dict(

    # Key for display parameters (in displays.yaml)
    display_name="mlw-mbpro",

    # Brightness of the background
    display_luminance=0,

    # Use eye tracking
    monitor_eye=True,

    # Require fixation to initiate trial
    eye_fixation=True,

    # Collect responses based on eye position
    eye_response=True,

    # Position of the two saccade targets
    target_pos=[(-10, 5), (10, 5)],

    # Unique dot coherence values
    dot_coh=[0, .016, .032, .064, .128, .256, .512],

    # Unique dot direction values (0 is rightward)
    dot_dir=[180, 0],

    # Central position of the dot aperture
    aperture_pos=(0, 5),

    # Size (in degrees) of the dot aperture
    aperture_size=10,

    # Duration to wait before showing the dots
    wait_dots=("truncexpon", (.5 - .2) / .1, .2, .1),

    # (Minimal) duration to wait between trial
    wait_iti=2,

    # Total number of trials to perform
    n_trials=50,

    # Goal value for choice accuracy
    perform_acc_target=.8,

    # Goal value for reaction time
    perform_rt_target=1,

)

Remote module (remote.py)

import numpy as np
import pandas as pd
import matplotlib as mpl
from matplotlib.figure import Figure


def create_stim_artists(app):

    dots = mpl.patches.Circle(app.p.aperture_pos,
                              app.p.aperture_size / 2,
                              fc="firebrick", lw=0,
                              alpha=.5,
                              animated=True)

    return dict(dots=dots)


def initialize_trial_figure(app):

    # Note that we use mpl.figure.Figure, not pyplot.figure
    fig = Figure((5, 5), dpi=100, facecolor="white")
    axes = fig.subplots(3)

    axes[0].set(ylim=(-.1, 1.1),
                yticks=[0, 1],
                yticklabels=["No", "Yes"],
                ylabel="Responded")

    axes[1].set(ylim=(-.1, 1.1),
                yticks=[0, 1],
                yticklabels=["No", "Yes"],
                ylabel="Correct")

    axes[2].set(ylim=(0, None),
                xlabel="RT (s)")

    fig.subplots_adjust(.15, .125, .95, .95)

    return fig, axes


def update_trial_figure(app, trial_data):

    trial_data = pd.read_json(trial_data, typ="series")

    app.trial_data.append(trial_data)
    trial_df = pd.DataFrame(app.trial_data)

    resp_ax, cor_ax, rt_ax = app.axes

    # We are taking the approach of creating new artists on each trial,
    # drawing them, then removing them before adding the next trial's data.
    # Another approach would be to keep around references to the artists
    # and update their data using the appropriate matplotlib methods.

    resp_line, = resp_ax.plot(trial_df.trial, trial_df.responded, "ko")
    resp_ax.set(xlim=(.5, trial_df.trial.max() + .5))

    # Draw correct and incorrect responses, color by signed coherence
    dir_sign = dict(zip(app.remote_app.p.dot_dir, [-1, +1]))
    signed_coh = trial_df.dot_coh * trial_df.dot_dir.map(dir_sign)
    max_coh = max(app.remote_app.p.dot_coh)
    cor_line = cor_ax.scatter(
        trial_df.trial, trial_df.correct, c=signed_coh,
        vmin=-max_coh, vmax=+max_coh, cmap="coolwarm",
        linewidth=.5, edgecolor=".1",
    )
    cor_ax.set(xlim=(.5, trial_df.trial.max() + .5))

    # Draw a histogram of RTs
    bins = np.arange(0, 5.2, .2)
    heights, bins = np.histogram(trial_df.rt.dropna(), bins)
    rt_bars = rt_ax.bar(bins[:-1], heights, .2,
                        facecolor=".1", edgecolor="w", linewidth=.5)
    rt_ax.set(ylim=(0, heights.max() + 1))

    # Draw the canvas to show the new data
    app.fig_canvas.draw()

    # By removing the stimulus artists after drawing the canvas,
    # we are in effect clearing before drawing the new data on
    # the *next* trial.
    resp_line.remove()
    cor_line.remove()
    rt_bars.remove()