simulation.py

#!/usr/bin/env python3

import re
import argparse
import pylab as plt
import seaborn as sns
import pandas as pd
import numpy as np
from collections import defaultdict

import models
from models import Student, Items, get_answer_sequences
import evaluation_metrics as em
import optimization
from config import SKILLS, LOW_SKILL, HIGH_SKILL, AVG_SKILL
import config as cf

sns.set_theme(style="ticks")


def get_short_name(metric_name):
    return re.sub(r'\(.*\)', '', metric_name).replace("_", " ")


def visualize_progress(pm_list, student, items, n=100, short_name=True):
    """Visualizes the dynamics of progress measures in pm_list for
    a single sequence of student answers.
    Corresponds to Fig. 4 and Fig. 8 in the paper."""
    for pm in pm_list:
        pm.reset()
    answers = student.get_answer_sequence(items, n)
    for i in range(n):
        for pm in pm_list:
            pm.update(answers[i])
    plt.scatter([x for x in range(len(answers)) if answers[x] == 1],
                [1.03 for y in answers if y == 1], marker="o", color="green", s=15)
    plt.scatter([x for x in range(len(answers)) if answers[x] == 0],
                [-0.03 for y in answers if y == 0], marker="o", color="red", s=15)
    for pm in pm_list:
        name = str(pm)
        if short_name:
            name = get_short_name(name)
        plt.plot(pm.history, label=name)
        print(pm.mastery_step, str(pm), sep="\t")
    plt.xlabel("attempt")
    plt.ylabel("progress measurement")
    plt.legend()
    plt.show()


def compute_durations(scenario, pm_list, skills=None, rep=500, as_time=False, by_skill=True):
    multiplyer = 1
    if as_time:
        multiplyer = scenario.time_intensity
    if skills is None:
        skills = SKILLS
    if by_skill:
        durations = defaultdict(dict)
    else:
        durations = defaultdict(list)
    limit = cf.get_attempts_limit(scenario)
    for skill in skills:
        answer_seqs = get_answer_sequences(scenario, skill, rep, limit)
        for pm in pm_list:
            results = [pm.steps_to_mastery(answer_seqs[i]) * multiplyer
                       for i in range(rep)]
            if by_skill:
                durations[pm.name][skill] = results
            else:
                durations[pm.name].extend(results)
    return durations


def plot_specs_vlines(eval_specs):
    plt.vlines(AVG_SKILL,
               0,
               eval_specs["A_under_1"],
               color="red", alpha=0.7, lw=4)
    plt.vlines(AVG_SKILL,
               eval_specs["A_under_1"],
               eval_specs["A_under_0"],
               color="red", alpha=0.7, lw=2, linestyle="--")
    plt.vlines(AVG_SKILL,
               eval_specs["A_over_0"],
               eval_specs["A_over_1"],
               color="red", alpha=0.7, lw=2, linestyle="--")
    plt.vlines(AVG_SKILL,
               eval_specs["A_over_1"],
               eval_specs["L_under_0"],
               color="red", alpha=0.7, lw=4)
    plt.vlines(HIGH_SKILL,
               eval_specs["H_over_0"],
               eval_specs["H_over_1"],
               color="red", alpha=0.7, lw=2, linestyle="--")
    plt.vlines(HIGH_SKILL,
               eval_specs["H_over_1"],
               eval_specs["L_under_0"],
               color="red", alpha=0.7, lw=4)
    plt.vlines(LOW_SKILL,
               0,
               eval_specs["L_under_1"],
               color="red", alpha=0.7, lw=4)
    plt.vlines(LOW_SKILL,
               eval_specs["L_under_1"],
               eval_specs["L_under_0"],
               color="red", alpha=0.7, lw=2, linestyle="--")


