# -*- coding: utf-8 -*- ''' Created on Wed Jun 17 15:21:46 2020 @author: JTM Some example code for working with Pupil Core and getting data in real time. For more info go here: https://docs.pupil-labs.com/developer/core/network-api/#pupil-remote ''' from time import time from threading import Thread import msgpack import zmq class PupilCore(): ''' A Class for Pupil Core and the remote helper. ''' def __init__(self, address='127.0.0.1', request_port='50020'): ''' Initialise the connection with Pupil Core. Parameters ---------- address : string, optional The IP address of the device. The default is '127.0.0.1'. request_port : string, optional Pupil Remote accepts requests via a REP socket, by default on port 50020. Alternatively, you can set a custom port in Pupil Capture or via the --port application argument. The default is '50020'. Returns ------- None. ''' self.address = address self.request_port = request_port # connect to pupil remote self.context = zmq.Context() self.remote = zmq.Socket(self.context, zmq.REQ) self.remote.connect('tcp://{}:{}'.format(self.address, self.request_port)) # request 'SUB_PORT' for reading data self.remote.send_string('SUB_PORT') self.sub_port = self.remote.recv_string() # request 'PUB_PORT' for writing data self.remote.send_string('PUB_PORT') self.pub_port = self.remote.recv_string() # open socket for publishing self.pub_socket = zmq.Socket(self.context, zmq.PUB) self.pub_socket.connect('tcp://{}:{}'.format(self.address, self.pub_port)) def command(self, cmd): ''' Send a command via Pupil Remote. Parameters ---------- cmd : string Must be one of the following: 'R' - start recording with auto generated session name 'R rec_name' - start recording named "rec_name" 'r' - stop recording 'C' - start currently selected calibration 'c' - stop currently selected calibration 'T 1234.56' - resets current Pupil time to given timestamp 't' - get current Pupil time; returns a float as string 'v' - get the Pupil Core software version string 'PUB_PORT' - return the current pub port of the IPC Backbone 'SUB_PORT' - return the current sub port of the IPC Backbone Returns ------- string the result of the command. If the command was not acceptable, this will be 'Unknown command.' ''' # For every message that you send to Pupil Remote, you need to receive # the response. If you do not call recv(), Pupil Capture might become # unresponsive... self.remote.send_string(cmd) return self.remote.recv_string() def notify(self, notification): ''' Send a notification to Pupil Remote. Every notification has a topic and can contain potential payload data. The payload data has to be serializable, so not every Python object will work. To find out which plugins send and receive notifications, open the codebase and search for `.notify_all(` and `def on_notify(`. Parameters ---------- pupil_remote : zmq.sugar.socket.Socket the pupil remote helper. notification : dict the notification dict. Some examples: - {'subject':'start_plugin', 'name':'Annotation_Capture', 'args':{}}) - {'subject':'recording.should_start', 'session_name':'my session'} - {'subject':'recording.should_stop'} Returns ------- string the response. ''' topic = 'notify.' + notification['subject'] payload = msgpack.dumps(notification, use_bin_type=True) self.remote.send_string(topic, flags=zmq.SNDMORE) self.remote.send(payload) return self.remote.recv_string() def send_trigger(self, trigger): ''' Send an annotation (a.k.a 'trigger') to Pupil Capture. Use to mark the timing of events. Parameters ---------- pub_socket : zmq.sugar.socket.Socket a socket to publish the trigger. trigger : dict customiseable - see the new_trigger(...) function. Returns ------- None. ''' payload = msgpack.dumps(trigger, use_bin_type=True) self.pub_socket.send_string(trigger['topic'], flags=zmq.SNDMORE) self.pub_socket.send(payload) class PupilGrabber(Thread): ''' A thread-bound class for grabbing data from Pupil Core. Example ------- pupil = PupilCore() pg = PupilGrabber(pupil, topic='pupil.0.3d', secs=10) pg.start() ''' def __init__(self, pupil, topic, secs=None): ''' Prepare to start grabbing some data from pupil. Follow up with PupilGrabber.start() to begin. Parameters ---------- pupil : pupil.PupilCore PupilCore class instance topic : string Subscription topic. Can be: 'pupil.0.2d' - 2d pupil datum (left) 'pupil.1.2d' - 2d pupil datum (right) 'pupil.0.3d' - 3d pupil datum (left) 'pupil.1.3d' - 3d pupil datum (right) 'gaze.3d.1.' - monocular gaze datum 'gaze.3d.01.' - binocular gaze datum 'logging' - logging data Other topics are available from plugins (e.g. fixations, surfaces) and custom topics can be defined. secs : float, optional Ammount of time to spend grabbing data. Will run indefinitely if no value is passed, in which case requires PupilGrabber.join(). Returns ------- None. ''' super(PupilGrabber, self).__init__() self.pupil = pupil self.topic = topic self.secs = secs self.data = [] # a unique, encapsulated subscription to frame.world self.subscriber = self.pupil.context.socket(zmq.SUB) self.subscriber.connect('tcp://{}:{}'.format(self.pupil.address, self.pupil.sub_port)) self.subscriber.subscribe(self.topic) # TODO: add check on topic subscription def run(self): ''' Override the threading.Thread.run() method with code for grabbing data. Returns ------- None. ''' print('PupilGrabber now grabbing {} seconds of {}'.format( '?' if not self.secs else self.secs, self.topic)) if not self.secs: self.secs, t1, t2 = 0, -1, -2 # dummy values else: t1, t2 = time(), time() while t2 - t1 < self.secs: topic, payload = self.subscriber.recv_multipart() message = msgpack.loads(payload) self.data.append(message) t2 = time() print('PupilGrabber done grabbing {} seconds of {}'.format( self.secs, self.topic)) def get(self, what): ''' Get grabbed data. Parameters ---------- what : string The key of the data to be accessed. E.g. 'diameter_3d', 'timestamp', 'gaze_point_3d'. Returns ------- np.array() The requested data. ''' return [entry[what.encode()] for entry in self.data]