API#

CLI#

class pupil_labs.lsl_relay.cli.DeviceDiscoverer(search_timeout: float)#

Bases: object

async get_device_from_list()#

pupil_labs.lsl_relay.cli.epoch_is(year: int, month: int, day: int) → bool#

pupil_labs.lsl_relay.cli.evaluate_user_input(user_input: str, device_list: Sequence[DiscoveredDeviceInfo]) → DiscoveredDeviceInfo | None#

async pupil_labs.lsl_relay.cli.get_device_info_for_outlet(device_ip: str, device_port: int)#

pupil_labs.lsl_relay.cli.get_user_defined_device(device_address: str)#

async pupil_labs.lsl_relay.cli.input_async()#

pupil_labs.lsl_relay.cli.logger_setup(file_name: str, debug_level: str | int = 10)#

async pupil_labs.lsl_relay.cli.main_async(device_address: str | None = None, outlet_prefix: str = '', time_sync_interval: int = 60, timeout: int = 10)#

pupil_labs.lsl_relay.cli.print_device_list(network: Network, n_reload: int)#

Relay#

class pupil_labs.lsl_relay.relay.DataReceiver(device_ip: str, device_port: int)#

Bases: object

async cleanup()#

async estimate_clock_offset()#

Estimate the Companion-Device-to-Relay clock offset.

Uses the Pupil Labs Time Echo Protocol to measure the clock offset between the Companion Device and the relay. The offset is used to transform incoming gaze and event timestamps from Companion to relay time domain. If the Companion app version does not support the time echo protocol or the clock offset estimation fails, the relay will fall back to NTP-based time sync.

async make_status_update_notifier()#

async on_update(component: Phone | Hardware | Sensor | Recording | NetworkDevice)#

class pupil_labs.lsl_relay.relay.EventAdapter(sample: Event, clock_offset_ns: int)#: Bases: object

class pupil_labs.lsl_relay.relay.GazeAdapter(sample: GazeData, clock_offset_ns: int)#: Bases: object

class pupil_labs.lsl_relay.relay.Relay(device_ip: str, device_port: int, receiver: DataReceiver, device_identifier: str, outlet_prefix: str, model: str, module_serial: str, time_sync_interval: int)#

Bases: object

async initialise_tasks() → List[Task[NoReturn]]#

async publish_event_from_queue()#

async publish_gaze_sample(timeout: float)#

async receive_gaze_sample()#

async relay_receiver_to_publisher()#

async classmethod run(device_ip: str, device_port: int, device_identifier: str, outlet_prefix: str, model: str, module_serial: str, time_sync_interval: int)#

async start_publishing_event()#

async start_publishing_gaze()#

async start_receiving_task()#

pupil_labs.lsl_relay.relay.handle_done_pending_tasks(done: Iterable[Task[NoReturn]], pending: Iterable[Task[NoReturn]])#

async pupil_labs.lsl_relay.relay.send_events_in_interval(device_ip: str, device_port: int, session_id: str, sec: int = 60)#

async pupil_labs.lsl_relay.relay.send_timesync_event(device_ip: str, device_port: int, message: str)#

Channels#

class pupil_labs.lsl_relay.channels.CompanionChannel(sample_query: Callable[[Any], Any], channel_information_dict: Dict[str, str])#

Bases: object

append_to(channels: XMLElement)#

class pupil_labs.lsl_relay.channels.Gaze(*args, **kwargs)#

Bases: Protocol

x: float#

y: float#

pupil_labs.lsl_relay.channels.companion_event_channels()#

pupil_labs.lsl_relay.channels.companion_gaze_channels() → List[CompanionChannel]#

pupil_labs.lsl_relay.channels.pi_extract_from_sample(value: str) → Callable[[Any], Any]#

pupil_labs.lsl_relay.channels.pi_extract_gaze_query(dim: Literal[0, 1]) → Callable[[Gaze], float]#

Outlets#

class pupil_labs.lsl_relay.outlets.PupilCompanionEventOutlet(device_id: str, outlet_prefix: str, model: str, module_serial: str, session_id: str, clock_offset_ns: int = 0)#: Bases: PupilCompanionOutlet

class pupil_labs.lsl_relay.outlets.PupilCompanionGazeOutlet(device_id: str, outlet_prefix: str, model: str, module_serial: str, session_id: str, clock_offset_ns: int = 0)#: Bases: PupilCompanionOutlet

class pupil_labs.lsl_relay.outlets.PupilCompanionOutlet(channel_func: Callable[[], List[CompanionChannel]], outlet_type: str, outlet_format: Literal[1, 2, 3, 4, 5, 6, 7, 0], outlet_name_prefix: str, outlet_uuid: str, acquisition_info: Dict[str, str])#

Bases: object

push_sample_to_outlet(sample: Sample)#

class pupil_labs.lsl_relay.outlets.Sample(*args, **kwargs)#

Bases: Protocol

timestamp_unix_seconds: float#: Unix-epoch timestamp in seconds

pupil_labs.lsl_relay.outlets.compose_acquisition_info(version: str, module_serial: str, session_id: str, manufacturer: str = 'Pupil Labs', model: str = 'Pupil Labs Device', clock_offset_ns: int = 0) → Dict[str, str]#

pupil_labs.lsl_relay.outlets.get_lsl_time_offset()#

pupil_labs.lsl_relay.outlets.pi_create_outlet(outlet_uuid: str, channels: List[CompanionChannel], outlet_type: str, outlet_format: Literal[1, 2, 3, 4, 5, 6, 7, 0], outlet_name_prefix: str, acquisition_info: Dict[str, str])#

pupil_labs.lsl_relay.outlets.pi_streaminfo(outlet_uuid: str, channels: List[CompanionChannel], type_name: str, channel_format: Literal[1, 2, 3, 4, 5, 6, 7, 0], outlet_name_prefix: str, acquisition_info: Dict[str, str])#