ska-referee-test/ref-test/app/tools/data.py

from ..models import Dataset, Test
from ..tools.logs import write

from flask import current_app as app
from flask.helpers import abort, flash, redirect, url_for

import json
from pathlib import Path
from random import shuffle
from statistics import mean, median, stdev
from typing import Union
from functools import wraps

def load(filename:str):
    data_dir = Path(app.config.get('DATA'))
    with open(f'./{data_dir}/{filename}') as file:
        return json.load(file)

def save(data:dict, filename:str):
    data_dir = Path(app.config.get('DATA'))
    with open(f'./{data_dir}/{filename}', 'w') as file:
        json.dump(data, file, indent=4)

def check_is_json(file):
    if not '.' in file.filename or not file.filename.rsplit('.',1)[-1] == 'json': return False
    return True

def validate_json(data):
    if not isinstance(data, list): return False
    for block in data:
        block_type = block.get('type', None)
        if block_type not in ['block', 'question']: return False
        if block_type == 'question':
            if not all (key in block for key in ['q_no', 'text', 'options', 'correct', 'q_type', 'tags']): return False
            if not isinstance(block['q_no'], int): return False
            if not isinstance(block['text'], str): return False
            if not isinstance(block['options'], list): return False
            for option in block['options']:
                if not isinstance(option, str): return False
            if not isinstance(block['correct'], int): return False
            if not isinstance(block['q_type'], str): return False
            if block['q_type'] not in ['Multiple Choice', 'Yes/No', 'List']: return False
            if not isinstance(block['tags'], list): return False
            for tag in block['tags']:
                if not isinstance(tag, str): return False
        if block_type == 'block':
            if not all (key in block for key in ['question_header', 'questions']): return False
            if not isinstance(block['question_header'], str): return False
            if not isinstance(block['questions'], list): return False
            for question in block['questions']:
                if not all (key in question for key in ['q_no', 'text', 'options', 'correct', 'q_type', 'tags']): return False
                if not isinstance(question['text'], str): return False
                if not isinstance(question['q_no'], int): return False
                if not isinstance(question['options'], list): return False
                for option in question['options']:
                    if not isinstance(option, str): return False
                if not isinstance(question['correct'], int): return False
                if not isinstance(question['q_type'], str): return False
                if question['q_type'] not in ['Multiple Choice', 'Yes/No', 'List']: return False
                if not isinstance(question['tags'], list): return False
                for tag in question['tags']:
                    if not isinstance(tag, str): return False
    return True

def randomise_list(list:list):
    _list = list.copy()
    shuffle(_list)
    return(_list)

def get_tag_list(dataset:list):
    output = []
    for block in dataset:
        if block['type'] == 'question': output = list(set(output) | set(block['tags']))
        if block['type'] == 'block':
            for question in block['questions']: output = list(set(output) | set(question['tags']))
    return output

def check_dataset_exists(function):
    @wraps(function)
    def wrapper(*args, **kwargs):
        try: datasets = Dataset.query.all()
        except Exception as exception:
            write('system.log', f'Database error when checking existing datasets: {exception}')
            return abort(500)
        if not datasets:
            flash('There are no available question datasets. Please upload a question dataset first, or use the question editor to create a new dataset.', 'error')
            return redirect(url_for('admin._questions'))
        return function(*args, **kwargs)
    return wrapper

def check_test_exists(function):
    @wraps(function)
    def wrapper(*args, **kwargs):
        try: tests = Test.query.all()
        except Exception as exception:
            write('system.log', f'Database error when checking existing datasets: {exception}')
            return abort(500)
        if not tests:
            flash('There are no exams configured. Please create an exam first.', 'error')
            return redirect(url_for('admin._tests'))
        return function(*args, **kwargs)
    return wrapper

def analyse(subject:Union[Dataset,Test]) -> dict:
    output = {
        'answers': {},
        'entries': 0,
        'grades': {
            'merit': 0,
            'pass': 0,
            'fail': 0
        },
        'scores': {
            'mean': 0,
            'median': 0,
            'stdev': 0
        }
    }
    scores_raw = []
    if isinstance(subject, Test):
        for entry in subject.entries:
            if entry.answers:
                for question, answer in entry.answers.items():
                    if int(question) not in output['answers']: output['answers'][int(question)] = {}
                    if int(answer) not in output['answers'][int(question)]: output['answers'][int(question)][int(answer)] = 0
                    output['answers'][int(question)][int(answer)] += 1
            if entry.result:
                output['entries'] += 1
                output['grades'][entry.result['grade']] += 1
                scores_raw.append(int(entry.result['score']))        
    else:
        for test in subject.tests:
            for entry in test.entries:
                if entry.answers:
                    for question, answer in entry.answers.items():
                        if int(question) not in output['answers']: output['answers'][int(question)] = {}
                        if int(answer) not in output['answers'][int(question)]: output['answers'][int(question)][int(answer)] = 0
                        output['answers'][int(question)][int(answer)] += 1
                if entry.result:
                    output['entries'] += 1
                    output['grades'][entry.result['grade']] += 1
                    scores_raw.append(entry.result['score'])
    output['scores']['mean'] = mean(scores_raw)
    output['scores']['median'] = median(scores_raw)
    output['scores']['stdev'] = stdev(scores_raw, output['scores']['mean']) if len(scores_raw) > 1 else None
    return output

def parse_questions(dataset:list):
    output = []
    for block in dataset:
        if block['type'] == 'question':
            question = {
                'q_no': block['q_no'],
                'tags': block['tags'],
                'correct': block['correct']
            }
            question['options'] = [*enumerate(block['options'])]
            output.append(question)
        elif block['type'] == 'block':
            for _question in block['questions']:
                question = {
                    'q_no': _question['q_no'],
                    'tags': _question['tags'],
                    'correct': _question['correct']
                }
                question['options'] = [*enumerate(_question['options'])]
                output.append(question)
    return output