Skip to content

Instantly share code, notes, and snippets.

@papr
Last active September 29, 2022 15:10
Show Gist options
  • Save papr/bae0910a162edfd99d8ababaf09c643a to your computer and use it in GitHub Desktop.
Save papr/bae0910a162edfd99d8ababaf09c643a to your computer and use it in GitHub Desktop.

Generate a Pupil-Player-compatible recording from a single, externally recorded, video file.

Dependencies:

pip install -r requirements.txt

Usage

python generate_pupil_player_recording.py <path to video file>
import json
import logging
import pathlib
import time
import uuid
import av
import click
import numpy as np
@click.command()
@click.argument("input_file", type=click.Path(exists=True))
@click.option(
"-f", "--force", is_flag=True, help="Force overwriting of existing directory"
)
def main(input_file, force):
input_file = pathlib.Path(input_file)
if not force and _rec_dir_from_input_file(input_file).exists():
raise FileExistsError(
f"Target recording directory {_rec_dir_from_input_file(input_file)} "
"already exists"
)
create_recording_directory(input_file)
def create_recording_directory(input_file):
rec_dir = _rec_dir_from_input_file(input_file)
target_path = rec_dir / f"world{input_file.suffix.lower()}"
if click.confirm(
f"{input_file.name} will be moved to {target_path}. Continue?",
default=True,
abort=True,
):
logging.info(f"Creating recording direcotry at {rec_dir}")
rec_dir.mkdir(exist_ok=True)
logging.info(f"Moving {input_file.name} to {target_path}")
input_file.rename(target_path)
logging.info(f"Generating timestamps for {target_path.name}")
world_timestamps = extract_and_save_timestamps_with_offset(target_path)
logging.info("Generating info.player.json")
create_info_player_json(rec_dir, world_timestamps)
logging.info("Recording created successfully")
def extract_and_save_timestamps_with_offset(path_video, offset=0.0):
pts_timestamps = sorted(extract_pts_timestamps(path_video))
pts_timestamps = np.array(pts_timestamps)
pts = pts_timestamps[:, 0].astype(int)
timestamps = pts_timestamps[:, 1]
timestamps += offset
lookup_entry = np.dtype(
[
("container_idx", "<i8"),
("container_frame_idx", "<i8"),
("timestamp", "<f8"),
("pts", "<i8"),
]
)
lookup = np.empty(timestamps.size, dtype=lookup_entry).view(np.recarray)
lookup.timestamp = timestamps
lookup.container_idx = 0
lookup.container_frame_idx = np.arange(timestamps.size, dtype=int)
lookup.pts = pts
np.save(path_ts_from_video_path(path_video), timestamps)
np.save(path_lut_from_video_path(path_video), lookup)
return timestamps
def create_info_player_json(rec_dir, world_timestamps):
info_player_json = rec_dir / "info.player.json"
duration = world_timestamps[-1] - world_timestamps[0]
with info_player_json.open("w") as f:
json.dump(
{
"duration_s": duration,
"meta_version": "2.3",
"min_player_version": "2.0",
"recording_name": rec_dir.name,
"recording_software_name": "Pupil Capture",
"recording_software_version": "3.4.0",
"recording_uuid": str(uuid.uuid4()),
"start_time_synced_s": 0.0,
"start_time_system_s": time.time(),
"system_info": __file__,
},
f,
)
def extract_pts_timestamps(path_scene_video):
container = av.open(str(path_scene_video))
for packet in container.demux(video=0):
if packet.pts is None:
continue
yield packet.pts, float(packet.pts * packet.time_base)
def path_ts_from_video_path(path_video):
return path_video.parent / (path_video.stem + "_timestamps.npy")
def path_lut_from_video_path(path_video):
return path_video.parent / (path_video.stem + "_lookup.npy")
def _rec_dir_from_input_file(input_file):
return input_file.with_suffix("")
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
logging.getLogger("libav").setLevel(logging.ERROR)
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment