Skip to content

Instantly share code, notes, and snippets.

@kleinschmidt
Created April 24, 2020 18:10
Show Gist options
  • Save kleinschmidt/fd3d66b908784c542b917b1acd09ba63 to your computer and use it in GitHub Desktop.
Save kleinschmidt/fd3d66b908784c542b917b1acd09ba63 to your computer and use it in GitHub Desktop.
create a "tech support" HIT based on an existing supersubmiterator config
#!/usr/bin/env python
import json, argparse, os, csv
import boto3
mturk = boto3.client('mturk')
def main():
parser = argparse.ArgumentParser(description="Technical Support for MTurk workers")
parser.add_argument(
"--qualification_id",
metavar="id",
type=str,
help="The QualificationTypeId to grant to workers who need support. " +
"If ommited, a new qualification type will be created",
default=None)
parser.add_argument(
"label",
metavar="label",
help="supersubmiterator-style label. workerIds will be loaded from {label}-tech-support-workerids.csv " +
"and config from {label}.config. A new config file will be created called {label}-tech-support.config",
type=str)
args = parser.parse_args()
label = args.label
qual_id = args.qualification_id
config = parse_config(label)
workers = load_workers(label)
qual_id = check_qual_id(label, qual_id)
grant_qual_to_workers(qual_id, workers)
new_config = update_config(config, workers, qual_id)
new_config_fn = label + "-tech-support.config"
with open(new_config_fn, "w") as new_config_file:
print("Writing updated config...")
json.dump(new_config, new_config_file)
print("✔ {}".format(new_config_fn))
balance = mturk.get_account_balance()
if float(balance["AvailableBalance"]) < float(config["reward"]) * 1.1 * len(workers):
warn("⚠ available balance of {} is not sufficient to cover {} workers at {} each".format(
balance["AvailableBalance"],
len(workers),
config["reward"]
))
print("""
Next steps:
1. use supersubmiterator to run HIT
""")
def parse_config(label):
config_fn = label + ".config"
with open(config_fn, "r") as config_file:
config = json.load(config_file)
if config["liveHIT"] != "yes":
warn("⚠ Trying to grant tech support based on sandbox config in {}".format(config_fn))
return(config)
def load_workers(label):
workers_fn = label + "-tech-support-workerids.csv"
print("Loading workers needing tech support from {}...".format(workers_fn))
try:
with open(workers_fn, "r") as workers_file:
contents = csv.DictReader(workers_file)
workers = []
for row in contents:
workers.append(row["workerId"])
print("✔ loaded {} workers".format(len(workers)))
except:
print("✘ does {} exist and have workerId column?".format(workers_fn))
return(workers)
# create a qualification type
def check_qual_id(label, qual_id):
if qual_id == None:
qual_name = label + "-tech-support"
print("Creating new qualification type: {}...".format(qual_name))
try:
response = mturk.create_qualification_type(
Name=qual_name,
Description="Technical support for LeAP Lab study",
QualificationTypeStatus="Active",
AutoGranted=False
)
qual_id = response["QualificationType"]["QualificationTypeId"]
except:
# possibly failure due to existing qual id with same name...
print("✘ checking for existing type with same name...")
response = mturk.list_qualification_types(
Query=qual_name,
MustBeRequestable=False,
MustBeOwnedByCaller=True
)
if response["NumResults"] == 1:
qual_id = response["QualificationTypes"][0]["QualificationTypeId"]
else:
error("Need exactly one qualification type matching name {}\n{}".format(qual_name, response))
print("✔ {}".format(qual_id))
return(qual_id)
else:
print("Checking existing qualification type: {}".format(qual_id))
try:
response = mturk.get_qualification_type(
QualificationTypeId=qual_id
)
qual_type = response["QualificationType"]
print("✔ Found existing qualification:\n ", qual_type)
qual_id = qual_type["QualificationTypeId"]
return(qual_id)
except Exception as err:
print("✘", qual_id, err)
raise(err)
def grant_qual_to_workers(qual_id, workers):
print("Assigning qualification {} to {} workers".format(qual_id, len(workers)))
for worker in workers:
try:
mturk.associate_qualification_with_worker(
QualificationTypeId=qual_id,
WorkerId=worker,
IntegerValue=100,
SendNotification=False
)
print("✔ {}".format(worker))
except Exception as err:
print("✘ {}: {}".format(worker, err))
# create modified HIT config
def update_config(config, workers, qualification_id):
new_config = dict()
new_config["liveHIT"] = config["liveHIT"]
new_config["title"] = config["title"] + " TECH SUPPORT"
new_config["description"] = "Technical support for workers in LeAP Lab studies"
new_config["reward"] = config["reward"]
new_config["hitlifetime"] = 60 * 60 * 24 # one day
new_config["assignmentduration"] = 60 * 10 # 10 minutes
new_config["autoapprovaldelay"] = 60 * 60 # one hours
new_config["experimentURL"] = "https://experiments.leap-lab.org/tech-support/"
new_config["frameheight"] = config["frameheight"]
new_config["numberofassignments"] = len(workers)
new_config["doesHaveQualification"] = qualification_id
# these are mandatory keys that don't seem strictly necessary
new_config["USonly?"] = config["USonly?"]
new_config["minPercentPreviousHITsApproved"] = config["minPercentPreviousHITsApproved"]
return new_config
if __name__ == '__main__':
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment