from otree.api import * from .task import ArithmeticTaskFactory, ArithmeticTask from .magazines import all_magazines from datetime import datetime from typing import Tuple, List import time from treatments import Treatment from helpers import * doc = """ Your app description """ class C(BaseConstants): NAME_IN_URL = "real_effort_task" PLAYERS_PER_GROUP = None NUM_ROUNDS = 1 # Arithmetic Tasks TIME_ON_TASK_MIN = MIN_ON_TASK TIME_ON_TASK = SEC_ON_TASK NR_INITIAL_TASKS = 100 EXAMPLE_TASKS = ArithmeticTaskFactory.get_example_tasks() COMPREHENSION_TASKS = ArithmeticTaskFactory.get_comprehension_tasks() EXAMPLE_TASKS_ENUM = [(f"Beispiel {i + 1}", v) for i, v in enumerate(EXAMPLE_TASKS)] COMPREHENSION_TASKS_ENUM = [(f"Aufgabe {i + 1}", v) for i, v in enumerate(COMPREHENSION_TASKS)] NR_COMPREHENSION_TASKS = len(COMPREHENSION_TASKS) # Option Out MAGAZINES = all_magazines # Self-Assessment ASSESSMENT_PAYOFF = cu(1000) ASSESSMENT_RANGE = 5 # Treatments FIXED_WAGE_TREATMENT = Treatment.FIXED_WAGE PIECE_RATE_TREATMENT = Treatment.PIECE_RATE TOURNAMENT_TREATMENT = Treatment.TOURNAMENT class Subsession(BaseSubsession): pass def creating_session(subsession: Subsession): session = subsession.session players: List[Player] = subsession.get_players() ppg = session.config["players_per_group"] if len(players) % ppg: raise Exception(f"Number of participants must be multiple of {ppg}") # Create groups group_matrix = [] for i in range(0, len(players), ppg): group = players[i : i + ppg] group_matrix.append(group) subsession.set_group_matrix(group_matrix) # Generate initial Arithmetic Tasks session.tasks = ArithmeticTaskFactory.get_working_tasks(C.NR_INITIAL_TASKS) # Initialize participant vars for p in players: pt = p.participant pt.outside_option_start = None pt.worked = 0 pt.score = 0 pt.self_assessment = None pt.fixed_wage_payoff = None pt.piece_rate_payoff = None pt.tournament_payoff = None pt.ranking = None pt.self_assessment_payoff = None class Group(BaseGroup): pass class Player(BasePlayer): # Example task models second_example = models.BooleanField() # Comprehension task models comprehension_iteration = models.IntegerField(initial=0) test_error = models.IntegerField(initial=0) test_time = models.FloatField() # Working task models worked = models.IntegerField(initial=0) # nr of solved tasks score = models.IntegerField(initial=0) # nr of correctly solved tasks error = models.IntegerField(initial=0) # nr of incorrectly solved tasks outside_option_overall_time = models.FloatField() ranking = models.IntegerField() # Payoff models fixed_wage_payoff = models.CurrencyField(initial=0) piece_rate_payoff = models.CurrencyField(initial=0) tournament_payoff = models.CurrencyField(initial=0) self_assessment_payoff = models.CurrencyField(initial=0) # Self-Assessment models self_assessment = models.IntegerField(min=0) self_assessment_eval = models.BooleanField() def get_current_comprehension_task(self) -> Tuple[str, ArithmeticTask]: iteration = self.comprehension_iteration if iteration < C.NR_COMPREHENSION_TASKS: return C.COMPREHENSION_TASKS_ENUM[iteration] def get_current_working_task(self) -> ArithmeticTask: iteration = self.worked session = self.session if len(session.tasks) <= iteration: session.tasks += ArithmeticTaskFactory.get_working_tasks(50) return session.tasks[iteration] def create_option_out(self, start: float, end: float): option_out = OptionOut.create( player=self, iteration=self.worked + 1, start=start, start_str=str(datetime.fromtimestamp(start)), end=end, end_str=str(datetime.fromtimestamp(end)), timespan=end - start, ) class TaskInput(ExtraModel): player = models.Link(Player) iteration = models.IntegerField() timestamp = models.StringField() task = models.StringField() solution = models.IntegerField() answer = models.StringField() class OptionOut(ExtraModel): player = models.Link(Player) iteration = models.IntegerField() start_str = models.StringField() end_str = models.StringField() start = models.FloatField() end = models.FloatField() timespan = models.FloatField() class OpenPDF(ExtraModel): player = models.Link(Player) timestamp = models.StringField() name = models.StringField() def custom_export(players): # Export ExtraModel "TaskInput" yield ["session", "participant", "iteration", "timestamp", "task", "solution", "answer"] task_inputs = TaskInput.filter() for task_input in task_inputs: player = task_input.player participant = player.participant session = player.session yield [ session.code, participant.code, task_input.iteration, task_input.timestamp, task_input.task, task_input.solution, task_input.answer, ] # Export ExtraModel "OptionOut" yield ["session", "participant", "iteration", "start", "end", "timespan"] opt_outs = OptionOut.filter() for o in opt_outs: player = o.player participant = player.participant session = player.session yield [session.code, participant.code, o.iteration, o.start_str, o.end_str, o.timespan] # Export ExtraModel "OpenPDF" yield ["session", "participant", "timestamp", "pdf"] open_pdfs = OpenPDF.filter() for pdf in open_pdfs: player = pdf.player participant = player.participant session = player.session yield [session.code, participant.code, pdf.timestamp, pdf.name] # PAGES class Example1(Page): form_model = "player" form_fields = ["second_example"] @staticmethod def vars_for_template(player: Player): return dict(example=C.EXAMPLE_TASKS_ENUM[0]) @staticmethod def before_next_page(player: Player, timeout_happened): player.participant.comprehension_page_arrival = time.time() class Example2(Page): @staticmethod def vars_for_template(player: Player): return dict(example=C.EXAMPLE_TASKS_ENUM[1]) @staticmethod def before_next_page(player: Player, timeout_happened): player.participant.comprehension_page_arrival = time.time() @staticmethod def is_displayed(player: Player): return player.second_example class ComprehensionWait(WaitPage): title_text = "Bitte warten" body_text = "Es sind noch nicht alle TeilnehmerInnen bereit. Bitte haben Sie einen Moment Geduld. Das Experiment geht weiter, wenn alle TeilnehmerInnen bereit sind." wait_for_all_groups = True class ComprehensionInstructions(Page): pass class Comprehension(Page): @staticmethod def live_method(player: Player, message: dict): player_id = player.id_in_group if message["type"] == "answers": answers = message["answers"] if len(answers) != len(C.COMPREHENSION_TASKS_ENUM): return { player_id: { "type": "error", "err_msg": "Ein Fehler ist aufgetreten. Bitte versuchen Sie es erneut.", } } errors = [] for i in range(len(answers)): if answers[i] != C.COMPREHENSION_TASKS[i].get_solution(): errors.append(str(i + 1)) if error_len := len(errors): player.test_error += error_len return { player_id: { "type": "error", "err_msg": f"Die folgenden Aufgaben sind noch nicht korrekt beantwortet: {', '.join(errors)}. Bitte versuchen Sie es erneut.", } } return {player_id: {"type": "complete"}} @staticmethod def vars_for_template(player: Player): return dict(tasks=C.COMPREHENSION_TASKS_ENUM) @staticmethod def before_next_page(player: Player, timeout_happened): # Track the time spent on the Comprehension Page start = player.participant.comprehension_page_arrival end = time.time() player.test_time = end - start class ComprehensionFeedback(Page): pass class WorkingTaskInstructionsWait(WaitPage): title_text = "Bitte warten" body_text = "Es sind noch nicht alle TeilnehmerInnen bereit. Bitte haben Sie einen Moment Geduld. Das Experiment geht weiter, wenn alle TeilnehmerInnen bereit sind." wait_for_all_groups = True class WorkingTaskInstructions(Page): @staticmethod def vars_for_template(player: Player): fixed_wage = get_fixed_wage(player) piece_rate = get_piece_rate(player) [prize1, prize2, prize3, prize4] = get_tournament_prizes(player) return dict( treatment=get_treatment(player), fixed_wage=fixed_wage, piece_rate=piece_rate, prize1=prize1, prize2=prize2, prize3=prize3, prize4=prize4, ) class WorkingTaskWait(WaitPage): title_text = "Bitte warten" body_text = "Es sind noch nicht alle TeilnehmerInnen bereit. Bitte haben Sie einen Moment Geduld. Das Experiment geht weiter, wenn alle TeilnehmerInnen bereit sind." wait_for_all_groups = True class WorkingTask(Page): timeout_seconds = C.TIME_ON_TASK @staticmethod def live_method(player: Player, message: dict): now = time.time() message_type = message["type"] if message_type == "answer": answer = message["answer"] task = player.get_current_working_task() player.worked += 1 iteration = player.worked # Evaluate correctness if task.is_correct(answer): player.score += 1 else: player.error += 1 # Create a TaskInput data point TaskInput.create( player=player, iteration=iteration, timestamp=str(datetime.fromtimestamp(now)), solution=task.get_solution(), task=str(task), answer=str(answer), ) # Return next task next_task = player.get_current_working_task() return {player.id_in_group: {"type": "task", "equations": next_task.get_equations_as_string()}} elif message_type == "opt_out": if start_ts := player.participant.outside_option_start: # Player opt'ed back in to working task print( f"P{player.id_in_subsession} has opt'ed out of the working task at {now} for {now - start_ts} seconds." ) player.create_option_out(start=start_ts, end=now) player.participant.outside_option_start = None # Close Option Out else: # Player opt'ed out of working task print(f"P{player.id_in_subsession} has opt'ed out of the working task at {now}.") player.participant.outside_option_start = now # Open Option Out elif message_type == "reload": if start_ts := player.participant.outside_option_start: # Player was in outside option when reloaded player.create_option_out(start=start_ts, end=now) player.participant.outside_option_start = None # Close Option Out elif message_type == "open_pdf": pdf_name = message["pdf_name"] OpenPDF.create(player=player, timestamp=str(datetime.fromtimestamp(now)), name=pdf_name) @staticmethod def vars_for_template(player: Player): return dict(tsk=player.get_current_working_task(), magazines=C.MAGAZINES) @staticmethod def before_next_page(player: Player, timeout_happened): # Check if player was in outside option when timeout_happened now = time.time() if start_ts := player.participant.outside_option_start: player.create_option_out(start=start_ts, end=now) player.participant.outside_option_start = None # Close Option Out # Sum up total time spent in outside option time_spent_out = sum([o.timespan for o in OptionOut.filter(player=player) if o.timespan]) player.outside_option_overall_time = time_spent_out # Save to participant model pt = player.participant pt.worked = player.worked pt.score = player.score class WorkingTaskEnd(Page): pass class WorkingTaskResultWait(WaitPage): title_text = "Bitte warten" body_text = "Es sind noch nicht alle TeilnehmerInnen bereit. Bitte haben Sie einen Moment Geduld. Das Experiment geht weiter, wenn alle TeilnehmerInnen bereit sind." # TODO: Ensure that these work correctly since returning currencies instead of floats @staticmethod def after_all_players_arrive(group: Group): fixed_wage = get_fixed_wage(group) piece_rate = get_piece_rate(group) tournament_prizes = get_tournament_prizes(group) get_treatment(group).set_payoffs(group, fixed_wage, piece_rate, tournament_prizes) class SelfAssessment(Page): form_model = "player" form_fields = ["self_assessment"] @staticmethod def before_next_page(player: Player, timeout_happened): score = player.score assessment_range = range(score - C.ASSESSMENT_RANGE, score + C.ASSESSMENT_RANGE + 1) if player.self_assessment in assessment_range: player.self_assessment_eval = True player.self_assessment_payoff = C.ASSESSMENT_PAYOFF player.payoff += C.ASSESSMENT_PAYOFF else: player.self_assessment_eval = False # Save to participant model pt = player.participant pt.self_assessment = player.self_assessment pt.self_assessment_payoff = player.self_assessment_payoff page_sequence = [ Example1, Example2, ComprehensionWait, ComprehensionInstructions, Comprehension, ComprehensionFeedback, WorkingTaskInstructionsWait, WorkingTaskInstructions, WorkingTaskWait, WorkingTask, WorkingTaskEnd, WorkingTaskResultWait, SelfAssessment, ]