Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
launch_eye_process(eye_id)
else:
stop_eye_process(eye_id)
def set_detection_mapping_mode(new_mode):
n = {"subject": "set_detection_mapping_mode", "mode": new_mode}
ipc_pub.notify(n)
try:
from background_helper import IPC_Logging_Task_Proxy
IPC_Logging_Task_Proxy.push_url = ipc_push_url
from tasklib.background.patches import IPCLoggingPatch
IPCLoggingPatch.ipc_push_url = ipc_push_url
# display
import glfw
from version_utils import VersionFormat
from pyglui import ui, cygl, __version__ as pyglui_version
assert VersionFormat(pyglui_version) >= VersionFormat(
"1.24"
), "pyglui out of date, please upgrade to newest version"
from pyglui.cygl.utils import Named_Texture
import gl_utils
# helpers/utils
from file_methods import Persistent_Dict
from methods import normalize, denormalize, delta_t, get_system_info, timer
from uvc import get_time_monotonic
kwargs: Keyword arguments for the task.
pass_shared_memory: If True, a shared memory object will be passed to the
task (see tasklib.background.shared_memory.py). For this to work,
your task needs to accept a keyword argument `shared_memory`.
patches (collection of Patch): Patches that will be applied to your
background process right after start. Use them if you need to fix
something in the environment of the new process (see
tasklib.background.patches.py).
Per default, the IPC logging is patched.
Returns:
A new task with base class TaskInterface.
"""
if patches is ...:
patches = (IPCLoggingPatch(),)
task = tasklib.background.create(
name,
routine_or_generator_function,
pass_shared_memory,
args,
kwargs,
patches,
)
self._recently_added_tasks.append(task)
return task
g_pool.rec_dir,
)
args = (
calibration.result,
fake_gpool,
pupil_pos_in_mapping_range,
gaze_mapper.manual_correction_x,
gaze_mapper.manual_correction_y,
)
name = "Create gaze mapper {}".format(gaze_mapper.name)
return tasklib.background.create(
name,
_map_gaze,
args=args,
patches=[bg_patches.IPCLoggingPatch()],
pass_shared_memory=True,
)
for ref in all_reference_locations
if frame_start <= ref.frame_index <= frame_end
]
fake_gpool = _setup_fake_gpool(
g_pool.capture.frame_size,
g_pool.capture.intrinsics,
calibration.mapping_method,
g_pool.rec_dir,
calibration.minimum_confidence,
)
args = (fake_gpool, ref_dicts_in_calib_range, pupil_pos_in_calib_range)
name = "Create calibration {}".format(calibration.name)
return tasklib.background.create(
name, _create_calibration, args=args, patches=[bg_patches.IPCLoggingPatch()]
)
Normally, you would not use this directly, but use a PluginTaskManager and the
create_background_task() method there.
If you don't use a manager, you need to regularly call update() on the returned task
to trigger callbacks etc.
See the docstring for PluginTaskManager.create_background_task()
(in tasklib.manager.py) for information about the different parameters!
"""
if args is None:
args = ()
if kwargs is None:
kwargs = {}
if patches is None:
patches = [
IPCLoggingPatch(),
KeyboardInterruptHandlerPatch(),
]
if inspect.isgeneratorfunction(routine_or_generator_function):
return BackgroundGeneratorFunction(
name,
routine_or_generator_function,
pass_shared_memory,
args,
kwargs,
patches,
)
elif inspect.isroutine(routine_or_generator_function):
return BackgroundRoutine(
name,
routine_or_generator_function,