path: root/src/argaze/TobiiGlassesPro2/TobiiEntities.py

#!/usr/bin/env python

import datetime
import json
import gzip
import os

from argaze import *

import cv2 as cv

TOBII_DATETIME_FORMAT = '%Y-%m-%dT%H:%M:%S+%f'

TOBII_PROJECTS_DIRNAME = "projects"
TOBII_PROJECT_FILENAME = "project.json"

TOBII_PARTICIPANTS_DIRNAME = "participants"
TOBII_PARTICIPANT_FILENAME = "participant.json"

TOBII_RECORDINGS_DIRNAME = "recordings"
TOBII_RECORD_FILENAME = "recording.json"

TOBII_SEGMENTS_DIRNAME = "segments"
TOBII_SEGMENT_INFO_FILENAME = "segment.json"
TOBII_SEGMENT_VIDEO_FILENAME = "fullstream.mp4"
TOBII_SEGMENT_DATA_FILENAME = "livedata.json.gz"

class TobiiSegmentData:
    """Handle Tobii Glasses Pro 2 segment data file."""

    def __init__(self, segment_data_path):
        """Load segment data from segment directory."""

        self.__segment_data_path = segment_data_path
        self.__ts_start = 0

    def get_path(self):
        return self.__segment_data_path

    def load(self):

        ts_data_buffer_dict = {}

        # define a decoder function
        def decode(json_item):

            # accept only valid data (e.g. with status value equal to 0)
            if json_item.pop('s', -1) == 0:

                # convert timestamp into ms
                ts = json_item.pop('ts') / 1000.0

                # keep first timestamp to offset all timestamps
                if self.__ts_start == 0:
                    self.__ts_start = ts

                ts -= self.__ts_start

                # ignore negative timestamp
                if ts < 0:
                    return

                # convert json data into data object
                data_object_type = '-'.join(json_item.keys())
                data_object = DataStructures.DictObject(data_object_type, **json_item)

                # append a dedicated timestamped buffer for each data object type
                if data_object.type() not in ts_data_buffer_dict.keys():
                    ts_data_buffer_dict[data_object.type()] = DataStructures.TimeStampedBuffer()

                # store data object into the timestamped buffer dedicated to its type
                ts_data_buffer_dict[data_object.type()][ts] = data_object

        # start loading
        with gzip.open(self.__segment_data_path) as f:

            for item in f:
                json.loads(item.decode('utf-8'), object_hook=decode)

        return ts_data_buffer_dict

class TobiiSegmentVideo:
    """Handle Tobii Glasses Pro 2 segment video file."""

    def __init__(self, segment_video_path):
        """Load segment video from segment directory."""

        self.__segment_video_path = segment_video_path
        
        video = cv.VideoCapture(self.__segment_video_path)

        self.__width = int(video.get(cv.CAP_PROP_FRAME_WIDTH))
        self.__height = int(video.get(cv.CAP_PROP_FRAME_HEIGHT))
        self.__fps = int(video.get(cv.CAP_PROP_FPS))

    def get_path(self):
        return self.__segment_video_path

    def get_height(self):
        return self.__height

    def get_width(self):
        return self.__width

    def get_fps(self):
        return self.__fps

class TobiiSegment:
    """Handle Tobii Glasses Pro 2 segment info."""

    def __init__(self, segment_path):
        """Load segment info from segment directory."""

        self.__segment_id = os.path.basename(segment_path)
        self.__segment_path = segment_path
        
        with open(os.path.join(self.__segment_path, TOBII_SEGMENT_INFO_FILENAME)) as f:
            try:
                item = json.load(f)
            except:
                raise RuntimeError(f'JSON fails to load {self.__segment_path}/{TOBII_SEGMENT_INFO_FILENAME}')

        self.__length_us = int(item["seg_length_us"])
        self.__calibrated = bool(item["seg_calibrated"])
        self.__start_date = datetime.datetime.strptime(item["seg_t_start"], TOBII_DATETIME_FORMAT)
        self.__stop_date = datetime.datetime.strptime(item["seg_t_stop"], TOBII_DATETIME_FORMAT)

    def get_path(self):
        return self.__segment_path

    def get_id(self):
        return self.__segment_id

    def get_length_us(self):
        return self.__length_us

    def get_start_date(self):
        return self.__start_date

    def get_stop_date(self):
        return self.__stop_date

    def is_calibrated(self):
        return self.__calibrated

    def get_data(self):
        return TobiiSegmentData(os.path.join(self.__segment_path, TOBII_SEGMENT_DATA_FILENAME))

    def get_video(self):
        return TobiiSegmentVideo(os.path.join(self.__segment_path, TOBII_SEGMENT_VIDEO_FILENAME))

