import csv import os import random from enum import Enum from shared.reward_group import RewardGroup from shared import helpers class ReceiverType(Enum): ACCOUNT_PARTNER = 1 THIRD_PARTY = 2 def set_receivers(subsession, session) -> None: from . import Receiver, C def validate_row(): def is_integer(col): try: return int(row[col]) except ValueError: raise RuntimeError(f"{col}: {row[col]} must be an integer") def is_float(col): try: return float(row[col]) except ValueError: raise RuntimeError(f"{col}: {row[col]} must be a float") def is_not_empty(col): if val := row[col].strip(): return val else: raise RuntimeError(f"{col}: {val} must not be empty") def is_reward_group(col): if (val := row[col].strip()) in reward_groups: return val else: raise RuntimeError(f"{col}: {val} must be valid reward group") receiver = is_integer("participant.id_in_session") receiver_session = is_not_empty("session.code") reward_group = is_reward_group(f"real_effort.{round_nr}.player.reward_group") reward = is_float(f"real_effort.{round_nr}.player.reward") return receiver, receiver_session, reward_group, reward dirname = os.path.dirname(__file__) filename = os.path.join(dirname, "receivers.csv") types = [ReceiverType.ACCOUNT_PARTNER.value, ReceiverType.THIRD_PARTY.value] reward_groups = [str(RewardGroup.HIGH), str(RewardGroup.LOW)] round_nr = subsession.round_number with open(filename, newline="") as f: reader = csv.DictReader(f, delimiter=",", quotechar='"') for row in reader: # Validate row receiver, receiver_session, reward_group, reward = validate_row() # Fetch shared name and profile_picture receiver_key = helpers.get_receiver_key(receiver_session, receiver) receiver_data = session.receivers.get( receiver_key, dict( receiver_id=next(session.id_gen), name=random.choice(helpers.PSEUDONYMS), profile_picture=random.choice(helpers.PROFILE_PICTURES), paid_round=random.choice(range(1, C.NUM_ROUNDS + 1)), type=random.choices(types, weights=(70, 30), k=1)[0], allocations=[], ), ) # Store name and profile_picture to share for next subessions if receiver_key not in session.receivers: session.receivers[receiver_key] = receiver_data is_round_paid = True if receiver_data.get("paid_round") == round_nr else False Receiver.create( receiver_id=receiver_data.get("receiver_id"), subsession=subsession, dictator=None, # Assigned to no dictator initially type=receiver_data.get("type"), receiver_session=receiver_session, receiver=receiver, name=receiver_data.get("name"), reward_group=reward_group, reward=reward, profile_picture=receiver_data.get("profile_picture"), payoff=reward if is_round_paid else None, # Set their reward as default if this round is paid is_round_paid=is_round_paid, ) # Check that there are at least six Third-party receivers third_party_receivers = Receiver.filter(subsession=subsession, type=ReceiverType.THIRD_PARTY.value) high_receivers = [r for r in third_party_receivers if r.reward_group == str(RewardGroup.HIGH)] low_receivers = [r for r in third_party_receivers if r.reward_group == str(RewardGroup.LOW)] if len(high_receivers) < 3: raise RuntimeError("There must be at least 3 Third-Party-Receivers from the High reward group.") if len(low_receivers) < 3: raise RuntimeError("There must be at least 3 Third-Party-Receivers from the Low reward group.")