challenge.py

Go to the documentation of this file.

#!/usr/bin/python
# -*- coding: utf-8 -*- 
 
import json
import time
 
from statistics import median
 
from pafera.utils import *
 
import pafera.db
import pafera.modelbase
 
from pafera.validators import *
 
import apps.learn.challengeresult
import apps.learn.schoolclass
import apps.learn.student
import apps.learn.answer
 
# Challenge types
CHALLENGE_HOMEWORK            = 1
CHALLENGE_CLASSWORK           = 2
CHALLENGE_CLASSPARTICIPATION  = 3
CHALLENGE_QUIZ                = 4
CHALLENGE_TEST                = 5
CHALLENGE_EXAM                = 6
 
# Challenge flags
CHALLENGE_NEEDS_ANALYSIS    = 0x01
CHALLENGE_USE_STUDYLIST     = 0x02
 
# Challenge results
CHALLENGE_DIDNT_TRY         = 0
CHALLENGE_FAILED            = 1
CHALLENGE_COMPLETED         = 2
 
# *********************************************************************
class learn_challenge(pafera.modelbase.ModelBase):
  """In the Pafera Learning System, we handle homework and classwork 
  by issuing challenges to students. Points are then calculated based
  upon how the students did, and grades are automatically assigned by 
  the system. 
  
  There are a variety of challenges available, so check challenges.js 
  to see what they are and test them to see what you can do with them. 
  
  Problems for challenges can be assigned by adding lessons, individual
  problems, or can be added straight from their study list. 
  
  Upon a challenge ending, the next time that you visit /learn/home.html,
  student scores will be automatically calculated and applied to their
  grades. You can then see such statistics as the average score, what
  percentage of students passed, which problems were often answered
  incorrectly, and such in the analysis section.
  """
  
  _dbfields     = {
    'rid':                ('INTEGER', 'PRIMARY KEY NOT NULL',),
    'classid':            ('INT', 'NOT NULL', BlankValidator()),
    'starttime':          ('DATETIME', 'NOT NULL',),
    'endtime':            ('DATETIME', 'NOT NULL',),
    'challenge':          ('TEXT', 'NOT NULL', BlankValidator()),
    'title':              ('TRANSLATION', 'NOT NULL', BlankValidator()),
    'instructions':       ('TRANSLATION', 'NOT NULL',),
    'question':           ('TRANSLATION', 'NOT NULL',),
    'image':              ('IMAGEFILE', 'NOT NULL',),
    'sound':              ('SOUNDFILE', 'NOT NULL',),
    'video':              ('VIDEOFILE', 'NOT NULL',),
    'files':              ('NEWLINELIST', "NOT NULL DEFAULT ''",),
    'lessontitles':       ('NEWLINELIST', "NOT NULL DEFAULT ''",),
    'lessonids':          ('NEWLINELIST', "NOT NULL DEFAULT ''",),
    'problemids':         ('NEWLINELIST', "NOT NULL DEFAULT ''",),
    'results':            ('DICT', "NOT NULL DEFAULT ''",),
    'analysis':           ('DICT', "NOT NULL DEFAULT ''",),
    'averagescore':       ('INT16', 'NOT NULL DEFAULT 0',),
    'averageright':       ('INT16', 'NOT NULL DEFAULT 0',),
    'averagepercent':     ('INT16', 'NOT NULL DEFAULT 0',),
    'percenttried':       ('INT16', 'NOT NULL DEFAULT 0',),
    'percentcomplete':    ('INT16', 'NOT NULL DEFAULT 0',),
    'minscore':           ('INT16', 'NOT NULL DEFAULT 0',),
    'minpercentage':      ('INT16', 'NOT NULL DEFAULT 0',),
    'numpoints':          ('INT16', 'NOT NULL', BlankValidator()),
    'didnttrypenalty':    ('INT16', 'NOT NULL DEFAULT 0', ),
    'numtries':           ('INT16', 'NOT NULL', BlankValidator()),
    'numproblems':        ('INT16', 'NOT NULL DEFAULT 0', ),
    'timelimit':          ('INT16', 'NOT NULL DEFAULT 0',),
    'challengetype':      ('INT16', 'NOT NULL DEFAULT 0',),
    'flags':              ('INT16', 'NOT NULL DEFAULT 0',),
  }
  _dbindexes  = ()
  _dblinks    = [
  ]
  _dbdisplay  = [
    'classid', 
    'challenge', 
    'lessontitles', 
    'problemids', 
    'minscore', 
    'minpercentage',
    'numpoints'
  ]
  _dbflags    = 0
  
  # -------------------------------------------------------------------
  def __init__(self):
    super().__init__()
    
  # -------------------------------------------------------------------
  def GetNumRemainingAttempts(self, g):
    """Returns the number of remaining tries available to the current 
    user.
    
    As a side effect, all previous results will be available in 
    obj.previousresults as learn_challengeresult objects.
    """
    if not hasattr(self, 'previousresults'):
      self.previousresults = g.db.Find(
        apps.learn.challengeresult.learn_challengeresult,
        'WHERE challengeid = ? AND userid = ?',
        [
          self.rid, 
          g.session.userid
        ],
        fields    = 'rid, problemresults, score, percentright',
        orderby   = 'endtime',
      )
    
    return Bound(self.numtries - len(self.previousresults), 0, 9999)
  
  # -------------------------------------------------------------------
  def Analyze(self, g):
    """Calculates all scores and saves statistics for this challenge.
    
    Typically, this will be done automatically for you after endtime 
    has passed.
    """
    
    # Points for Write Answer challenges are manually assigned by the 
    # teacher, so no statistics or points need to be calculated.
    if self.challenge == 'Write Answer':
      return
    
    cls       = g.db.Load(apps.learn.schoolclass.learn_schoolclass, self.classid)    
    
    currenttime = time.time()
    
    self.results  = {}      
    studentnames  = {}
    wrongproblems = {}
    answers       = {}    
    
    # Classwork scores are automatically applied to the current answer object
    if (self.challengetype == CHALLENGE_CLASSWORK 
        or self.challengetype == CHALLENGE_CLASSPARTICIPATION
        or self.challenge == 'Typing'
      ):
      for r in g.db.Find(
          apps.learn.answer.learn_answer,
          'WHERE classid = ? AND NOT (flags & ?)',
          [
            self.classid,
            apps.learn.answer.ANSWER_TEACHER
          ],
          fields  = 'rid, userid, displayname, numright, numwrong, bonusscore, score'
        ):
        
        answers[r.userid] = r        
    
    # Class participation scores are calculated from the current answer
    # object
    if self.challengetype == CHALLENGE_CLASSPARTICIPATION:
      for k, v in answers.items():
        o = v.ToJSON('displayname, numright, numwrong, bonusscore, score')
        
        if v.numright + v.numwrong:
          o['percentage'] = int(v.numright / (v.numright + v.numwrong) * 100)
        else:
          o['percentage'] = 0
        
        self.results[ToShortCode(k)]  = o
        
      self.Set(results = self.results)
      
      if time.time() > self.endtime.timestamp():
        self.Set(flags = self.flags & (~CHALLENGE_NEEDS_ANALYSIS))
      
      g.db.Update(self)
      
      for k, v in answers.items():
        cls.UpdateGrades(g, k)
        
      g.db.Commit()
      return self      
    
    # Get all students for the class and see if any failed to even attempt the challenge
    for r in g.db.Linked(cls, apps.learn.student.learn_student, fields = 'userid, displayname'):
      studentid = ToShortCode(r.userid)
      
      studentnames[r.userid]  = r.displayname
      
      if studentid not in self.results:
        self.results[studentid] = [r.displayname, CHALLENGE_DIDNT_TRY, 0, 0, -self.didnttrypenalty]
    
    # Iterate through all results and calculate statistics
    for r in g.db.Find(
        apps.learn.challengeresult.learn_challengeresult,
        'WHERE challengeid = ?',
        self.rid
      ):
      
      studentid = ToShortCode(r.userid)
      
      if r.userid not in studentnames:
        studentnames[r.userid]  = '[Removed Student ' + studentid + ']'
        self.results[studentid] = [studentnames[r.userid], CHALLENGE_DIDNT_TRY, r.score, r.percentright, -self.didnttrypenalty]
      
      if r.flags & apps.learn.challengeresult.CHALLENGERESULT_PASSED:
        self.results[studentid] = [studentnames[r.userid], CHALLENGE_COMPLETED, r.score, r.percentright, self.numpoints]
      elif self.results[studentid][1] == CHALLENGE_DIDNT_TRY or self.results[studentid][2] > r.score:
        self.results[studentid] = [studentnames[r.userid], CHALLENGE_FAILED, r.score, r.percentright, 0]
      
      if r.problemresults:
        for p in r.problemresults:
          if 'id' in p:
            if p['status'] == 'wrong':
              if p['id'] not in wrongproblems:
                problemtext = '[Text not found]'
                
                try:
                  problemtext = g.db.Execute(f'''
                      SELECT content
                      FROM learn_problem
                      JOIN learn_card ON learn_problem.problemid = learn_card.rid
                      WHERE learn_problem.rid = ?
                    ''',
                    FromShortCode(p['id'])
                  )[0]
                except Exception as e:
                  print('>>> Problem getting problem text', e)
                
                wrongproblems[ p['id'] ]  = {
                    'problemtext':  problemtext,
                    'studentids':   [],
                  }
                
              wrongproblems[ p['id'] ]['studentids'].append(ToShortCode(r.userid))            
    
    # Apply penalties if someone didn't complete their classwork
    if self.challengetype == CHALLENGE_CLASSWORK and self.didnttrypenalty and self.endtime.timestamp() < currenttime:
      for k, v in self.results.items():
        if v[1] == CHALLENGE_DIDNT_TRY:
          userid  = FromShortCode(k)
          
          if userid not in answers:
            if userid not in studentnames:
              studentnames[userid]  = {
                'en': f'[Removed Student {k}]'
              }
            
            newanswer = apps.learn.answer.learn_answer()
            newanswer.Set(
              classid     = self.classid,
              userid      = userid,
              displayname = studentnames[userid],
            )
            
            newanswer.AddHomeworkScores(g)
            
            g.db.Insert(newanswer)
            g.db.Commit()
            
            answers[userid] = newanswer
            
          answerobj = answers[userid]
          
          answerobj.Set(
            bonusscore  = answerobj.bonusscore - self.didnttrypenalty,
          )
          
          answerobj.UpdateScore()
          
          g.db.Update(answerobj)
    
    self.Set(
      results   = self.results,
      analysis  = {
          'wrongproblems':  wrongproblems
      }
    )
    
    # Calculate averages 
    numdidnttry     = 0
    numfailed       = 0
    numcompleted    = 0
    averagescore    = []
    averagepercent  = []
    
    for v in self.results.values():
      if v[1] == CHALLENGE_DIDNT_TRY:
        numdidnttry += 1
      else:
        numfailed   += 1
        
        averagescore.append(v[2])
        averagepercent.append(v[3])
        
        if v[1] == CHALLENGE_COMPLETED:
          numcompleted  += 1
    
    totalids  = numdidnttry + numfailed + numcompleted
    
    percenttried     = numfailed / totalids * 100
    percentcomplete  = numcompleted / totalids * 100
    
    self.Set(
      percenttried      = percenttried,
      percentcomplete   = percentcomplete,
      averagescore      = median(averagescore) if averagescore else 0,
      averagepercent    = median(averagepercent) if averagepercent else 0,
    ) 
    
    # If endtime has passed, then no further data will be added and we don't
    # need to do this again
    if currenttime > self.endtime.timestamp():
      self.Set(flags  = self.flags & (~CHALLENGE_NEEDS_ANALYSIS))
      
    g.db.Update(self)
 