class TobiiRecording:
    """Handle Tobii Glasses Pro 2 recording info and segments."""

    def __init__(self, recording_path):
        """Load recording info from recording directory."""

        self.__recording_id = os.path.basename(recording_path)
        self.__recording_path = recording_path
        self.__project_path = os.path.dirname(os.path.dirname(os.path.abspath(self.__recording_path)))

        with open(os.path.join(self.__recording_path, TOBII_RECORD_FILENAME)) as f:
            try:
                item = json.load(f)
            except:
                raise RuntimeError(f'JSON fails to load {self.__recording_path}/{TOBII_RECORD_FILENAME}')

        self.__recording_created = datetime.datetime.strptime(item["rec_created"], TOBII_DATETIME_FORMAT)
        self.__recording_name = item["rec_info"]["Name"]
        self.__recording_length = int(item["rec_length"])
        self.__recording_segments = int(item["rec_segments"])
        self.__recording_et_samples = int(item["rec_et_samples"])
        self.__recording_et_valid_samples = int(item["rec_et_valid_samples"])
        self.__project = TobiiProject(self.__project_path)
        self.__participant = TobiiParticipant(self.__recording_path)

    def get_attributes(self):
        """Get recording attributes dictionnary."""

        attr = []
        names = []
        names.append("Recording name"); attr.append(self.__recording_name)
        names.append("Project name"); attr.append(self.__project.getName())
        names.append("Creation Date"); attr.append(str(self.__recording_created))
        names.append("Duration (s)"); attr.append(str(self.__recording_length))
        names.append("Participant name"); attr.append(self.__participant.getName())
        names.append("Segments"); attr.append(str(self.__recording_segments))
        names.append("Et samples"); attr.append(str(self.__recording_et_samples))
        names.append("Et valid samples"); attr.append(str(self.__recording_et_valid_samples))
        return (names, attr)

    def get_path(self):
        return self.__recording_path

    def get_creation_date(self):
        return self.__recording_created

    def get_et_samples(self):
        return self.__recording_et_samples

    def get_et_valid_samples(self):
        return self.__recording_et_valid_samples

    def get_id(self):
        return self.__recording_id

    def get_length(self):
        return self.__recording_length

    def get_name(self):
        return self.__recording_name

    def get_participant(self):
        return self.__participant

    def get_recording_directory(self):
        return self.__recording_dir

    def get_segment(self, segment_id):
        if segment_id > 0:
            return self.__segments[segment_id-1]
        raise ValueError('Cannot get segment less or equal to zero')

    def get_all_segments(self):

        all_segments = []
        segments_path = os.path.join(self.__recording_path, TOBII_SEGMENTS_DIRNAME)
        for item in os.listdir (segments_path):
            segment_path = os.path.join(segments_path, item)
            if os.path.isdir(segment_path):
                all_segments.append(TobiiSegment(segment_path))

        return all_segments

class TobiiParticipant:
    """Handle Tobii Glasses Pro 2 participant data."""

    def __init__(self, participant_path):
        """Load participant data from path"""

        self.__participant_id = os.path.basename(participant_path)
        self.__participant_path = participant_path

        with open(os.path.join(self.__participant_path, TOBII_PARTICIPANT_FILENAME)) as f:
            try:
                item = json.load(f)
            except:
                raise RuntimeError(f'JSON fails to load {source_dir}/{TOBII_PARTICIPANT_FILENAME}')

        self.__participant_name = item["pa_info"]["Name"]

    def get_path(self):
        return self.__participant_path

    def get_id(self):
        return self.__participant_id

    def get_name(self):
        return self.__participant_name

class TobiiProject:
    """Handle Tobii Glasses Pro 2 project data."""

    def __init__(self, project_path):
        """Load project data from projects directory and project id."""

        self.__project_id = os.path.basename(project_path)
        self.__project_path = project_path

        with open(os.path.join(self.__project_path, TOBII_PROJECT_FILENAME)) as f:
            try:
                item = json.load(f)
            except:
                raise RuntimeError(f'JSON fails to load {self.__project_path}/{TOBII_PROJECT_FILENAME}')

        self.__project_created = datetime.datetime.strptime(item["pr_created"], TOBII_DATETIME_FORMAT)

        try:
            self.__project_name = item["pr_info"]["Name"]
        except:
            self.__project_name = None

    def get_path(self):
        return self.__project_path

    def get_creation_date(self):
        return self.__project_created

    def get_id(self):
        return self.__project_id

    def get_name(self):
        return self.__project_name

    def get_all_participants(self):

        all_participants = []
        participants_path = os.path.join(self.__project_path, TOBII_PARTICIPANTS_DIRNAME)
        for item in os.listdir(participants_path):
            participant_path = os.path.join(participants_path, item)
            if os.path.isdir(participant_path):
                all_participants.append(TobiiParticipant(participant_path))

        return all_participants

    def get_all_recordings(self):

        all_recordings = []
        recordings_path = os.path.join(self.__project_path, TOBII_RECORDINGS_DIRNAME)
        for item in os.listdir(recordings_path):
            recording_path = os.path.join(recordings_path, item)
            if os.path.isdir(recording_path):
                all_recordings.append(TobiiRecording(recording_path))

        return all_recordings

class TobiiDrive:
    """Handle Tobii Glasses Pro 2 drive data."""

    def __init__(self, drive_path):
        """Load drive data from drive directory path."""

        self.__drive_path = drive_path

    def get_path(self):
        return self.__drive_path

    def get_all_projects(self):

        all_projects = []
        projects_path = os.path.join(self.__drive_path, TOBII_PROJECTS_DIRNAME)
        for item in os.listdir(projects_path):
            project_path = os.path.join(projects_path, item)
            if os.path.isdir(project_path):
                all_projects.append(TobiiProject(project_path))

        return all_projects