Sync eye movements with external events

Question

This solution was used to synchronize events between two applications: An eye tracking software, Python, and a stimulus control software, object Free Pascal/Delphi. It avoided a rewrite of the stimulus control app from ObjFPC/Delphi to Python.

Pupil Eye Tracker sends some info via ZeroMQ to somewhere (I get used to call "net stack" to this place, please do not laugh). Indeed, a lot of info is sent, so they provided a way to filter this events; I have done this in a way that only timestamps should be sent.

This is a dummy stand alone broadcast:

"""

    Broadcast dummy Pupil timestamps

"""

import zmq
#from ctypes import create_string_buffer
from time import sleep

def test_msg():
    test_msg = "Pupil\ntimestamp:1389761135.56\n"
    return test_msg

def main():
    context = zmq.Context()
    socket = context.socket(zmq.PUB)
    #address = create_string_buffer("tcp://127.0.0.1:5020",512)
    address = "tcp://127.0.0.1:5020"
    try:
        #socket.bind(address.value)
        socket.bind(address)
    except zmq.ZMQError:
        print "Could not set Socket."

    for i in range(120):
        socket.send( test_msg() )
        sleep(0.5)

    context.destroy()


if __name__ == '__main__':
    main() 

A timestamp is the basic unit to synchronize frames, eye-gaze, events, basically all the stuff. Initially, they used the Python now() function to generate them. The current version can use the hardware of some cameras as well: "Hardware time-stamping: Running Pupil Pro with Linux now uses hardware timestamps taken by the camera hardware at the start of exposure".

Since they are using a Publisher-Subscriber protocol, Pupil Team suggested a way to receive this info. So, all I have done was to call the 'receive' client from inside the stimulus control app using a single thread for each call. Of course, it requires a broadcast server, in my case, during the real time Pupil Capture Server broadcast.

unit client;

{$mode objfpc}{$H+}

interface

uses
   Classes
 , SysUtils
 , Process
 , FileUtil
 , zmqapi
 ;

type

  TShowStatusEvent = procedure(Status: String) of object;

  { TClientThread }

  TClientThread = class(TThread)
  private
    FServerAddress: string;
    FContext : TZMQContext;
      FSubscriber : TZMQSocket;
    FMsg : string;
    FTrialIndex : string;
    FCode : string;
    FTimestampsPath : string;
    FOnShowStatus: TShowStatusEvent;
    FTimestampsFile : TextFile;
    function GetTimestampFromMessage(aMessage : Utf8String) : Utf8String;
    procedure SetServerAddress(AValue: string);
    procedure ShowStatus;
  protected
    procedure Execute; override;
  public
    constructor Create(TrialIndex : integer; Code, OutputPath : string; CreateSuspended : boolean = True);
    destructor Destroy; override;
    property OnShowStatus: TShowStatusEvent read FOnShowStatus write FOnShowStatus;
    property ServerAddress : string read FServerAddress write SetServerAddress;
  end;

implementation

constructor TClientThread.Create(TrialIndex : integer; Code, OutputPath : string; CreateSuspended : boolean = True);
begin
  FreeOnTerminate := True;

  FTrialIndex := IntToStr(TrialIndex);
  FCode := Code;
  FTimestampsPath := OutputPath;
  FServerAddress := '127.0.0.1:5000';

  ForceDirectoriesUTF8(ExtractFilePath(FTimestampsPath));
    AssignFile(FTimestampsFile, FTimestampsPath);
    if FileExistsUTF8(FTimestampsPath) then
      Append(FTimestampsFile)
    else Rewrite(FTimestampsFile);

  inherited Create(CreateSuspended);
end;

destructor TClientThread.Destroy;
begin
  //
  inherited Destroy;
end;

procedure TClientThread.ShowStatus;
// this method is executed by the mainthread and can therefore access all GUI elements.
begin
  if Assigned(FOnShowStatus) then
  begin
    FOnShowStatus(FMsg);
  end;
end;

function TClientThread.GetTimestampFromMessage(aMessage: Utf8String
  ): Utf8String;

  var aKey, aHeader : Utf8String;
