recopytex/recopytex/store/filesystem/loader.py

#!/usr/bin/env python
# encoding: utf-8

import yaml
import os
from pathlib import Path
import pandas as pd
from .. import Loader
from .lib import *


DEFAULT_CONFIG_FILE = os.path.join(os.path.dirname(__file__), "default_config.yml")
with open(DEFAULT_CONFIG_FILE, "r") as config:
    DEFAULT_CONFIG = yaml.load(config, Loader=yaml.FullLoader)


class CSVLoader(Loader):

    """Loader when scores and metadatas are stored in csv files

    :config:

    :example:
    >>> loader = CSVLoader()
    >>> loader.get_config()
    {'source': './', 'competences': {'Chercher': {'name': 'Chercher', 'abrv': 'Cher'}, 'Représenter': {'name': 'Représenter', 'abrv': 'Rep'}, 'Modéliser': {'name': 'Modéliser', 'abrv': 'Mod'}, 'Raisonner': {'name': 'Raisonner', 'abrv': 'Rai'}, 'Calculer': {'name': 'Calculer', 'abrv': 'Cal'}, 'Communiquer': {'name': 'Communiquer', 'abrv': 'Com'}}, 'valid_scores': {'BAD': 0, 'FEW': 1, 'NEARLY': 2, 'GOOD': 3, 'NOTFILLED': None, 'NOANSWER': '.', 'ABS': 'a'}, 'csv_fields': {'term': 'Trimestre', 'exam': 'Nom', 'date': 'Date', 'exercise': 'Exercice', 'question': 'Question', 'competence': 'Competence', 'theme': 'Domaine', 'comment': 'Commentaire', 'score_rate': 'Bareme', 'is_leveled': 'Est_nivele'}}
    >>> loader = CSVLoader("./test_config.yml")
    >>> loader.get_config()
    {'source': './example', 'competences': {'Chercher': {'name': 'Chercher', 'abrv': 'Cher'}, 'Représenter': {'name': 'Représenter', 'abrv': 'Rep'}, 'Modéliser': {'name': 'Modéliser', 'abrv': 'Mod'}, 'Raisonner': {'name': 'Raisonner', 'abrv': 'Rai'}, 'Calculer': {'name': 'Calculer', 'abrv': 'Cal'}, 'Communiquer': {'name': 'Communiquer', 'abrv': 'Com'}}, 'valid_scores': {'BAD': 0, 'FEW': 1, 'NEARLY': 2, 'GOOD': 3, 'NOTFILLED': None, 'NOANSWER': '.', 'ABS': 'a'}, 'csv_fields': {'term': 'Trimestre', 'exam': 'Nom', 'date': 'Date', 'exercise': 'Exercice', 'question': 'Question', 'competence': 'Competence', 'theme': 'Domaine', 'comment': 'Commentaire', 'score_rate': 'Bareme', 'is_leveled': 'Est_nivele'}, 'output': './output', 'templates': 'templates/', 'tribes': {'Tribe1': {'name': 'Tribe1', 'type': 'Type1', 'students': 'tribe1.csv'}, 'Tribe2': {'name': 'Tribe2', 'students': 'tribe2.csv'}}}

    """

    CONFIG = DEFAULT_CONFIG

    def get_config(self):
        """ Get config """
        return self._config

    def rename_columns(self, dataframe):
        """Rename dataframe column to match with `csv_fields` """
        return dataframe.rename(columns=self._config["csv_fields"])

    def reverse_csv_field(self, keys):
        """ Reverse csv field from keys """
        return [self._config["csv_fields"][k] for k in keys]

    def get_tribes(self, only_names=False):
        """Get tribes list

        :example:
        >>> loader = CSVLoader("./test_config.yml")
        >>> loader.get_tribes()
        {'Tribe1': {'name': 'Tribe1', 'type': 'Type1', 'students': 'tribe1.csv'}, 'Tribe2': {'name': 'Tribe2', 'students': 'tribe2.csv'}}
        >>> loader.get_tribes(only_names=True)
        ['Tribe1', 'Tribe2']
        """
        if only_names:
            return list(self._config["tribes"].keys())
        return self._config["tribes"]

    def get_exams(self, tribes=[]):
        """Get exams list

        :param tribes: get only exams for those tribes
        :return: list of dictionaries of exams (fields: `["exam", "tribe", "date", "term", "score_file"]`)

        :example:
        >>> loader = CSVLoader("./test_config.yml")
        >>> loader.get_exams(["Tribe1"])
           Nom        Date  Trimestre                     score_file   tribe
        0   DS  12/01/2021          1   example/Tribe1/210112_DS.csv  Tribe1
        0  DS6  22/01/2021          1  example/Tribe1/210122_DS6.csv  Tribe1
        """
        exams = []
        for tribe in tribes:
            tribe_path = Path(self._config["source"]) / tribe
            csvs = list_csvs(tribe_path)
            for csv in csvs:
                fields = self.reverse_csv_field(["exam", "date", "term"])
                exam = extract_fields(csv, fields)
                exam = self.rename_columns(exam)
                exam["score_file"] = csv
                exam["tribe"] = tribe
                exams.append(exam)
        return pd.concat(exams)

    def get_students(self, tribes=[]):
        """Get student list

        :param tribes: concerned tribes
        """
        return ""

    def get_exam_questions(self, score_files=[]):
        """Get questions for exams stored in score_files

        :param exams: questions for those exams only

        :example:
        >>> loader = CSVLoader()
        >>> loader.get_exam_questions(["example/Tribe1/210112_DS.csv"])
             Exercice  Question Competence  ... Commentaire Bareme  Est_nivele
        0  Exercice 1         1   Calculer  ...      Coucou      1           1
        1  Exercice 1         2   Calculer  ...      Coucou      1           1
        2  Exercice 1         3   Calculer  ...      Coucou      1           1
        3  Exercice 1         3   Calculer  ...       DChic      1           1
        <BLANKLINE>
        [4 rows x 7 columns]
        """
        questions = []
        for score_file in score_files:
            fields = self.reverse_csv_field(
                [
                    "exercise",
                    "question",
                    "competence",
                    "theme",
                    "comment",
                    "score_rate",
                    "is_leveled",
                ]
            )
            question = extract_fields(score_file, fields)
            question = self.rename_columns(question)
            questions.append(question)

        return pd.concat(questions)

    def get_questions_scores(self, questions=[]):
        """Get scores of those questions

        :param questions: score for those questions
        """
        return ""

    def get_student_scores(self, student):
        """Get scores of the student

        :param student:
        """
        return ""