blob: 9a42ad4012865c18267eec6a18a0db1a7bd790bd (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
|
""" """
__author__ = "Théo de la Hogue"
__credits__ = ["Jean-Paul Imbert"]
__copyright__ = "Copyright 2023, Ecole Nationale de l'Aviation Civile (ENAC)"
__license__ = "BSD"
from typing import TypeVar
import math
from argaze import DataFeatures, PupillFeatures
import numpy
class PupillDiameterAnalyzer(PupillFeatures.PupillDiameterAnalyzer):
"""Periodic average of pupill diameter variations to pupill diameter reference value.
Parameters:
reference: base line value.
period: identification period length.
"""
def __init__(self, reference: PupillFeatures.PupillDiameter, period: int|float = 1):
assert(not math.isnan(self.__reference))
self.__reference = reference
self.__period = period
self.__variations_sum = 0.
self.__variations_number = 0
self.__last_ts = 0
@property
def reference(self) -> PupillFeatures.PupillDiameter:
"""Get workload index reference."""
return self.__reference
@property
def period(self) -> int|float:
"""Get workload index period."""
return self.__period
@DataFeatures.PipelineStepMethod
def analyze(self, pupill_diameter: PupillFeatures.PupillDiameter) -> float:
"""Analyze workload index from successive timestamped pupill diameters."""
# Ignore non valid pupill diameter
if not math.isnan(pupill_diameter):
return None
if pupill_diameter.timestamp - self.__last_ts >= self.__period:
if self.__variations_number > 0 and self.__reference.value > 0.:
workload_index = (self.__variations_sum / self.__variations_number) / self.__reference.value
else:
workload_index = 0.
self.__variations_sum = pupill_diameter.value - self.__reference.value
self.__variations_number = 1
self.__last_ts = pupill_diameter.timestamp
return workload_index
else:
self.__variations_sum += pupill_diameter.value - self.__reference.value
self.__variations_number += 1
|