Pupil Labs Neon Recording¶
Functionality for loading Neon recordings in native recording format
Installation¶
pip install pupil-labs-neon-recording
or
pip install -e git+https://github.com/pupil-labs/pl-neon-recording.git
Documentation¶
Documentation is available at https://pupil-labs.github.io/pl-neon-recording/
Usage¶
Basic Usage¶
import sys
import pupil_labs.neon_recording as nr
if len(sys.argv) < 2:
print("Usage:")
print("python basic_usage.py path/to/recording/folder")
# Open a recording
recording = nr.open(sys.argv[1])
# get basic info
print("Recording Info:")
print(f"\tStart time (ns): {recording.start_time}")
print(f"\tWearer : {recording.wearer['name']}")
print(f"\tDevice serial : {recording.device_serial}")
print(f"\tGaze samples : {len(recording.gaze)}")
print("")
# read 10 gaze samples
print("First 10 gaze samples:")
timestamps = recording.gaze.time[:10]
subsample = recording.gaze.sample(timestamps)
for gaze_datum in subsample:
print(
f"\t{gaze_datum.time} :",
f"({gaze_datum.point[0]:0.2f}, {gaze_datum.point[1]:0.2f})",
)
recording.close()
Blinks And Fixations¶
import sys
import cv2
import numpy as np
from tqdm import tqdm
# Workaround for https://github.com/opencv/opencv/issues/21952
cv2.imshow("cv/av bug", np.zeros(1))
cv2.destroyAllWindows()
import pupil_labs.neon_recording as nr # noqa: E402
from pupil_labs.neon_recording import match_ts # noqa: E402
from pupil_labs.video import Writer # noqa: E402
def write_text(image, text, x, y):
return cv2.putText(
image,
text,
(x, y),
cv2.FONT_HERSHEY_SIMPLEX,
1,
(0, 0, 255),
2,
cv2.LINE_AA,
)
def match_events(target_time, events):
# Blink start needs to be <= target_time
matches = match_ts(target_time, events.start_time, method="backward")
# Blink end needs to be >= target_time
matches_end = match_ts(target_time, events.stop_time, method="forward")
matches[np.isnan(matches_end)] = np.nan
matches[matches != matches_end] = np.nan
return matches
def make_overlaid_video(recording_dir, output_video_path):
recording = nr.open(recording_dir)
video_writer = Writer(
output_video_path,
)
video_start_time = recording.eye.time[0]
blink_matches = match_events(recording.eye.time, recording.blinks)
fixation_matches = match_events(recording.eye.time, recording.fixations)
blink_count = 0
fixation_count = 0
for frame, blink_index, fixation_index in tqdm(
zip(recording.eye, blink_matches, fixation_matches, strict=False),
total=len(recording.eye),
):
frame_pixels = frame.bgr
if not np.isnan(blink_index):
blink_count = int(blink_index + 1)
frame_pixels = write_text(frame_pixels, f"Blinks: {blink_count}", 0, 40)
if not np.isnan(fixation_index):
fixation_count = int(fixation_index + 1)
frame_pixels = write_text(frame_pixels, f"Fixations: {fixation_count}", 0, 80)
video_time = (frame.time - video_start_time) / 1e9
video_writer.write_image(frame_pixels, time=video_time)
cv2.imshow("Frame", frame_pixels)
cv2.pollKey()
video_writer.close()
recording.close()
if __name__ == "__main__":
make_overlaid_video(sys.argv[1], "blinks-and-fixations.mp4")
CSV Export¶
import json
import sys
import time
from pathlib import Path
import cv2
import numpy as np
import pandas as pd
from scipy.spatial.transform import Rotation
import pupil_labs.neon_recording as nr
def unproject_points(points_2d, camera_matrix, distortion_coefs, normalize=False):
"""Undistorts points according to the camera model.
:param pts_2d, shape: Nx2
:return: Array of unprojected 3d points, shape: Nx3
"""
# Convert type to numpy arrays (OpenCV requirements)
camera_matrix = np.array(camera_matrix)
distortion_coefs = np.array(distortion_coefs)
points_2d = np.asarray(points_2d, dtype=np.float32)
# Add third dimension the way cv2 wants it
points_2d = points_2d.reshape((-1, 1, 2))
# Undistort 2d pixel coordinates
points_2d_undist = cv2.undistortPoints(points_2d, camera_matrix, distortion_coefs)
# Unproject 2d points into 3d directions; all points. have z=1
points_3d = cv2.convertPointsToHomogeneous(points_2d_undist)
points_3d.shape = -1, 3
if normalize:
# normalize vector length to 1
points_3d /= np.linalg.norm(points_3d, axis=1)[:, np.newaxis]
return points_3d
def cart_to_spherical(points_3d, apply_rad2deg=True):
points_3d = np.asarray(points_3d)
# convert cartesian to spherical coordinates
# source: http://stackoverflow.com/questions/4116658/faster-numpy-cartesian-to-spherical-coordinate-conversion
x = points_3d[:, 0]
y = points_3d[:, 1]
z = points_3d[:, 2]
radius = np.sqrt(x**2 + y**2 + z**2)
# elevation: vertical direction
# positive numbers point up
# negative numbers point bottom
elevation = np.arccos(y / radius) - np.pi / 2
# azimuth: horizontal direction
# positive numbers point right
# negative numbers point left
azimuth = np.pi / 2 - np.arctan2(z, x)
if apply_rad2deg:
elevation = np.rad2deg(elevation)
azimuth = np.rad2deg(azimuth)
return radius, elevation, azimuth
def find_ranged_index(values, left_boundaries, right_boundaries):
left_ids = np.searchsorted(left_boundaries, values, side="right") - 1
right_ids = np.searchsorted(right_boundaries, values, side="right")
return np.where(left_ids == right_ids, left_ids, -1)
def export_gaze(recording, export_path):
try:
fixations = recording.fixations
fixation_ids = (
find_ranged_index(
recording.gaze.time, fixations.start_time, fixations.stop_time
)
+ 1
)
except AttributeError:
fixation_ids = -1
try:
blink_ids = (
find_ranged_index(
recording.gaze.time,
recording.blinks.start_time,
recording.blinks.stop_time,
)
+ 1
)
except AttributeError:
blink_ids = -1
spherical_coords = cart_to_spherical(
unproject_points(
recording.gaze.point,
recording.calibration.scene_camera_matrix,
recording.calibration.scene_distortion_coefficients,
)
)
gaze = pd.DataFrame({
"recording id": recording.info["recording_id"],
"timestamp [ns]": recording.gaze.time,
"gaze x [px]": recording.gaze.point[:, 0],
"gaze y [px]": recording.gaze.point[:, 1],
"worn": recording.worn.worn,
"fixation id": fixation_ids,
"blink id": blink_ids,
"azimuth [deg]": spherical_coords[2],
"elevation [deg]": spherical_coords[1],
})
gaze["fixation id"] = gaze["fixation id"].replace(0, None)
gaze["blink id"] = gaze["blink id"].replace(0, None)
export_file = export_path / "gaze.csv"
gaze.to_csv(export_file, index=False)
print(f"Wrote {export_file}")
def export_blinks(recording, export_path):
blinks = pd.DataFrame({
"recording id": recording.info["recording_id"],
"blink id": 1 + np.arange(len(recording.blinks)),
"start timestamp [ns]": recording.blinks.start_time,
"end timestamp [ns]": recording.blinks.stop_time,
"duration [ms]": (recording.blinks.stop_time - recording.blinks.start_time)
/ 1e6,
})
export_file = export_path / "blinks.csv"
blinks.to_csv(export_file, index=False)
print(f"Wrote {export_file}")
def export_fixations(recording, export_path):
fixations = recording.fixations
spherical_coords = cart_to_spherical(
unproject_points(
fixations.mean_gaze_point,
recording.calibration.scene_camera_matrix,
recording.calibration.scene_distortion_coefficients,
)
)
fixations_df = pd.DataFrame({
"recording id": recording.info["recording_id"],
"fixation id": 1 + np.arange(len(fixations)),
"start timestamp [ns]": fixations.start_time,
"end timestamp [ns]": fixations.stop_time,
"duration [ms]": (fixations.stop_time - fixations.start_time) / 1e6,
"fixation x [px]": fixations.mean_gaze_point[:, 0],
"fixation y [px]": fixations.mean_gaze_point[:, 1],
"azimuth [deg]": spherical_coords[2],
"elevation [deg]": spherical_coords[1],
})
export_file = export_path / "fixations.csv"
fixations_df.to_csv(export_file, index=False)
print(f"Wrote {export_file}")
def export_saccades(recording, export_path):
saccades = recording.saccades
saccades_df = pd.DataFrame({
"recording id": recording.info["recording_id"],
"saccade id": 1 + np.arange(len(saccades)),
"start timestamp [ns]": saccades.start_time,
"end timestamp [ns]": saccades.stop_time,
"duration [ms]": (saccades.stop_time - saccades.start_time) / 1e6,
"amplitude [deg]": saccades.amplitude,
"mean velocity [px/s]": saccades.mean_velocity,
"peak velocity [px/s]": saccades.max_velocity,
})
export_file = export_path / "saccades.csv"
saccades_df.to_csv(export_file, index=False)
print(f"Wrote {export_file}")
def export_eyestates(recording, export_path):
eyeball = recording.eyeball
pupil = recording.pupil
eyelid = recording.eyelid
eyestates = pd.DataFrame({
"recording id": recording.info["recording_id"],
"timestamp [ns]": eyeball.time,
"pupil diameter left [mm]": pupil.diameter_left,
"pupil diameter right [mm]": pupil.diameter_right,
"eyeball center left x [mm]": eyeball.center_left[:, 0],
"eyeball center left y [mm]": eyeball.center_left[:, 1],
"eyeball center left z [mm]": eyeball.center_left[:, 2],
"eyeball center right x [mm]": eyeball.center_right[:, 0],
"eyeball center right y [mm]": eyeball.center_right[:, 1],
"eyeball center right z [mm]": eyeball.center_right[:, 2],
"optical axis left x": eyeball.optical_axis_left[:, 0],
"optical axis left y": eyeball.optical_axis_left[:, 1],
"optical axis left z": eyeball.optical_axis_left[:, 2],
"optical axis right x": eyeball.optical_axis_right[:, 0],
"optical axis right y": eyeball.optical_axis_right[:, 1],
"optical axis right z": eyeball.optical_axis_right[:, 2],
"eyelid angle top left [rad]": eyelid.angle_left[:, 0],
"eyelid angle bottom left [rad]": eyelid.angle_left[:, 1],
"eyelid aperture left [mm]": eyelid.aperture_left,
"eyelid angle top right [rad]": eyelid.angle_right[:, 0],
"eyelid angle bottom right [rad]": eyelid.angle_right[:, 1],
"eyelid aperture right [mm]": eyelid.aperture_right,
})
export_file = export_path / "3d_eye_states.csv"
eyestates.to_csv(export_file, index=False)
print(f"Wrote {export_file}")
def export_imu(recording, export_path):
rotations = Rotation.from_quat(recording.imu.rotation)
eulers = rotations.as_euler(seq="yxz", degrees=True)
imu = pd.DataFrame({
"recording id": recording.info["recording_id"],
"timestamp [ns]": recording.imu.time,
"gyro x [deg/s]": recording.imu.angular_velocity[:, 0],
"gyro y [deg/s]": recording.imu.angular_velocity[:, 1],
"gyro z [deg/s]": recording.imu.angular_velocity[:, 2],
"acceleration x [g]": recording.imu.acceleration[:, 0],
"acceleration y [g]": recording.imu.acceleration[:, 1],
"acceleration z [g]": recording.imu.acceleration[:, 2],
"roll [deg]": eulers[:, 0],
"pitch [deg]": eulers[:, 1],
"yaw [deg]": eulers[:, 2],
"quaternion w": recording.imu.rotation[:, 3],
"quaternion x": recording.imu.rotation[:, 0],
"quaternion y": recording.imu.rotation[:, 1],
"quaternion z": recording.imu.rotation[:, 2],
})
export_file = export_path / "imu.csv"
imu.to_csv(export_file, index=False)
print(f"Wrote {export_file}")
def export_events(recording, export_path):
events = pd.DataFrame({
"recording id": recording.info["recording_id"],
"timestamp [ns]": recording.events.time,
"name": recording.events.event,
"type": "recording",
})
export_file = export_path / "events.csv"
events.to_csv(export_file, index=False)
print(f"Wrote {export_file}")
def export_info(recording, export_path):
with (export_path / "info.json").open("w") as f:
json.dump(recording.info, f, indent=4, sort_keys=True)
def export_scene_camera_calibration(recording, export_path):
distortion = recording.calibration.scene_distortion_coefficients.reshape([1, -1])
camera_info = {
"camera_matrix": recording.calibration.scene_camera_matrix.tolist(),
"distortion_coefficients": distortion.tolist(),
"serial_number": recording.calibration.serial,
}
with (export_path / "scene_camera.json").open("w") as f:
json.dump(camera_info, f, indent=4, sort_keys=True)
def export_world_timestamps(recording, export_path):
events = pd.DataFrame({
"recording id": recording.info["recording_id"],
"timestamp [ns]": recording.scene.time,
})
export_file = export_path / "world_timestamps.csv"
events.to_csv(export_file, index=False)
print(f"Wrote {export_file}")
if __name__ == "__main__":
func_map = {
"gaze": export_gaze,
"blinks": export_blinks,
"fixations": export_fixations,
"saccades": export_saccades,
"eyestates": export_eyestates,
"imu": export_imu,
"events": export_events,
"info": export_info,
"scene-camera": export_scene_camera_calibration,
"world-timestamps": export_world_timestamps,
}
if len(sys.argv) < 2 or "--help" in sys.argv:
arg_list = "[--help] [--all] "
arg_list += " ".join([f"[--{s}]" for s in func_map])
print(f"Usage: python csv_export.py path/to/recording/folder {arg_list}")
sys.exit(0)
recording_path = Path(sys.argv[1])
recording = nr.open(recording_path)
timestamp = time.strftime("%Y-%m-%d_%H-%M-%S")
safe_timestamp = timestamp.replace(":", "-").replace(" ", "_")
export_path = recording_path / "exports" / safe_timestamp
export_path.mkdir(parents=True, exist_ok=True)
if "--all" in sys.argv or len(sys.argv) == 2:
sys.argv += [f"--{k}" for k in func_map]
for stream_name, export_func in func_map.items():
if f"--{stream_name}" in sys.argv:
try:
export_func(recording, export_path)
except AttributeError as err:
print(f"Could not export {stream_name}: {err}", file=sys.stderr)
recording.close()
Data Access¶
import sys
from collections.abc import Mapping
from datetime import datetime
import numpy as np
import pupil_labs.neon_recording as nr
from pupil_labs.neon_recording.timeseries.timeseries import Timeseries
if len(sys.argv) < 2:
print("Usage:")
print("python basic_usage.py path/to/recording/folder")
# Open a recording
recording = nr.open(sys.argv[1])
def pretty_format(mapping: Mapping):
output = []
pad = " "
keys = mapping.keys()
n = max(len(key) for key in keys)
for k, v in mapping.items():
v_repr_lines = str(v).splitlines()
output.append(f"{pad}{k:>{n}}: {v_repr_lines[0]}")
if len(v_repr_lines) > 1:
output.extend(f"{pad + ' '}{n * ' '}{line}" for line in v_repr_lines[1:])
return "\n".join(output)
print("Basic Recording Info:")
print_data = {
"Recording ID": recording.id,
"Start time (ns since unix epoch)": f"{recording.start_time}",
"Start time (datetime)": f"{datetime.fromtimestamp(recording.start_time / 1e9)}",
"Duration (nanoseconds)": f"{recording.duration}",
"Duration (seconds)": f"{recording.duration / 1e9}",
"Wearer": f"{recording.wearer['name']} ({recording.wearer['uuid']})",
"Device serial": recording.device_serial,
"App version": recording.info["app_version"],
"Data format": recording.info["data_format_version"],
"Gaze Offset": recording.info["gaze_offset"],
}
print(pretty_format(print_data))
streams: list[Timeseries] = [
recording.gaze,
recording.imu,
recording.eyeball,
recording.pupil,
recording.eyelid,
recording.blinks,
recording.fixations,
recording.saccades,
recording.worn,
recording.eye,
recording.scene,
recording.audio,
]
print()
print("Recording streams:")
print(
pretty_format({
f"{stream.name} ({len(stream)} samples)": "\n" + pretty_format(stream[0])
for stream in streams
})
)
print()
print("Getting data from a stream:")
print()
print("Gaze points", recording.gaze.point)
# GazeArray([(1741948698620648018, 966.3677 , 439.58817),
# (1741948698630654018, 965.9669 , 441.60403),
# (1741948698635648018, 964.2665 , 442.4974 ), ...,
# (1741948717448190018, 757.85815, 852.34644),
# (1741948717453190018, 766.53174, 857.3709 ),
# (1741948717458190018, 730.93604, 851.53723)],
# dtype=[('ts', '<i8'), ('x', '<f4'), ('y', '<f4')])
print()
print("Gaze timestamps", recording.gaze.time)
# array([1741948698620648018, 1741948698630654018, 1741948698635648018, ...,
# 1741948717448190018, 1741948717453190018, 1741948717458190018])
# All stream data can also be accesses as structured numpy arrays and pandas dataframes.
print()
print("Gaze data as a structured numpy array:")
gaze_np = recording.gaze.data
print()
print(gaze_np)
print()
print("Gaze data as pandas dataframe:")
gaze_df = recording.gaze.pd
print()
print(gaze_df)
print()
print("Sampling data:")
print()
print("Get closest gaze for scene frames")
closest_gaze_to_scene = recording.gaze.sample(recording.scene.time)
print(closest_gaze_to_scene)
print(
"closest_gaze_to_scene_times",
(closest_gaze_to_scene.time - recording.start_time) / 1e9,
)
print()
print("Sampled data can be resampled")
print()
print("Closest gaze sampled at 1 fps")
closest_gaze_to_scene_at_one_fps = closest_gaze_to_scene.sample(
np.arange(
closest_gaze_to_scene.time[0],
closest_gaze_to_scene.time[-1],
1e9 / 1,
dtype=np.int64,
),
)
print(closest_gaze_to_scene_at_one_fps)
print(
"closest_gaze_to_scene_at_one_fps_times",
(closest_gaze_to_scene_at_one_fps.time - recording.start_time) / 1e9,
)
# Close the recording when done to free up resources
recording.close()
Eye Overlay¶
import sys
import cv2
import numpy as np
from tqdm import tqdm
# Workaround for https://github.com/opencv/opencv/issues/21952
cv2.imshow("cv/av bug", np.zeros(1))
cv2.destroyAllWindows()
import pupil_labs.neon_recording as nr # noqa: E402
from pupil_labs.neon_recording.timeseries.av.video import ( # noqa: E402
GrayFrame,
)
def overlay_image(img, img_overlay, x, y):
"""Overlay `img_overlay` onto `img` at (x, y)."""
# Image ranges
y1, y2 = max(0, y), min(img.shape[0], y + img_overlay.shape[0])
x1, x2 = max(0, x), min(img.shape[1], x + img_overlay.shape[1])
# Overlay ranges
y1o, y2o = max(0, -y), min(img_overlay.shape[0], img.shape[0] - y)
x1o, x2o = max(0, -x), min(img_overlay.shape[1], img.shape[1] - x)
if y1 >= y2 or x1 >= x2 or y1o >= y2o or x1o >= x2o:
return
img_crop = img[y1:y2, x1:x2]
img_overlay_crop = img_overlay[y1o:y2o, x1o:x2o]
img_crop[:] = img_overlay_crop
def make_overlaid_video(recording_dir, output_video_path, fps=30):
recording = nr.open(recording_dir)
video_writer = cv2.VideoWriter(
str(output_video_path),
cv2.VideoWriter_fourcc(*"MJPG"),
fps,
(recording.scene.width, recording.scene.height),
)
output_timestamps = np.arange(
recording.scene.time[0], recording.scene.time[-1], int(1e9 / fps)
)
combined_data = zip(
output_timestamps,
recording.scene.sample(output_timestamps),
recording.eye.sample(output_timestamps),
strict=False,
)
frame_idx = 0
for ts, scene_frame, eye_frame in tqdm(combined_data, total=len(output_timestamps)):
frame_idx += 1 # noqa: SIM113
if abs(scene_frame.time - ts) < 2e9 / fps:
frame_pixels = scene_frame.bgr
else:
# if the video frame timestamp is too far ahead or behind temporally,
# replace it with a gray frame
frame_pixels = GrayFrame(scene_frame.width, scene_frame.height).bgr
if abs(eye_frame.time - ts) < 2e9 / fps:
eye_pixels = cv2.cvtColor(eye_frame.gray, cv2.COLOR_GRAY2BGR)
else:
# if the video frame timestamp is too far ahead or behind temporally,
# replace it with a gray frame
eye_pixels = GrayFrame(eye_frame.width, eye_frame.height).bgr
overlay_image(frame_pixels, eye_pixels, 50, 50)
video_writer.write(frame_pixels)
cv2.imshow("Frame", frame_pixels)
cv2.pollKey()
video_writer.release()
recording.close()
if __name__ == "__main__":
make_overlaid_video(sys.argv[1], "eye-overlay-output-video.avi")
Eye State¶
import sys
import cv2
import numpy as np
from tqdm import tqdm
# Workaround for https://github.com/opencv/opencv/issues/21952
cv2.imshow("cv/av bug", np.zeros(1))
cv2.destroyAllWindows()
import pupil_labs.neon_recording as nr # noqa: E402
from pupil_labs.video import Writer # noqa: E402
def overlay_image(img, img_overlay, x, y):
"""Overlay `img_overlay` onto `img` at (x, y)."""
# Image ranges
y1, y2 = max(0, y), min(img.shape[0], y + img_overlay.shape[0])
x1, x2 = max(0, x), min(img.shape[1], x + img_overlay.shape[1])
# Overlay ranges
y1o, y2o = max(0, -y), min(img_overlay.shape[0], img.shape[0] - y)
x1o, x2o = max(0, -x), min(img_overlay.shape[1], img.shape[1] - x)
if y1 >= y2 or x1 >= x2 or y1o >= y2o or x1o >= x2o:
return
img_crop = img[y1:y2, x1:x2]
img_overlay_crop = img_overlay[y1o:y2o, x1o:x2o]
img_crop[:] = img_overlay_crop
def plot(img, data, value_range, x_width, color, line_width=2):
for idx in range(1, len(data)):
x_values = [int(idx2 * x_width) for idx2 in [idx - 1, idx]]
y_norms = [
(data[idx2] - value_range[0]) / (value_range[1] - value_range[0])
for idx2 in [idx - 1, idx]
]
y_values = [int(y_norm * img.shape[0]) for y_norm in y_norms]
points = [[*v] for v in zip(x_values, y_values, strict=False)]
cv2.line(img, points[0], points[1], color, line_width)
def make_eye_state_video(recording_dir, output_video_path):
recording = nr.open(recording_dir)
fps = 200
video_writer = Writer(output_video_path)
video_start_time = recording.eye.time[0]
plot_config = [
{"color": [0, 0, 255]},
{"color": [0, 255, 0]},
{"color": [255, 0, 0]},
]
for dim, config in enumerate(plot_config):
config["range"] = (
np.min(recording.eyeball.optical_axis_left[:, dim]),
np.max(recording.eyeball.optical_axis_left[:, dim]),
)
plot_duration_secs = 0.5
plot_point_count = plot_duration_secs * fps
plot_x_width = recording.eye.width / plot_point_count
for eye_sample in tqdm(recording.eye):
eye_pixels = eye_sample.bgr
for dim, config in enumerate(plot_config):
min_ts = eye_sample.time - plot_duration_secs * 1e9
mask = (min_ts < recording.eyeball.time) & (
recording.eyeball.time <= eye_sample.time
)
plot_data = recording.eyeball.optical_axis_left[mask, dim]
plot(
eye_pixels,
plot_data,
config["range"],
plot_x_width,
config["color"],
)
video_time = (eye_sample.time - video_start_time) / 1e9
video_writer.write_image(eye_pixels, time=video_time)
cv2.imshow("Frame", eye_pixels)
cv2.pollKey()
video_writer.close()
recording.close()
if __name__ == "__main__":
make_eye_state_video(sys.argv[1], "eye-state-output-video.avi")
Find Clap¶
import sys
import numpy as np
from tqdm import tqdm
import pupil_labs.neon_recording as nr
def find_clap(recording_dir, window_size_seconds=0.1, first_n_seconds=10):
recording = nr.open(recording_dir)
audio_data = np.array([], dtype=np.float32)
ts_lookup = []
for frame in recording.audio:
if first_n_seconds is not None:
rel_time = (frame.time - recording.start_time) / 1e9
if rel_time > first_n_seconds:
break
# Create a timestamp lookup table
ts_lookup.append([len(audio_data), frame.time])
# Gather all the audio samples
audio_data = np.concatenate((audio_data, frame.to_ndarray().flatten()))
ts_lookup = np.array(ts_lookup)
# Calculate RMS over a sliding window
# Remember the sample index of the loudest window
max_rms = 0
loudest_sample_idx = 0
samples_per_window = int(window_size_seconds * recording.audio.rate)
for i in tqdm(range(len(audio_data) - samples_per_window)):
segment = audio_data[i : i + samples_per_window]
rms = np.sqrt(np.mean(np.square(segment)))
if rms > max_rms:
max_rms = rms
loudest_sample_idx = int(i + samples_per_window / 2)
# Find the reference timestamp from the lookup table
lookup_idx = np.searchsorted(ts_lookup[:, 0], loudest_sample_idx) - 1
reference_ts = ts_lookup[lookup_idx, 1]
# Calculate the sample timestamp using the reference timestamp
samples_after_reference = loudest_sample_idx - ts_lookup[lookup_idx, 0]
loudest_time = reference_ts + (samples_after_reference / recording.audio.rate) * 1e9
print(f"The loudest audio occurs at {loudest_time:.0f} rms = {max_rms:.3f}.")
print(
f" Relative to recording start: "
f"{(loudest_time - recording.start_time) / 1e9:0.3f}s"
)
print(
f" Relative to video start : "
f"{(loudest_time - recording.scene.time[0]) / 1e9:0.3f}s"
)
print(
f" Relative to audio start : "
f"{(loudest_time - recording.audio.time[0]) / 1e9:0.3f}s"
)
recording.close()
if __name__ == "__main__":
find_clap(sys.argv[1])
Gaze Overlay¶
import sys
import cv2
import numpy as np
from tqdm import tqdm
# Workaround for https://github.com/opencv/opencv/issues/21952
cv2.imshow("cv/av bug", np.zeros(1))
cv2.destroyAllWindows()
import pupil_labs.neon_recording as nr # noqa: E402
from pupil_labs.video import Writer # noqa: E402
def make_overlaid_video(recording_dir, output_video_path):
rec = nr.open(recording_dir)
combined_data = zip(
rec.scene,
rec.gaze.sample(rec.scene.time),
strict=True,
)
video_writer = Writer(output_video_path)
video_start_time = rec.scene.time[0]
for scene_frame, gaze_datum in tqdm(combined_data, total=len(rec.scene.time)):
frame_pixels = scene_frame.bgr
frame_pixels = cv2.circle(
frame_pixels,
(int(gaze_datum.point[0]), int(gaze_datum.point[1])),
50,
(0, 0, 255),
10,
)
video_time = (scene_frame.time - video_start_time) / 1e9
video_writer.write_image(frame_pixels, time=video_time)
cv2.imshow("Frame", frame_pixels)
cv2.waitKey(30)
video_writer.close()
rec.close()
if __name__ == "__main__":
make_overlaid_video(sys.argv[1], "gaze-overlay-output-video.mp4")
IMU¶
import sys
import cv2
import numpy as np
from scipy.spatial.transform import Rotation
from tqdm import tqdm
# Workaround for https://github.com/opencv/opencv/issues/21952
cv2.imshow("cv/av bug", np.zeros(1))
cv2.destroyAllWindows()
import pupil_labs.neon_recording as nr # noqa: E402
if len(sys.argv) < 2:
print("Usage:")
print("python imu.py path/to/recording/folder")
# Open a recording
recording = nr.open(sys.argv[1])
# Sample the IMU data at 60Hz
fps = 60
z = recording.gaze[:10]
timestamps = np.arange(
recording.imu.time[0], recording.imu.time[-1], 1e9 / fps, dtype=np.int64
)
imu_data = recording.imu.sample(timestamps)
# Use scipy to convert the quaternions to euler angles
quaternions = np.array([s.rotation for s in imu_data])
rotations = Rotation.from_quat(quaternions).as_euler(seq="yxz", degrees=True) % 360
# Combine the timestamps and eulers
rotations_with_time = np.column_stack((timestamps, rotations))
timestamped_eulers = np.array(
[tuple(row) for row in rotations_with_time],
dtype=[
("time", np.int64),
("roll", np.float64),
("pitch", np.float64),
("yaw", np.float64),
],
)
# Display the angles
frame_size = 512
colors = {"pitch": (0, 0, 255), "yaw": (0, 255, 0), "roll": (255, 0, 0)}
for row in tqdm(timestamped_eulers):
# Create a blank image
frame = np.zeros((frame_size, frame_size, 3), dtype=np.uint8)
# Define the center and radius of the circles
center = [frame_size // 2] * 2
radius = frame.shape[0] // 3
# Calculate the end points for the angles
for field, color in colors.items():
pitch_end = (
int(center[0] + radius * np.cos(np.deg2rad(row[field]))),
int(center[1] - radius * np.sin(np.deg2rad(row[field]))),
)
cv2.line(frame, center, pitch_end, color, 2)
# Write the angle values on the image
cv2.putText(
frame,
f"{field}: {row[field]:.2f}",
(10, 30 + list(colors.keys()).index(field) * 30),
cv2.FONT_HERSHEY_SIMPLEX,
0.7,
color,
2,
)
# Display the image
cv2.imshow("IMU Angles", frame)
if cv2.waitKey(1000 // fps) == 27:
break
cv2.destroyAllWindows()
recording.close()
Interpolate Gaze Data¶
import sys
import cv2
import numpy as np
from tqdm import tqdm
# Workaround for https://github.com/opencv/opencv/issues/21952
cv2.imshow("cv/av bug", np.zeros(1))
cv2.destroyAllWindows()
import pupil_labs.neon_recording as nr # noqa: E402
from pupil_labs.video import Writer # noqa: E402
def make_overlaid_video(recording_dir, output_video_path, fps=None):
# Open a recording
recording = nr.open(recording_dir)
video_writer = Writer(output_video_path)
video_start_time = recording.scene.time[0]
# get closest gaze data to scene frame timestamps
matched_gazes = recording.gaze.sample(recording.scene.time)
# interpolate gaze data to scene frame timestamps
interpolated_gazes = recording.gaze.interpolate(recording.scene.time)
# visualize both
scene_gaze_pairs = zip(
recording.scene, matched_gazes, interpolated_gazes, strict=False
)
for scene_frame, matched_gaze, interpolated_gaze in tqdm(
scene_gaze_pairs, total=len(recording.scene)
):
# draw the nearest-time gaze sample in red
frame = cv2.circle(
scene_frame.bgr,
(int(matched_gaze.point[0]), int(matched_gaze.point[1])),
50,
(0, 0, 255),
10,
)
# draw the interpolated gaze sample in blue
if not np.isnan(interpolated_gaze.point[0]) and not np.isnan(
interpolated_gaze.point[1]
):
frame = cv2.circle(
frame,
(int(interpolated_gaze.point[0]), int(interpolated_gaze.point[1])),
50,
(255, 0, 0),
10,
)
video_time = (scene_frame.time - video_start_time) / 1e9
video_writer.write_image(frame, time=video_time)
cv2.imshow("Gaze sample comparison", frame)
cv2.pollKey()
cv2.destroyAllWindows()
video_writer.close()
recording.close()
if __name__ == "__main__":
make_overlaid_video(sys.argv[1], "interpolation-compare.mp4", 30)
Multi-Gaze Overlay¶
import sys
import cv2
import numpy as np
from tqdm import tqdm
# Workaround for https://github.com/opencv/opencv/issues/21952
cv2.imshow("cv/av bug", np.zeros(1))
cv2.destroyAllWindows()
import pupil_labs.neon_recording as nr # noqa: E402
from pupil_labs.video import Writer # noqa: E402
def make_overlaid_video(recording_dir, output_video_path):
rec = nr.open(recording_dir)
combined_data = zip(
rec.scene,
rec.gaze.sample(rec.scene.time),
rec.gaze_monocular_left.sample(rec.scene.time),
rec.gaze_monocular_right.sample(rec.scene.time),
strict=True,
)
video_writer = Writer(output_video_path)
video_start_time = rec.scene.time[0]
gaze_colors = [
(0, 255, 0),
(255, 0, 0),
(0, 0, 255),
]
for scene_frame, left_gaze, right_gaze, binocular_gaze in tqdm(
combined_data, total=len(rec.scene.time)
):
# Prepare the next frame
frame_pixels = scene_frame.bgr
for gaze_datum, color in zip(
[binocular_gaze, left_gaze, right_gaze], gaze_colors
):
frame_pixels = cv2.circle(
frame_pixels,
(int(gaze_datum.point[0]), int(gaze_datum.point[1])),
50,
color,
10,
)
video_time = (scene_frame.time - video_start_time) / 1e9
video_writer.write_image(frame_pixels, time=video_time)
# Render current frame and mark start time
cv2.imshow("Frame", frame_pixels)
cv2.waitKey(30)
video_writer.close()
rec.close()
if __name__ == "__main__":
make_overlaid_video(sys.argv[1], "gaze-overlay-output-video.mp4")
Worn¶
import sys
import cv2
import numpy as np
from tqdm import tqdm
# Workaround for https://github.com/opencv/opencv/issues/21952
cv2.imshow("cv/av bug", np.zeros(1))
cv2.destroyAllWindows()
from pupil_labs import neon_recording as nr # noqa: E402
def write_text(image, text, x, y):
return cv2.putText(
image,
text,
(x, y),
cv2.FONT_HERSHEY_SIMPLEX,
1,
(0, 0, 255),
2,
cv2.LINE_AA,
)
def make_overlaid_video(recording_dir, output_video_path, fps=30):
recording = nr.open(recording_dir)
video_writer = cv2.VideoWriter(
str(output_video_path),
cv2.VideoWriter_fourcc(*"MJPG"),
fps,
(recording.eye.width, recording.eye.height),
)
output_timestamps = np.arange(
recording.eye.time[0], recording.eye.time[-1], int(1e9 / fps)
)
eyes_and_worn = zip(
recording.eye.sample(output_timestamps),
recording.worn.sample(output_timestamps),
strict=False,
)
for frame, worn_record in tqdm(eyes_and_worn, total=len(output_timestamps)):
frame_pixels = frame.bgr
text_y = 40
if worn_record.worn:
frame_pixels = write_text(frame_pixels, "Worn", 0, text_y)
video_writer.write(frame_pixels)
cv2.imshow("Frame", frame_pixels)
cv2.pollKey()
video_writer.release()
recording.close()
if __name__ == "__main__":
make_overlaid_video(sys.argv[1], "worn.mp4")