#!/usr/bin/env python from typing import TypeVar from dataclasses import dataclass, field import json from argaze import DataStructures @dataclass(frozen=True) class PupilDiameter(): """Define pupil diameter as ...""" value: float = field(default=0.) """Pupil diameter value.""" @property def valid(self) -> bool: """Is the value not 0""" return self.value != 0. def __repr__(self): """String representation""" return json.dumps(self, ensure_ascii = False, default=vars) class UnvalidPupilDiameter(PupilDiameter): """Unvalid pupil diameter.""" def __init__(self, message=None): self.message = message super().__init__(0.) TimeStampedPupilDiametersType = TypeVar('TimeStampedPupilDiameters', bound="TimeStampedPupilDiameters") # Type definition for type annotation convenience class TimeStampedPupilDiameters(DataStructures.TimeStampedBuffer): """Define timestamped buffer to store pupil diameters.""" def __setitem__(self, key, value: PupilDiameter|dict): """Force PupilDiameter storage.""" # Convert dict into PupilDiameter if type(value) == dict: assert(set(['value']).issubset(value.keys())) if 'message' in value.keys(): value = UnvalidPupilDiameter(value['message']) else: value = PupilDiameter(value['value']) assert(type(value) == PupilDiameter or type(value) == UnvalidPupilDiameter) super().__setitem__(key, value) @classmethod def from_json(self, json_filepath: str) -> TimeStampedPupilDiametersType: """Create a TimeStampedPupilDiametersType from .json file.""" with open(json_filepath, encoding='utf-8') as ts_buffer_file: json_buffer = json.load(ts_buffer_file) return TimeStampedPupilDiameters({ast.literal_eval(ts_str): json_buffer[ts_str] for ts_str in json_buffer}) TimeStampedBufferType = TypeVar('TimeStampedBuffer', bound="TimeStampedBuffer") # Type definition for type annotation convenience class PupilDiameterAnalyzer(): """Abstract class to define what should provide a pupil diameter analyser.""" def analyze(self, ts, pupil_diameter) -> float: """Analyze pupil diameter from successive timestamped pupil diameters.""" raise NotImplementedError('analyze() method not implemented') def browse(self, ts_pupil_diameters: TimeStampedPupilDiameters) -> TimeStampedBufferType: """Analyze by browsing timestamped pupil diameters.""" assert(type(ts_pupil_diameters) == TimeStampedPupilDiameters) ts_analyzis = DataStructures.TimeStampedBuffer() # Iterate on pupil diameters for ts, pupil_diameter in ts_pupil_diameters.items(): analysis = self.analyze(ts, pupil_diameter) if analysis is not None: ts_analyzis[ts] = analysis return ts_analyzis