EyeTrackVR/RANSAC3d/pye3dcustom/eye_model/asynchronous.py
Prohurtz 3bcfc3280a
RANSAC3d BEST METHOD
finally. .. good code lol
2022-05-29 18:43:08 -05:00

321 lines
11 KiB
Python

"""
(*)~---------------------------------------------------------------------------
Pupil - eye tracking platform
Copyright (C) 2012-2019 Pupil Labs
Distributed under the terms of the GNU
Lesser General Public License (LGPL v3.0).
See COPYING and COPYING.LESSER for license details.
---------------------------------------------------------------------------~(*)
"""
import ctypes
import logging
import typing as T
import numpy as np
from ..constants import DEFAULT_SPHERE_CENTER
from .abstract import (
TwoSphereModelAbstract,
CameraModel,
Circle,
Observation,
ObservationStorage,
SphereCenterEstimates,
)
from .background_helper import BackgroundProcess, mp
from .base import TwoSphereModel
logger = logging.getLogger(__name__)
class TwoSphereModelAsync(TwoSphereModelAbstract):
def __init__(
self,
camera: CameraModel,
storage_cls: T.Type[ObservationStorage] = None,
storage_kwargs: T.Dict = None,
):
synced_sphere_center = mp.Array(ctypes.c_double, 3)
synced_corrected_sphere_center = mp.Array(ctypes.c_double, 3)
synced_projected_sphere_center = mp.Array(ctypes.c_double, 2)
synced_observation_count = mp.Value(ctypes.c_long)
synced_rms_residual = mp.Value(ctypes.c_double)
is_estimation_ongoing_flag = mp.Event()
self._frontend = _TwoSphereModelSyncedFrontend(
synced_sphere_center,
synced_corrected_sphere_center,
synced_projected_sphere_center,
synced_observation_count,
synced_rms_residual,
is_estimation_ongoing_flag,
camera=camera,
)
self._backend_process = BackgroundProcess(
function=self._process_relayed_commands,
setup=self._setup_backend,
setup_args=(
synced_sphere_center,
synced_corrected_sphere_center,
synced_projected_sphere_center,
synced_observation_count,
synced_rms_residual,
is_estimation_ongoing_flag,
),
setup_kwargs=dict(
camera=camera,
storage_cls=storage_cls,
storage_kwargs=storage_kwargs,
),
cleanup=self._cleanup_backend,
log_handlers=logging.getLogger().handlers,
)
@property
def sphere_center(self) -> np.ndarray:
return self._frontend.sphere_center
@property
def corrected_sphere_center(self) -> np.ndarray:
return self._frontend.corrected_sphere_center
@property
def projected_sphere_center(self) -> np.ndarray:
return self._frontend.projected_sphere_center
@property
def rms_residual(self) -> float:
return self._frontend.rms_residual
def relay_command(self, function_name: str, *args, **kwargs):
self._backend_process.send(function_name, *args, **kwargs)
@staticmethod
def _process_relayed_commands(
backend: "_TwoSphereModelSyncedBackend", function_name: str, *args, **kwargs
):
function = getattr(backend, function_name)
return function(*args, **kwargs)
@staticmethod
def _setup_backend(*args, **kwargs) -> "_TwoSphereModelSyncedBackend":
logger = logging.getLogger(__name__)
logger.debug(f"Setting up backend: {args}, {kwargs}")
return _TwoSphereModelSyncedBackend(*args, **kwargs)
@staticmethod
def _cleanup_backend(backend: "_TwoSphereModelSyncedBackend"):
backend.cleanup()
logger = logging.getLogger(__name__)
logger.debug(f"Backend cleaned")
def add_observation(self, observation: Observation):
self.relay_command("add_observation", observation)
@property
def n_observations(self) -> int:
return self._frontend.n_observations
def set_sphere_center(self, new_sphere_center: np.ndarray):
raise NotImplementedError
def estimate_sphere_center(
self,
from_2d: T.Optional[np.ndarray] = None,
prior_3d: T.Optional[np.ndarray] = None,
prior_strength: float = 0.0,
calculate_rms_residual=False,
) -> SphereCenterEstimates:
if not self._frontend._is_estimation_ongoing_flag.is_set():
self.relay_command(
"estimate_sphere_center",
from_2d,
prior_3d,
prior_strength,
calculate_rms_residual,
)
self._frontend._is_estimation_ongoing_flag.set()
projected_sphere_center = self._frontend.projected_sphere_center
sphere_center = self._frontend.sphere_center
rms_residual = self._frontend.rms_residual
return SphereCenterEstimates(
projected_sphere_center, sphere_center, rms_residual
)
def estimate_sphere_center_2d(self) -> np.ndarray:
raise NotImplementedError
def estimate_sphere_center_3d(
self,
sphere_center_2d: np.ndarray,
prior_3d: T.Optional[np.ndarray] = None,
prior_strength: float = 0.0,
calculate_rms_residual: bool = False,
) -> T.Tuple[np.array, T.Optional[float]]:
raise NotImplementedError
# GAZE PREDICTION
def _extract_unproject_disambiguate(self, pupil_datum: T.Dict) -> Circle:
return self._frontend._extract_unproject_disambiguate(pupil_datum)
def _disambiguate_circle_3d_pair(
self, circle_3d_pair: T.Tuple[Circle, Circle]
) -> Circle:
return self._frontend._disambiguate_circle_3d_pair(circle_3d_pair)
def predict_pupil_circle(
self, observation: Observation, use_unprojection: bool = False
) -> Circle:
return self._frontend.predict_pupil_circle(observation, use_unprojection)
def apply_refraction_correction(self, pupil_circle: Circle) -> Circle:
return self._frontend.apply_refraction_correction(pupil_circle)
def cleanup(self):
logger.debug("Cancelling backend process")
self._backend_process.cancel()
self._frontend.cleanup()
def mean_observation_circularity(self) -> float:
raise NotImplementedError
class _TwoSphereModelSyncedAbstract(TwoSphereModel):
def __init__(
self,
synced_sphere_center: mp.Array, # c_double_Array_3
synced_corrected_sphere_center: mp.Array, # c_double_Array_3
synced_projected_sphere_center: mp.Array, # c_double_Array_2
synced_observation_count: mp.Value, # c_long
synced_rms_residual: mp.Value, # c_double
flag_is_estimation_ongoing: mp.Event,
**kwargs,
):
self._synced_sphere_center = synced_sphere_center
self._synced_corrected_sphere_center = synced_corrected_sphere_center
self._synced_projected_sphere_center = synced_projected_sphere_center
self._synced_observation_count = synced_observation_count
self._synced_rms_residual = synced_rms_residual
self._is_estimation_ongoing_flag = flag_is_estimation_ongoing
super().__init__(**kwargs)
@property
def sphere_center(self):
with self._synced_sphere_center:
return np.array(self._synced_sphere_center.get_obj())
@sphere_center.setter
def sphere_center(self, coordinates: np.array):
raise NotImplementedError
@property
def corrected_sphere_center(self):
with self._synced_corrected_sphere_center:
return np.array(self._synced_corrected_sphere_center.get_obj())
@corrected_sphere_center.setter
def corrected_sphere_center(self, coordinates: np.array):
raise NotImplementedError
@property
def projected_sphere_center(self):
with self._synced_projected_sphere_center:
return np.array(self._synced_projected_sphere_center.get_obj())
@projected_sphere_center.setter
def projected_sphere_center(self, coordinates: np.array):
raise NotImplementedError
def mean_observation_circularity(self) -> float:
raise NotImplementedError
@property
def rms_residual(self) -> float:
with self._synced_rms_residual:
return self._synced_rms_residual.value
@rms_residual.setter
def rms_residual(self, residual: float):
raise NotImplementedError
class _TwoSphereModelSyncedFrontend(_TwoSphereModelSyncedAbstract):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
del self.storage # There is no storage in the frontend
def _set_default_model_params(self):
with self._synced_sphere_center:
self._synced_sphere_center[:] = DEFAULT_SPHERE_CENTER
corrected_sphere_center = self.refractionizer.correct_sphere_center(
np.asarray([[*self.sphere_center]])
)[0]
with self._synced_corrected_sphere_center:
self._synced_corrected_sphere_center[:] = corrected_sphere_center
@property
def n_observations(self) -> int:
return self._synced_observation_count.value
class _TwoSphereModelSyncedBackend(_TwoSphereModelSyncedAbstract):
@property
def sphere_center(self):
return super().sphere_center
@sphere_center.setter
def sphere_center(self, coordinates: np.array):
with self._synced_sphere_center:
self._synced_sphere_center[:] = coordinates
@property
def corrected_sphere_center(self):
return super().corrected_sphere_center
@corrected_sphere_center.setter
def corrected_sphere_center(self, coordinates: np.array):
with self._synced_corrected_sphere_center:
self._synced_corrected_sphere_center[:] = coordinates
@property
def projected_sphere_center(self):
return super().projected_sphere_center
@projected_sphere_center.setter
def projected_sphere_center(self, coordinates: np.array):
with self._synced_projected_sphere_center:
self._synced_projected_sphere_center[:] = coordinates
def add_observation(self, observation: Observation):
super().add_observation(observation=observation)
n_observations = super().n_observations
with self._synced_observation_count:
self._synced_observation_count.value = n_observations
@property
def n_observations(self) -> int:
return self._synced_observation_count.value
def estimate_sphere_center(self, *args, **kwargs):
result = super().estimate_sphere_center(*args, **kwargs)
self._is_estimation_ongoing_flag.clear()
return result
def estimate_sphere_center_2d(self) -> np.ndarray:
estimated: np.ndarray = super().estimate_sphere_center_2d()
self.projected_sphere_center = estimated
return estimated
@property
def rms_residual(self) -> float:
with self._synced_rms_residual:
return self._synced_rms_residual.value
@rms_residual.setter
def rms_residual(self, residual: float):
with self._synced_rms_residual:
self._synced_rms_residual.value = residual