This is some code I developed a while back, with slight modifications to try it out on a single computer. The solution is based on SQLAlchemy, which allows you to map database structures to Python objects (object relational mapping, ORM). It’s currently using the SQLite backend, a file-based, serverless DB system, which is great for playing around with it locally.Do not place an SQLite DB on a fileserver for concurrent access from different computers. Most network file systems do not implement locking correctly, and the DB will break. You may want to use MariaDB instead once you need a “remote” DB.
The example code assumes we have 5 independent players (implemented as Python threads via joblib); it runs 10 trials. On each trial, a random response and RT are generated, as well as a timestamp – independently for all participants. These data are then stored in the database – again, each of the five threads accessing the database “at random” – we do not have to worry about locking, the DB takes care of this!
The database structure is defined in the class ResponseState – that’s our ORM right there. We drop the DB table storing the results every time the script is run via ResponseState.__table__.drop(engine); this is because we use the trial number as a primary key, and it wouldn’t be unique anymore on a second run.
The wait_for_responses() function simply checks whether we have already collected responses from all players in the current trial; once we do, it sets the all_done entry in the database to True, and we can proceed to the next trial.
The simulation can be run by invoking run_experiment(). Please adjust the path in base_dir first.
Once the experiment is over, we can very easily extract the data from the DB using pandas; see extract_results(): It’s basically a one-liner!!!
Please refer to the SQLAlchemy and respective pandas docs for further info.
For local testing with SQLite, I always have my DB open in SQLite Manager, a Firefox addon.
I hope this helps you a bit, it’s really simple, although some things may look a bit weird at first. But once you understand the key concepts, it’s really not difficult anymore.
And a big sorry again for responding so late. Too many commitments, too little time. 
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import os
import datetime
import sqlalchemy as sa
import pandas as pd
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import Column, Integer, DateTime, Float, Boolean
from sqlalchemy.exc import (IntegrityError, InvalidRequestError,
OperationalError)
from numpy.random import random_sample, random_integers
from psychopy.core import wait
from joblib import Parallel, delayed
def run_trial(trial_num):
session = Session()
response_state = ResponseState(trial=trial_num)
session.add(response_state)
try:
session.commit()
except (IntegrityError, InvalidRequestError):
session.rollback()
session.close()
raise RuntimeError('Something went wrong here!!')
# Five "Parallel" "players" (Python threads)
Parallel(n_jobs=5, backend='threading')(
delayed(gen_response)(player_num=i) for i in range(1, 5 + 1))
def gen_response(player_num):
session = Session()
r = (session
.query(ResponseState)
.order_by(ResponseState.trial.desc())
.first())
# Create random response as integer in the interval [1, 10].
response = random_integers(1, 10)
# Create random RT in the range [1, 3] sec, and wait for the 'response'.
rt = (3 - 1) * random_sample() + 1
wait(rt)
time = datetime.datetime.now()
r.__setattr__('player_%i_response' % player_num,
response)
r.__setattr__('player_%i_rt' % player_num,
rt)
r.__setattr__('player_%i_time' % player_num,
time)
session.add(r)
try:
session.commit()
except (IntegrityError, InvalidRequestError):
session.rollback()
session.close()
raise RuntimeError('Something went wrong here!!')
def wait_for_responses():
session = Session()
latest_trial = (session.query(ResponseState)
.order_by(ResponseState.trial.desc())
.first())
player_num = 1
while player_num <= 5:
response = latest_trial.__getattribute__('player_%i_response'
% player_num)
if response is None:
wait(0.1)
continue
player_num += 1
latest_trial.all_done = True
session.add(latest_trial)
try:
session.commit()
except (IntegrityError, InvalidRequestError):
session.rollback()
session.close()
raise RuntimeError('Something went wrong here!!')
def run_experiment():
trials = range(1, 10 + 1)
for trial in trials:
run_trial(trial)
wait_for_responses()
print('Experiment completed.')
def extract_results():
data = pd.read_sql('responses', engine)
return data
if __name__ == '__main__':
base_dir = os.path.normpath('D:/Development/Testing/database')
db_file = os.path.join(base_dir, 'database.db')
engine = sa.create_engine('sqlite:///' + db_file, echo=True)
Session = sessionmaker(bind=engine)
Base = declarative_base()
class ResponseState(Base):
__tablename__ = 'responses'
trial = Column(Integer, primary_key=True)
all_done = Column(Boolean)
player_1_response = Column(Integer)
player_2_response = Column(Integer)
player_3_response = Column(Integer)
player_4_response = Column(Integer)
player_5_response = Column(Integer)
player_1_rt = Column(Float)
player_2_rt = Column(Float)
player_3_rt = Column(Float)
player_4_rt = Column(Float)
player_5_rt = Column(Float)
player_1_time = Column(DateTime)
player_2_time = Column(DateTime)
player_3_time = Column(DateTime)
player_4_time = Column(DateTime)
player_5_time = Column(DateTime)
try:
ResponseState.__table__.drop(engine)
except OperationalError:
pass
Base.metadata.create_all(engine)
