calibration_during_gaze_recording.py

"""
This example is more comprehensive than most, as it aims to illustrate
how you can put together a complete, ready-to-run script, by combining
information from different examples. Note that running this script
will produce a small CSV data file in the parent directory.
"""
import csv
import logging
import time
import sys

import tobii_research as tr

# <BeginExample>
# Enter the address to the desired eye tracker if you have
# more than one (otherwise leave this as an empty string).
# (example address: "tobii-prp://TPFC2-100123456789")
EYETRACKER_ADDRESS = ""

# This list will be filled with dictionaries, with one dictionary per
# recording sample.
gaze_data_samples = []


def get_eyetracker():
    # Check if a specific eye tracker address has been provided, and if so,
    # try to locate it and return the corresponding eye tracker object.
    if EYETRACKER_ADDRESS:
        eyetracker = tr.EyeTracker(EYETRACKER_ADDRESS)
        if not eyetracker:
            sys.exit("Specified eye tracker not found, please check the address.")
        return eyetracker
    # If we reach this point, no specific address was provided, so return
    # the first found eye tracker.
    all_eyetrackers = tr.find_all_eyetrackers()
    if not all_eyetrackers:
        sys.exit(
            "No connected eye trackers found. Please check the connection "
            "and/or install any missing drivers with Tobii Pro Eye Tracker Manager."
        )
    return all_eyetrackers[0]


def gaze_data_callback(gaze_data):
    # Note that no data processing or saving to file is done here,
    # as stream callbacks should normally be kept minimal (to ensure
    # that the system is able to keep up with the eye tracker)
    # unless you need to e.g. provide real-time feedback.
    global gaze_data_samples
    gaze_data_samples.append(gaze_data)


def buffer_overflow_notifications_callback(notification_data):
    logging.error(
        f"Buffer overflow occurred at system time: {notification_data['system_time_stamp']}"
    )


def calibrate(eyetracker):
    calibration = tr.ScreenBasedCalibration(eyetracker)

    # Enter calibration mode.
    calibration.enter_calibration_mode()
    print("Entered calibration mode for eye tracker with serial number {0}.".format(eyetracker.serial_number))

    # Define the points on screen we should calibrate at.
    # The coordinates are normalized, i.e. (0.0, 0.0) is the upper left corner and (1.0, 1.0) is the lower right corner.
    points_to_calibrate = [(0.5, 0.5), (0.1, 0.1), (0.1, 0.9), (0.9, 0.1), (0.9, 0.9)]

    for point in points_to_calibrate:
        # Normally, you would use a Python package like pygame, PsychoPy or PIL to present an image
        # indicating the calibration point here.
        print("Show a point on screen at {0}.".format(point))

        # Wait a little for user to focus.
        time.sleep(0.7)

        print("Collecting data at {0}.".format(point))
        if calibration.collect_data(point[0], point[1]) != tr.CALIBRATION_STATUS_SUCCESS:
            # Try again if it didn't go well the first time.
            # Not all eye tracker models will fail at this point, but instead fail on ComputeAndApply.
            calibration.collect_data(point[0], point[1])

    print("Computing and applying calibration.")
    calibration_result = calibration.compute_and_apply()
    print("Compute and apply returned {0} and collected at {1} points.".
          format(calibration_result.status, len(calibration_result.calibration_points)))

    # Analyze the data and maybe remove points that weren't good.
    recalibrate_point = (0.1, 0.1)
    print("Removing calibration point at {0}.".format(recalibrate_point))
    calibration.discard_data(recalibrate_point[0], recalibrate_point[1])

    # Redo collection at the discarded point
    print("Show a point on screen at {0}.".format(recalibrate_point))
    calibration.collect_data(recalibrate_point[0], recalibrate_point[1])

    # Compute and apply again.
    print("Computing and applying calibration.")
    calibration_result = calibration.compute_and_apply()
    print("Compute and apply returned {0} and collected at {1} points.".
          format(calibration_result.status, len(calibration_result.calibration_points)))

    # See that you're happy with the result.

    # The calibration is done. Leave calibration mode.
    calibration.leave_calibration_mode()

    print("Left calibration mode.")


def save_gaze_data(gaze_samples_list):
    if not gaze_samples_list:
        print("No gaze samples were collected. Skipping saving")
        return
    # To show what kinds of data are available in each sample's dictionary,
    # we print the available keys here.
    print("Sample dictionary keys:", gaze_samples_list[0].keys())
    # This is meant to serve as a simple example of how you can save
    # some of the gaze data - check the keys to see what else is available.
    file_handle = open("my_gaze_data.csv", "w")
    gaze_writer = csv.writer(file_handle)
    gaze_writer.writerow(["time_seconds", "left_x", "left_y", "right_x", "right_y"])
    start_time = gaze_samples_list[0]["system_time_stamp"]
    for recording_dict in gaze_samples_list:
        sample_time_from_start = recording_dict["system_time_stamp"] - start_time
        # convert from microseconds to seconds
        sample_time_from_start = sample_time_from_start / (10**(6))
        # x is horizontal coordinate on the screen
        # y is vertical coordinate on the screen
        left_x, left_y = recording_dict["left_gaze_point_on_display_area"]
        right_x, right_y = recording_dict["right_gaze_point_on_display_area"]
        gaze_writer.writerow(
            [sample_time_from_start, left_x, left_y, right_x, right_y]
        )
    file_handle.close()


def main():
    eyetracker = get_eyetracker()
    # Start gaze recording.
    print("Subscribing to gaze data for eye tracker with serial number {0}.".format(eyetracker.serial_number))
    eyetracker.subscribe_to(tr.EYETRACKER_GAZE_DATA, gaze_data_callback, as_dictionary=True)
    # listen for buffer overflow notifications (to see e.g. if the gaze callback is too slow)
    eyetracker.subscribe_to(
        tr.EYETRACKER_NOTIFICATION_STREAM_BUFFER_OVERFLOW,
        buffer_overflow_notifications_callback,
        as_dictionary=True
    )
    calibrate(eyetracker)
    # End gaze recording.
    print("Unsubscribing from gaze data for eye tracker with serial number {0}.".format(eyetracker.serial_number))
    eyetracker.unsubscribe_from(tr.EYETRACKER_GAZE_DATA, gaze_data_callback)
    # Save the data, now that all recording is finished.
    save_gaze_data(gaze_data_samples)

# <EndExample>


# This is a Python convention used with scripts that are to be run directly
# (rather than only being used by other scripts).
if __name__ == "__main__":
    main()