begin
    aHeader := 'Pupil';
    aKey := 'timestamp:';
    Delete(  aMessage, Pos(aHeader, aMessage), Length(aHeader)  );
    Delete(  aMessage, Pos(aKey, aMessage), Length(aKey)  );

    while Pos(#10, aMessage) <> 0 do
        Delete(  aMessage, Pos(#10, aMessage), Length(#10)  );

    Result := aMessage;

end;

procedure TClientThread.SetServerAddress(AValue: string);
begin
  if FServerAddress = AValue then Exit;

  if Length(AValue) > 0 then FServerAddress := AValue;
end;


procedure TClientThread.Execute;
var
  data : string;
  message : UTF8String;
begin
  try

    FContext := TZMQContext.Create;
      FSubscriber := FContext.Socket( stSub );
    FSubscriber.connect( 'tcp://' + FServerAddress);
    FSubscriber.subscribe( '' );

    FSubscriber.recv( message );
    // ('value', 'value', 'value')
    data := #40#39 + FTrialIndex + #39#44#32#39 + GetTimestampFromMessage(message) + #39#44#32#39 + FCode + #39#41;

      Writeln(FTimestampsFile, data);


    FMsg := data + #10#10 +
            '"' + FTimestampsPath + '"'  + #10 +
            '"' + FServerAddress + '"';

    Synchronize( @Showstatus );

  finally
    CloseFile(FTimestampsFile);
    FSubscriber.Free;
      FContext.Free;
    Terminate;
  end;
end;

end.

Call example:

unit SomeForm;

{$mode objfpc}{$H+}

interface

uses
...

    , client
    ;

...

type

    procedure CreateClientThread(Code : string);
    procedure DebugStatus(msg : string);

resourcestring

    Event1 = 'Something happens';

implementation

procedure CreateClientThread(Code : string);
var
  Client : TClientThread;
  Filename : string;
  SomeInteger: integer;
begin
  SomeInteger : -1;
  Filename := GetCurrentDirUTF8 + '/timestamps';

  Client := TClientThread.Create( SomeInteger, Code, FileName );
  Client.OnShowStatus := @DebugStatus;
  Client.ServerAddress := '127.0.0.1:5020';
  Client.Start;
end; 

procedure TSomeform.DebugStatus(msg: string);
begin
  ShowMessage(msg);
end;  

For post-facto analysis, is simple to read the yielded data, here timestamps_by_trial, from inside Pupil Player:

def set_var(self, timestamps, current_path):
    import os
    import ast

    self.timestamps = timestamps
    self.frame_count = len(self.timestamps)

    timestamps_by_trial_path = current_path + os.sep + 'timestamps'
    timestamps_by_trial = [[]]
    with open(timestamps_by_trial_path) as f:
        for line in f:
            temp = ast.literal_eval(line)

            # temp = [0, 1, 2] = (trial_index, timestamp, a_code)
            i = int(temp[0])
            timestamp = (temp [1], temp[2])

            if i > len(timestamps_by_trial):
                timestamps_by_trial.append([])
            timestamps_by_trial[i - 1].append(timestamp)
        f.close
    self.timestamps_by_trial = timestamps_by_trial

Based on this data, we could draw some visual feedback too. The following picture shows the "Trial Events" in the bottom seek bar of Pupil Player:

Note that the delphizmq binding was used in a scope with zmq_version 2.2.0 and under Unix OS's. The following is a debug Subscriber client.

Please run this program from a terminal that you know will stay opened after execution.

program zmq_subscriber;


uses
    SysUtils
    , Classes
    , zmqapi
    ;

const
    UPDATE_COUNT = 1;

var
    context : TZMQContext;
    subscriber : TZMQSocket;

    i : integer;

    address, filter : string;
    message : Utf8String;
    stringlist : TStringList;
begin

    // Socket to talk to server
    address := '127.0.0.1:5020';
    Writeln ( 'Connecting to Pupil server...' + #32#40 + 'address:' + #32 + address + #32#41);
    context := TZMQContext.Create;
    subscriber := Context.Socket( stSub );
    subscriber.connect( 'tcp://' + address );

    // You can choose a filter here
    if ParamCount > 0 then
        filter := ParamStr( 1 )
    else
        filter := '';

    subscriber.subscribe( filter );
    stringlist := TStringList.Create;

    try
        // \n = #10
        stringlist.Delimiter := #10;

        for i := 0 to UPDATE_COUNT - 1 do
            begin
                subscriber.recv( message );
                stringlist.Clear;
                stringlist.DelimitedText := message;

                Writeln( stringlist.Text );
            end;

    finally
        stringlist.Free;
        subscriber.Free;
        context.Free;       
    end;
end. 

I have some questions:

• Since the Python code is just what you need to test the client, and a specific analytic example to give you the dimension of the application of this stuff, I am specially concerned about the Pascal code. I choose to call the client from inside a thread and to create a single thread every time an Event occur. Everything should run just fine, except when the granularity of the Events increase a lot in time, which becomes very processor consuming. Any comments about that would be appreciated.

• Is the code readable?

Answer 1

I don't know any Pascal or Delphi, so I can't be any help here, but I can make a few suggestions to improve your general Python style.

---

Standalone broadcast

• The Python style guide has some rules for formatting module imports. In particular, you're supposed to group your imports into standard library, third-party and project-specific imports. Within each group, they should be listed alphabetically.

  Assuming that the ZeroMQ module is third-party, this means your module imports should be organised like this:

  # from ctypes import create_string_buffer
  from time import sleep
  
  import zmq
  

• Within test_msg(), making a variable of the same name seems bound to cause confusion and/or errors. Either rename it, or return the string directly.

  Either

  def test_msg():
      return_str = "Pupil\ntimestamp:1389761135.56\n"
      return return_str
  

  or

  def test_msg():
      return "Pupil\ntimestamp:1389761135.56\n"
  

  although it's not clear why this needs to be a function, if all it returns is a string. Could you not just make the return string a global variable?

• Your script needs more comments. I can follow what the program is doing, but I can't see why it's doing this. For example, the string in test_msg(). What are those numbers? Where do they come from? Or in main(), why are you connecting to port 5020?

  It's good to provide some background information to make it easier for future readers.

• You use i as the index variable in your for loop in main(), but never access the value of i. It's common practice to use _ as your index variable in such a situation: this is generally understood to mean that the value of the index variable is unimportant.

Post-factor analysis

• Again, comments

• It's fairly unusual to put module imports within a function body. Move them to the top of the script.

• For constructing the timestamps_by_trial_path, rather than doing string concatenation with os.sep, use os.path.join():

  timestamps_by_trial_path = os.path.join(current_path, 'timestamps')
  

• When you do with open(...), you need to specify the mode with which to open the file. The common values are 'r' (read-only) and 'w' (write), but others are available. It should be:

  with open(timestamps_by_trial_path, 'r') as f:
  

• It's clear the components of temp have some meaning – it's expressed by your comment. So rather than giving them opaque names like i or temp[1], actually use their names. You can use tuple unpacking to do this incredibly easily:

  (trial_index, timestamp, a_code) = temp
  

  You can now use those names, which makes it easier to see the purpose behind your code.

• If you use the with open(...) as f: construction, you don't need an f.close at the end. The file object f is closed automatically when you finish the with block. That's the advantage of using with open instead of f.open(); ...; f.close() – it's handled for you.