progress_measures.py

"""Progress measures implementation.
Implemented using classes, with inheratance used to capture shared aspects."""

import numpy as np
import math


def clip(value, lower=0, upper=1):
    return max(lower, min(upper, value))


class ProgressMeasure:
    def __init__(self):
        self.status = 0
        self.history = [0]
        self.mastery_step = None
        self.name = self.__str__()

    def update(self, answer):
        self.status = clip(self.status)
        if self.status == 1 and self.mastery_step is None:
            self.mastery_step = len(self.history)
        # history stays 1 once reaches 1 (status may go down)
        # reason: history is used for visualizations, where this behavior is better
        v = self.status if self.history[-1] != 1 else 1
        self.history.append(v)

    def get_status(self):
        return round(self.status, 2)

    def is_mastered(self):
        return self.history and self.history[-1] == 1

    def reset(self):
        self.status = 0
        self.history = [0]

    def steps_to_mastery(self, answer_seq):
        self.reset()
        for i in range(len(answer_seq)):
            self.update(answer_seq[i])
            if self.is_mastered():
                return i+1
        return len(answer_seq)


class ProgressMeasureTA(ProgressMeasure):
    """Total Answers"""

    def __init__(self, n):
        self.n = n
        super().__init__()

    def __str__(self):
        return f"TA({self.n})"

    def update(self, answer):
        self.status = self.status + 1 / self.n
        super().update(answer)


class ProgressMeasureTC(ProgressMeasure):
    """Total Correct"""
    optimization_params = [("n", "int", 3, 50)]

    def __init__(self, n):
        self.n = n
        super().__init__()

    def __str__(self):
        return f"TC({self.n})"

    def update(self, answer):
        if answer:
            self.status = self.status + 1 / self.n
        super().update(answer)


class ProgressMeasureNCC(ProgressMeasure):
    """N Consecutive Correct"""
    optimization_params = [("n", "int", 3, 20)]

    def __init__(self, n):
        self.n = n
        self.counter = 0
        super().__init__()

    def __str__(self):
        return f"NCC({self.n})"

    def update(self, answer):
        if answer:
            self.counter += 1
            self.status = self.counter / self.n
        else:
            self.counter = 0
            self.status = 0
        super().update(answer)

    def reset(self):
        super().reset()
        self.counter = 0


class ProgressMeasureTOW(ProgressMeasure):
    """Tug Of War"""
    optimization_params = [("bonus", "float", 0, 0.2),
                           ("pen", "float", 0, 0.2)]

    def __init__(self, bonus, pen):
        self.bonus = bonus
        self.pen = pen
        super().__init__()

    def __str__(self):
        return f"TOW({self.bonus}, {self.pen})"

    def update(self, answer):
        if answer:
            step = self.bonus
        else:
            step = -self.pen
        self.status = self.status + step
        super().update(answer)


class ProgressMeasureEMA(ProgressMeasure):
    """Exponential Moving Average"""
    optimization_params = [("decay", "float", 0.5, 1),
                           ("threshold", "float", 0.8, 1)]

    def __init__(self, decay, threshold):
        self.decay = decay
        self.threshold = threshold
        self.skill = 0
        super().__init__()

    def __str__(self):
        return f"EMA({self.decay}, {self.threshold})"

    def update(self, answer):
        self.skill = self.decay * self.skill + (1-self.decay) * answer
        if self.skill > self.threshold:
            self.status = 1
        else:
            self.status = 1 - math.log((1-self.threshold) / (1-self.skill), 1-self.threshold)
        super().update(answer)

    def reset(self):
        super().reset()
        self.skill = 0

class ProgressMeasureVSLinear(ProgressMeasure):
    """Variable speed, linear skill-to-speed."""
    optimization_params = [("decay", "float", 0.5, 1),
                           ("slope", "float", 0, 2),
                           ("base_step", "float", 0, 0.2)]

    def __init__(self, base_step=0.05, guess_chance=0, slope=1, decay=0.9):
        self.base_step = base_step
        self.slope = slope
        self.decay = decay
        self.odds = guess_chance / (1 - guess_chance)
        self.skill = 0.5
        super().__init__()

    def reset(self):
        super().reset()
        self.skill = 0.5

    def __str__(self):
        return f"VS_Linear({self.base_step}, {self.slope}, {self.decay})"

    def update(self, answer):
        if answer:
            step = (1 + self.slope * (self.skill - 0.5)) * self.base_step
            performance = 1
        else:
            step = (1 - self.slope * (self.skill - 0.5)) * (-self.odds) * self.base_step
            performance = - self.odds
        self.status = self.status + step
        self.skill = self.decay * self.skill + (1 - self.decay) * performance
        super().update(answer)


class ProgressMeasureVSHMM(ProgressMeasure):
    """Variable speed, HMM model of skill, 3 states."""
    optimization_params = [("slip_chance", "float", 0, 0.25),
                           ("skill_change", "float", 0, 0.1),
                           ("high_speed", "float", 1, 2),
                           ("low_speed", "float", 0, 1),
                           ("base_step", "float", 0, 0.2)]

    def __init__(self, base_step=0.05, guess_chance=0.05, slip_chance=0.05, skill_change=0.05,
                 low_speed=0.5, high_speed=1.5):
        self.base_step = base_step
        self.low_speed = low_speed
        self.high_speed = high_speed
        self.guess_chance = guess_chance
        self.slip_chance = slip_chance
        self.skill_change = skill_change
        self.odds = guess_chance / (1 - guess_chance)
        self.state_prob = [0.25, 0.5, 0.25]
        self.skill = 0.5
        super().__init__()

    def reset(self):
        super().reset()
        self.state_prob = [0.25, 0.5, 0.25]

    def __str__(self):
        return f"VS_HMM({self.base_step}, {self.slip_chance}, {self.skill_change}, {self.low_speed}, {self.high_speed})"

    def prob(self, state, answer):
        if state == 0:
            correct_prob = self.guess_chance
        elif state == 1:
            correct_prob = 0.5 + 0.5 * (self.guess_chance - self.slip_chance)
        else:
            correct_prob = 1 - self.slip_chance
        if answer:
            return correct_prob
        return 1 - correct_prob

    def state_speed(self, state, answer):
        if answer:
            return [self.low_speed, 1, self.high_speed][state]
        else:
            return [-self.low_speed-self.odds, -self.odds, 0][state]

    def update(self, answer):
        new_state_prob = np.array([self.prob(state, answer) * self.state_prob[state] for state in range(3)])
        self.state_prob = new_state_prob / new_state_prob.sum()
        speed = 0
        for state in range(3):
            speed += self.state_prob[state] * self.state_speed(state, answer)
        self.status = self.status + speed * self.base_step
        self.skill = 0
        new_state_prob = self.state_prob.copy()
        for state in range(3):
            pred = self.state_prob[state-1] if state > 0 else 0
            suc = self.state_prob[state+1] if state < 2 else 0
            outflow = self.state_prob[state]
            if state == 1:
                outflow *= 2
            new_state_prob[state] += self.skill_change * (pred + suc - outflow)
            self.skill += new_state_prob[state] * (state / 2)
        self.state_prob = new_state_prob
        super().update(answer)