# python standard library import random from datetime import datetime import logging # libraries import networkx as nx # this project # import questions # setup logger for this module logger = logging.getLogger(__name__) # ---------------------------------------------------------------------------- # kowledge state of each student....?? # Contains: # state - dict of topics with state of unlocked topics # deps - access to dependency graph shared between students # topic_sequence - list with the order of recommended topics # ---------------------------------------------------------------------------- class StudentKnowledge(object): # ======================================================================= # methods that update state # ======================================================================= def __init__(self, deps, factory, state={}): self.deps = deps # dependency graph shared among students self.factory = factory # question factory self.state = state # {'topic': {'level':0.5, 'date': datetime}, ...} self.update_topic_levels() # applies forgetting factor self.topic_sequence = self.recommend_topic_sequence() # ['a', 'b', ...] self.unlock_topics() # whose dependencies have been completed self.current_topic = None # ------------------------------------------------------------------------ # Updates the proficiency levels of the topics, with forgetting factor # FIXME no dependencies are considered yet... # ------------------------------------------------------------------------ def update_topic_levels(self): now = datetime.now() for tref, s in self.state.items(): dt = now - s['date'] s['level'] *= 0.95 ** dt.days # forgetting factor 0.95 FIXME # ------------------------------------------------------------------------ # Unlock topics whose dependencies are satisfied (> min_level) # ------------------------------------------------------------------------ def unlock_topics(self): # minimum level that the dependencies of a topic must have # for the topic to be unlocked. min_level = 0.01 for topic in self.topic_sequence: if topic not in self.state: # if locked pred = self.deps.predecessors(topic) if all(d in self.state and self.state[d]['level'] > min_level for d in pred): # all dependencies are done self.state[topic] = { 'level': 0.0, # unlocked 'date': datetime.now() } logger.debug(f'Unlocked "{topic}".') # ------------------------------------------------------------------------ # Start a new topic. # questions: list of generated questions to do in the topic # current_question: the current question to be presented # ------------------------------------------------------------------------ # FIXME async mas nao tem awaits... async def start_topic(self, topic): logger.debug(f'StudentKnowledge.start_topic({topic})') # do not allow locked topics if self.is_locked(topic): logger.debug(f'Topic {topic} is locked') return False # starting new topic self.current_topic = topic self.correct_answers = 0 self.wrong_answers = 0 t = self.deps.node[topic] k = t['choose'] if t['shuffle']: questions = random.sample(t['questions'], k=k) else: questions = t['questions'][:k] logger.debug(f'Questions: {", ".join(questions)}') # generate instances of questions gen = lambda qref: self.factory[qref].generate() self.questions = [gen(qref) for qref in questions] # self.questions = [gen(qref) for qref in questions] logger.debug(f'Total: {len(self.questions)} questions') # get first question self.next_question() return True # ------------------------------------------------------------------------ # The topic has finished and there are no more questions. # The topic level is updated in state and unlocks are performed. # The current topic is unchanged. # ------------------------------------------------------------------------ def finish_topic(self): logger.debug(f'StudentKnowledge.finish_topic({self.current_topic})') self.state[self.current_topic] = { 'date': datetime.now(), 'level': self.correct_answers / (self.correct_answers + self.wrong_answers) } self.unlock_topics() # ------------------------------------------------------------------------ # corrects current question with provided answer. # implements the logic: # - if answer ok, goes to next question # - if wrong, counts number of tries. If exceeded, moves on. # ------------------------------------------------------------------------ async def check_answer(self, answer): logger.debug('StudentKnowledge.check_answer()') q = self.current_question q['answer'] = answer q['finish_time'] = datetime.now() grade = await q.correct_async() logger.debug(f'Grade {grade:.2} ({q["ref"]})') if grade > 0.999: self.correct_answers += 1 self.next_question() action = 'right' else: self.wrong_answers += 1 self.current_question['tries'] -= 1 logger.debug(f'Wrong answers = {self.wrong_answers}; Tries = {self.current_question["tries"]}') if self.current_question['tries'] > 0: action = 'try_again' else: action = 'wrong' if self.current_question['append_wrong']: logger.debug("Appending new instance of this question to the end") self.questions.append(self.factory[q['ref']].generate()) self.next_question() # returns corrected question (not new one) which might include comments return q, action # ------------------------------------------------------------------------ # Move to next question # ------------------------------------------------------------------------ def next_question(self): try: self.current_question = self.questions.pop(0) except IndexError: self.current_question = None self.finish_topic() else: self.current_question['start_time'] = datetime.now() default_maxtries = self.deps.nodes[self.current_topic]['max_tries'] self.current_question['tries'] = self.current_question.get('max_tries', default_maxtries) logger.debug(f'Next question is "{self.current_question["ref"]}"') return self.current_question # question or None # ======================================================================== # pure functions of the state (no side effects) # ======================================================================== def topic_has_finished(self): return self.current_question is None # ------------------------------------------------------------------------ # compute recommended sequence of topics ['a', 'b', ...] # ------------------------------------------------------------------------ def recommend_topic_sequence(self): return list(nx.topological_sort(self.deps)) # ------------------------------------------------------------------------ def get_current_question(self): return self.current_question # ------------------------------------------------------------------------ def get_current_topic(self): return self.current_topic # ------------------------------------------------------------------------ def is_locked(self, topic): return topic not in self.state # ------------------------------------------------------------------------ # Return list of {ref: 'xpto', name: 'long name', leve: 0.5} # Levels are in the interval [0, 1] if unlocked or None if locked. # Topics unlocked but not yet done have level 0.0. # ------------------------------------------------------------------------ def get_knowledge_state(self): # print(self.topic_sequence) return [{ 'ref': ref, 'type': self.deps.nodes[ref]['type'], 'name': self.deps.nodes[ref]['name'], 'level': self.state[ref]['level'] if ref in self.state else None } for ref in self.topic_sequence ] # ------------------------------------------------------------------------ def get_topic_progress(self): return self.correct_answers / (1 + self.correct_answers + len(self.questions)) # ------------------------------------------------------------------------ def get_topic_level(self, topic): return self.state[topic]['level'] # ------------------------------------------------------------------------ def get_topic_date(self, topic): return self.state[topic]['date'] # ------------------------------------------------------------------------ # Recommends a topic to practice/learn from the state. # ------------------------------------------------------------------------ # def get_recommended_topic(self): # FIXME untested # return min(self.state.items(), key=lambda x: x[1]['level'])[0]