alt_experiment_design.py

from datetime import datetime
import enum
from infoaggexperiment.database import db
import random
from sqlalchemy.ext.hybrid import hybrid_property
import uuid


NO_FEEDBACK_GROUP = 1


class Treatment(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    profiles = db.relationship('Profile')
    name = db.Column(db.String(10))
    p = db.Column(db.Float, nullable=False)
    q = db.Column(db.Float, nullable=False)
    H = db.Column(db.Integer, nullable=False)
    L = db.Column(db.Integer, nullable=False)
    s = db.Column(db.Float, nullable=False)

    def __repr__(self):
        return '<Treatment {}>'.format(self.name)


class Profile(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    treatment = db.Column(db.Integer, db.ForeignKey('treatment.id'))
    reliability = db.Column(db.PickleType, nullable=False)
    prob_unbias = db.Column(db.PickleType, nullable=False)


class Experiment(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    session_id = db.Column(db.Integer, nullable=False)
    treatments = db.relationship('Treatment', secondary=session_treatment_table)
    profiles = db.relationship('Profile')
    elections = db.relationship('Election')
    group = db.Column(db.Integer, nullable=False)
    subject_id = db.Column(db.Integer, db.ForeignKey('subject.id'))
    subject = db.relationship('Subject', back_populates='experiment')
    completed = db.Column(db.Boolean, defualt=False)

    def __repr__(self):
        return '<Session={}Group={}>'.format(self.session_id, self.group)

    @property
    def feedback(self):
        """Should the subject receive feedback."""
        return self.group == NO_FEEDBACK_GROUP
    
    @property
    def current_election(self):
        """Return the latest not completed election."""
        return Election.query.filter_by(session_id=self.session_id,
                                        completed=False).first()

    @property
    def current_treatment(self):
        """Return the treatment in the current election."""
        if self.current_election is not None:
            return Treatment.query.get(self.current_election.treatment_id)

    @hybrid_property
    def last_vote_time(self):
        return self.current_election.vote_time


class Vote(enum.Enum):
    incorrect = 0
    correct = 1
    abstain = 2


class Election(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    experiment_id = db.Column(db.Integer, db.ForeignKey('experiment.id'), nullable=False)
    experiment = db.relationship('Experiment')
    treatment_id = db.Column(db.Integer, db.ForeignKey('treatment.id'), nullable=False)
    election_num = db.Column(db.Integer, nullable=False)
    reliability = db.Column(db.Float, nullable=False)
    state = db.Column(db.String(10), nullable=False)
    info = db.Column(db.Integer, nullable=False)
    is_biased = db.Column(db.Boolean)
    vote = db.Column(db.Integer)
    vote_time = db.Column(db.DateTime)
    _group_decision = db.Column(db.Integer)

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.state = random.choice(['red', 'blue'])
        self.info = Vote.correct if random.random() < self.reliability else Vote.incorrect

        # nonexperts are not biased. Experts are biased with probability 1 - s
        treatment = Treatment.query.get(self.treatment_id)
        if self.reliability == treatment.p:
            # subject is an expert
            self.is_biased = random.random() < 1 - treatment.s
        else:
            # subject is not an expert and cannot be biased
            self.is_biased = False

    def vote(self, choice):
        # record vote
        assert vote in ('red', 'blue', 'abstain')
        if vote == self.correct_jar:
            self.subject_player.votes.append(Vote.correct)
        elif vote == self.incorrect_jar:
            self.subject_player.votes.append(Vote.incorrect)
        else:
            self.subject_player.votes.append(Vote.abstain)
        self.vote_time = datetime.utcnow()

        check_completed(self.experiment)
    
    def voters(self):
        """Collate the other voters in this election.

        Get the current election in the other session-treatment-group.

        If the group is the NO_FEEDBACK_GROUP, then we use the current group.
        Otherwise, we use the previous group.

        """
        Voter = namedtuple('Voter', 'reliability vote')

        if self.experiment.group == NO_FEEDBACK_GROUP:
            group = self.subject.experiment.group
        else:
            group = self.subject.experiment.group - 1

        # experiments in same session and group
        elections = Election.query.filter_by(session_id=self.session_id,
                                             treatment_id=self.treatment_id,
                                             group=group,
                                             election_num=self.election_num).all()
        return [Voter(e.reliability, e.vote) for e in elections]

    @hybrid_property
    def session_id(self):
        return self.experiment.session_id

    @hybrid_property
    def group(self):
        return self.experiment.group

    @hybrid_property
    def completed(self):
        return self.vote is not None 

    @property
    def group_decision(self):
        if self._group_decision:
            return self._group_decision

        voters = self.voters()
        if all(v.vote is not None for v in voters):
            # only do it if everyone completed their votes
            votes = [v.vote for v in voters]
            self._group_decision = count_votes(votes).value
            db.session.commit()
            return self._group_decision

    @property
    def score(self):
        if self.group_decision is None:
            return 0

        # If the subject is biased, then they are paid if their bias was
        # chosen; unbiased subjects are paid if the group decision is correct.
        if self.is_biased:
            bias_correct = self.info == Vote.correct
            group_correct = self.group_decision == Vote.correct
            return 30 if bias_correct == group_correct else -70
        return 30 if self.group_decision == Vote.correct else -70


class Subject(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    hit_id = db.Column(db.String(120), nullable=False, unique=True)
    worker_id = db.Column(db.String(120), nullable=False, unique=True)
    assignment_id = db.Column(db.String(120), nullable=False, unique=True)
    experiment_id = db.Column(db.Integer, db.ForeignKey('experiment.id'), unique=True)
    experiment = db.relationship('Experiment')
    survey_code = db.Column(
        db.String(36),
        unique=True,
        default=lambda: str(uuid.uuid4()))
    paid = db.Column(db.Boolean, default=False)
    rewarded  = db.Column(db.Boolean, default=False)


# Election functions


def count_votes(votes):
    """Return Vote.correct or Vote.incorrect.

    This function is not pure; if the counts are equal, then
    we choose correct/incorrect randomly.

    """
    correct = votes.count(Vote.correct)
    incorrect = votes.count(Vote.incorrect)
    if correct > incorrect:
        return Vote.correct
    elif correct < incorrect:
        return Vote.incorrect
    else:
        # choose correct/incorrect randomly
        return Vote.correct if random.random() < 0.5 else Vote.incorrect


# Subject functions


def create_subject(hit_id, worker_id, assignment_id):
    """Create subject and associate with experiment entry.

    postconditions: See assert_create_postconditions.

    Raises
    - ExperimentDone: all experiment entries completed
    - ExperimentBusy: no available experiment.

    """
    experiment = find_available_experiment()

    # create new subject and associate with experiment entry
    subject = Subject(hit_id=hit_id,
                      worker_id=worker_id,
                      assignment_id=assignment_id,
                      experiment=experiment)

    experiment.subject = subject
    assert_create_postconditions()

    db.session.add(subject)
    db.session.commit()
    return subject


def find_available_experiment():
    """Find first available experiment.

    Find first available experiment. An experiment is available if
    all experiments in the previous group are completed.

    preconditions: at least one available experiment

    Raises:
    - ExperimentDone: all experiment entries completed
    - ExperimentBusy: no experiment available

    """
    if all(e.completed for e in Experiment.query.all()):
        raise ExperimentDone

    availableq = Experiment.query.filter(Experiment.subject_id.isnot(None))
    experiments = Experiment.query.filter(Experiment.subject_id.is_(None)).all()
    for experiment in experiments:
        if experiment.group == NO_FEEDBACK_GROUP:
            return experiment
        completed = availablq.filter_by(session_id=experiment.session_id,
                                        group=experiment.group - 1).all()
        if all(e.completed for e in completed.all()):
            return experiment

    if all(e.completed for e in Experiment.query.all()):
        raise ExperimentDone
    raise ExperimentBusy


def assert_create_postconditions(subject):
    """Assert create consistency.

    - experiment is not available (has subject)
    - experiment is not completed
    - all elections have no vote
    - all elections have no vote_time
    - all elections have no _group_decision
    - subject has an experiment assigned to him/her

    """

    assert subject.experiment.subject_id is not None
    assert subject.experiment.subject is not None
    assert not subject.experiment.completed

    elections = subject.experiment.elections
    assert all(e.vote is None for e in elections)
    assert all(e.vote_time is None for e in elections)
    assert all(e._group_decision is None for e in elections)

    assert subject.experiment_id is not None
    assert subject.experiment is not None


def drop_subject(subject):
    """Drop subject from experiment.

    postconditions: See assert_drop_postconditions

    """
    assert subject.experiment_id is not None

    # save these identifiers to assert
    subject_id = subject.id
    experiment_id = subject.experiment.id

    # reset elections
    for election in subject.experiment.elections:
        election.vote = None
        election.vote_time = None
        election._group_decision = None

    # reset experiment entry
    subject.experiment.completed = False
    subject.experiment.subject = None

    # reset subject
    subject.experiment = None
    db.session.commit()

    # assert postconditions
    subject = Subject.query.get(subject_id)
    experiment = Experiment.query.get(experiment_id)
    assert_drop_postconditions(subject, experiment)


def assert_drop_postconditions(subject, experiment):
    """Assert drop consistency.

    - experiment is available (no subject)
    - experiment is not completed
    - all elections have no vote
    - all elections have no vote_time
    - all elections have no _group_decision
    - subject does not have an experiment assigned to him/her

    """
    assert experiment.subject_id is None
    assert experiment.subject is None
    assert not experiment.completed

    elections = experiment.elections
    assert all(e.vote is None for e in elections)
    assert all(e.vote_time is None for e in elections)
    assert all(e._group_decision is None for e in elections)

    assert subject.experiment_id is None
    assert subject.experiment is None
    

def check_subject_inactive(subject):
    if subject.experiment is not None:
        time_since_last_vote = datetime.utcnow() - subject.Experiment.last_vote_time
        if time_since_last_vote > app.timeout_threshold:
            # subject is inactive and should be dropped
            drop_subject(subject)


def mark_subject_completed(subject):
    subject.experiment.completed = True
    db.session.commit()


def check_group_status(session_id, group_id):
    """check whether all experiments with session,group are completed"""
    experiments = Experiment.query.filter_by(session_id=session_id, group_id=group_id).all()
    if all(experiment.completed for experiment in experiments):
        calculate_results()


# Seed the database
# create 8 treatments
# create 120 experiments (session, treatments, profiles, groups)
# for each experiment
#     create elections_per_treatment elections per treatment
#     (election_num, bias, reliability)


# Experiment Flow
try:
    subject = create_subject(hit_id=1, worker_id=1, assignment_id=1)
except SessionBusy:
    print('Please come back in 45 minutes')
except ExperimentDone:
    print('The experiment is now completed. Thank you for your interest')

while not subject.completed:
	period = subject.create_period()
	period.vote('red')
	period.feedback()