def plot_skill_practice_distribution(scenario, pm_list, rep=1000):
    """For a given scenario and  each progress measure in pm_list, computes
    and displays the distribution of completion times for students with
    various skills.
    Corresponds to Fig. 10 in the paper."""
    skills = list(map(lambda x: round(x, 2),
                      np.linspace(LOW_SKILL, HIGH_SKILL, 18)))
    durations = compute_durations(scenario, pm_list, skills, rep, as_time=True)
    eval_specs = cf.get_eval_specs(scenario, as_time=True)
    plt.figure(figsize=(16, 9))
    plt.suptitle(f"{scenario.name}: {scenario.desc}")
    for j, pm in enumerate(pm_list):
        plt.subplot(2, 3, j+1)
        plt.title(get_short_name(pm.name))
        plt.plot(skills, [np.median(durations[pm.name][skill]) for skill in skills],
                 color="black", linewidth=2)
        plt.fill_between(skills,
                         [min(durations[pm.name][skill]) for skill in skills],
                         [max(durations[pm.name][skill]) for skill in skills],
                         color='gray', alpha=0.1)
        plt.fill_between(skills,
                         [np.percentile(durations[pm.name][skill], 5) for skill in skills],
                         [np.percentile(durations[pm.name][skill], 95) for skill in skills],
                         color='gray', alpha=0.2)
        plt.fill_between(skills,
                         [np.percentile(durations[pm.name][skill], 25) for skill in skills],
                         [np.percentile(durations[pm.name][skill], 75) for skill in skills],
                         color='gray', alpha=0.4)
        plot_specs_vlines(eval_specs)
        plt.xlabel("Skill")
        plt.ylabel("Time (s)")
    plt.tight_layout()
    plt.show()


def value_to_color(val, cmap='rocket_r', blend=0.5):
    cmap = plt.cm.get_cmap(cmap)
    rgba = cmap(val)
    r, g, b = [int((1 - blend) * 255 + blend * 255 * x) for x in rgba[:3]]
    return f'\\cellcolor[RGB]{{{r},{g},{b}}}{val}'


def evaluate(scenario, durations):
    pm_names = list(durations.keys())
    eval_specs = cf.get_eval_specs(scenario)
    metrics = defaultdict(dict)
    for pm_name in pm_names:
        dur = durations[pm_name]
        for metric_fun in em.METRICS:
            metrics[pm_name][metric_fun.__name__] = metric_fun(dur, eval_specs)
    metrics = pd.DataFrame(metrics).transpose()
    extended_metrics = metrics.copy()
    extended_metrics["mellow_max"] = metrics.apply(
        lambda row: em.mellow_max(row.tolist()), axis=1)
    extended_metrics.sort_values("mellow_max", inplace=True)
    print(extended_metrics)


def run_evaluation(scenario, pm_list, rep=2000):
    """Evaluates progress measures in pm_list for a given scenario.
    Corresponds to Table 4 in the paper."""
    durations = compute_durations(scenario, pm_list, rep=rep)
    evaluate(scenario, durations)


def scatter_plots(scenario, pm_list, students=100):
    """Creates scatter plots of completition times of individual progress measures with respect to the last progress measure in pm_list.
    Corresponds to Fig. 11 in the paper."""
    time_limit = cf.get_attempts_limit(scenario) * scenario.time_intensity
    durations = compute_durations(scenario, pm_list, rep=students, by_skill=True, as_time=True)
    colors = sns.color_palette("hls", len(SKILLS))
    plt.figure(figsize=(15,4))
    n = len(pm_list)
    selected = n-1
    for j in range(n-1):
        plt.subplot(1, n-1, j+1)
        plt.plot((0, time_limit), (0, time_limit), color="gray")
        for i, skill in enumerate(SKILLS):
            x, y = durations[pm_list[j].name][skill], durations[pm_list[selected].name][skill]
            plt.scatter(x, y, color=colors[i], alpha=0.3)
        if j == 0:
            plt.ylabel(f"{get_short_name(pm_list[selected].name)} time (s)")
        plt.xlabel("Completion time (s)")
        plt.title(get_short_name(pm_list[j].name))
    plt.tight_layout()
    plt.show()


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("cmd", type=str)
    parser.add_argument("-s", "--scenario", type=str, default="mcq")
    parser.add_argument("-p", "--pmlist", type=str, default="mcq-opt")
    parser.add_argument("-i", "--initskill", type=float, default=0)
    parser.add_argument("-l", "--learnrate", type=float, default=0)
    args = parser.parse_args()
    cmd = args.cmd

    if cmd == "vis":
        scenario = cf.scenarios[args.scenario]
        visualize_progress(cf.pm_lists[args.pmlist],
                           Student(args.initskill, args.learnrate),
                           Items(scenario),
                           n=int(300/scenario.time_intensity))
    elif cmd == "eval":
        run_evaluation(cf.scenarios[args.scenario], cf.pm_lists[args.pmlist])
    elif cmd == "optimize":
        optimization.optimize_all(cf.scenarios[args.scenario])
    elif cmd == "skilldistr":
        plot_skill_practice_distribution(cf.scenarios[args.scenario],
                                         cf.pm_lists[args.pmlist])
    elif cmd == "scatter":
        scatter_plots(cf.scenarios[args.scenario], cf.pm_lists[args.pmlist])


if __name__ == "__main__":
    main()