# =====================================================================
def GetCurrentChallenges(g, classid, challengetype = CHALLENGE_HOMEWORK):
  """Returns a list of all currently running challenges for the logged in user.
  
  Useful for seeing what this student has finished and what they still 
  need to do. 
  """
  
  challenges  = []
  currenttime = time.time()
  
  for r in g.db.Find(
      learn_challenge,
      'WHERE classid = ? AND starttime < ? AND endtime > ? AND challengetype = ?',
      [
        classid,
        currenttime,
        currenttime,
        challengetype,
      ],
      fields  = 'rid, challenge, title, endtime, numpoints, numtries, flags',
    ):
    
    completed = 0
    
    results = g.db.Find(
      apps.learn.learn_challengeresult,
      'WHERE challengeid = ? AND userid = ?',
      [r.rid, g.session.userid],
      fields  = 'rid, flags'
    )
      
    for s in results:
      if s.flags & apps.learn.challengeresult.CHALLENGERESULT_PASSED:
        completed = 1
        break
    
    c = r.ToJSON('rid, challenge, title, endtime, numpoints, numtries, flags')
    
    c['triesleft']      = Bound(r.numtries - len(results), 0, 9999)
    c['completed']      = completed
    c['timeremaining']  = int(r.endtime.timestamp() - currenttime)
    
    challenges.append(c)
    
  return challenges