Feat: Start filesystem loader
This commit is contained in:
@@ -1,207 +0,0 @@
|
||||
#!/usr/bin/env python
|
||||
# encoding: utf-8
|
||||
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from prompt_toolkit import HTML
|
||||
from ..config import NO_ST_COLUMNS
|
||||
import pandas as pd
|
||||
import yaml
|
||||
from .getconfig import config
|
||||
|
||||
|
||||
def try_parsing_date(text, formats=["%Y-%m-%d", "%Y.%m.%d", "%Y/%m/%d"]):
|
||||
for fmt in formats:
|
||||
try:
|
||||
return datetime.strptime(text[:10], fmt)
|
||||
except ValueError:
|
||||
pass
|
||||
raise ValueError("no valid date format found")
|
||||
|
||||
|
||||
def format_question(question):
|
||||
question["score_rate"] = float(question["score_rate"])
|
||||
return question
|
||||
|
||||
|
||||
class Exam:
|
||||
def __init__(self, name, tribename, date, term, **kwrds):
|
||||
self._name = name
|
||||
self._tribename = tribename
|
||||
|
||||
self._date = try_parsing_date(date)
|
||||
|
||||
self._term = term
|
||||
|
||||
try:
|
||||
kwrds["exercices"]
|
||||
except KeyError:
|
||||
self._exercises = {}
|
||||
else:
|
||||
self._exercises = kwrds["exercices"]
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def tribename(self):
|
||||
return self._tribename
|
||||
|
||||
@property
|
||||
def date(self):
|
||||
return self._date
|
||||
|
||||
@property
|
||||
def term(self):
|
||||
return self._term
|
||||
|
||||
def add_exercise(self, name, questions):
|
||||
""" Add key with questions in ._exercises """
|
||||
try:
|
||||
self._exercises[name]
|
||||
except KeyError:
|
||||
self._exercises[name] = format_question(questions)
|
||||
else:
|
||||
raise KeyError("The exercise already exsists. Use modify_exercise")
|
||||
|
||||
def modify_exercise(self, name, questions, append=False):
|
||||
"""Modify questions of an exercise
|
||||
|
||||
If append==True, add questions to the exercise questions
|
||||
|
||||
"""
|
||||
try:
|
||||
self._exercises[name]
|
||||
except KeyError:
|
||||
raise KeyError("The exercise already exsists. Use modify_exercise")
|
||||
else:
|
||||
if append:
|
||||
self._exercises[name] += format_question(questions)
|
||||
else:
|
||||
self._exercises[name] = format_question(questions)
|
||||
|
||||
@property
|
||||
def exercices(self):
|
||||
return self._exercises
|
||||
|
||||
@property
|
||||
def tribe_path(self):
|
||||
return Path(config["source"]) / self.tribename
|
||||
|
||||
@property
|
||||
def tribe_student_path(self):
|
||||
return (
|
||||
Path(config["source"])
|
||||
/ [t["students"] for t in config["tribes"] if t["name"] == self.tribename][
|
||||
0
|
||||
]
|
||||
)
|
||||
|
||||
@property
|
||||
def long_name(self):
|
||||
""" Get exam name with date inside """
|
||||
return f"{self.date.strftime('%y%m%d')}_{self.name}"
|
||||
|
||||
def path(self, extention=""):
|
||||
return self.tribe_path / (self.long_name + extention)
|
||||
|
||||
def to_dict(self):
|
||||
return {
|
||||
"name": self.name,
|
||||
"tribename": self.tribename,
|
||||
"date": self.date,
|
||||
"term": self.term,
|
||||
"exercices": self.exercices,
|
||||
}
|
||||
|
||||
def to_row(self):
|
||||
rows = []
|
||||
for ex, questions in self.exercices.items():
|
||||
for q in questions:
|
||||
rows.append(
|
||||
{
|
||||
"term": self.term,
|
||||
"assessment": self.name,
|
||||
"date": self.date.strftime("%d/%m/%Y"),
|
||||
"exercise": ex,
|
||||
"question": q["id"],
|
||||
**q,
|
||||
}
|
||||
)
|
||||
return rows
|
||||
|
||||
@property
|
||||
def themes(self):
|
||||
themes = set()
|
||||
for questions in self._exercises.values():
|
||||
themes.update([q["theme"] for q in questions])
|
||||
return themes
|
||||
|
||||
def display_exercise(self, name):
|
||||
pass
|
||||
|
||||
def display(self, name):
|
||||
pass
|
||||
|
||||
def write_yaml(self):
|
||||
print(f"Sauvegarde temporaire dans {self.path('.yml')}")
|
||||
self.tribe_path.mkdir(exist_ok=True)
|
||||
with open(self.path(".yml"), "w") as f:
|
||||
f.write(yaml.dump(self.to_dict()))
|
||||
|
||||
def write_csv(self):
|
||||
rows = self.to_row()
|
||||
|
||||
base_df = pd.DataFrame.from_dict(rows)[NO_ST_COLUMNS.keys()]
|
||||
base_df.rename(columns=NO_ST_COLUMNS, inplace=True)
|
||||
|
||||
students = pd.read_csv(self.tribe_student_path)["Nom"]
|
||||
for student in students:
|
||||
base_df[student] = ""
|
||||
|
||||
self.tribe_path.mkdir(exist_ok=True)
|
||||
base_df.to_csv(self.path(".csv"), index=False)
|
||||
|
||||
@property
|
||||
def score_rate(self):
|
||||
total = 0
|
||||
for ex, questions in self._exercises.items():
|
||||
total += sum([q["score_rate"] for q in questions])
|
||||
|
||||
return total
|
||||
|
||||
@property
|
||||
def competences_rate(self):
|
||||
""" Dictionnary with competences as key and total rate as value"""
|
||||
rates = {}
|
||||
for ex, questions in self._exercises.items():
|
||||
for q in questions:
|
||||
try:
|
||||
q["competence"]
|
||||
except KeyError:
|
||||
pass
|
||||
else:
|
||||
try:
|
||||
rates[q["competence"]] += q["score_rate"]
|
||||
except KeyError:
|
||||
rates[q["competence"]] = q["score_rate"]
|
||||
return rates
|
||||
|
||||
@property
|
||||
def themes_rate(self):
|
||||
""" Dictionnary with themes as key and total rate as value"""
|
||||
rates = {}
|
||||
for ex, questions in self._exercises.items():
|
||||
for q in questions:
|
||||
try:
|
||||
q["theme"]
|
||||
except KeyError:
|
||||
pass
|
||||
else:
|
||||
if q["theme"]:
|
||||
try:
|
||||
rates[q["theme"]] += q["score_rate"]
|
||||
except KeyError:
|
||||
rates[q["theme"]] = q["score_rate"]
|
||||
return rates
|
||||
@@ -1,9 +0,0 @@
|
||||
#!/usr/bin/env python
|
||||
# encoding: utf-8
|
||||
import yaml
|
||||
|
||||
CONFIGPATH = "recoconfig.yml"
|
||||
|
||||
with open(CONFIGPATH, "r") as config:
|
||||
config = yaml.load(config, Loader=yaml.FullLoader)
|
||||
|
||||
@@ -1,233 +0,0 @@
|
||||
#!/usr/bin/env python
|
||||
# encoding: utf-8
|
||||
|
||||
|
||||
from prompt_toolkit import prompt, HTML, ANSI
|
||||
from prompt_toolkit import print_formatted_text as print
|
||||
from prompt_toolkit.styles import Style
|
||||
from prompt_toolkit.validation import Validator
|
||||
from prompt_toolkit.completion import WordCompleter
|
||||
from unidecode import unidecode
|
||||
from datetime import datetime
|
||||
from functools import wraps
|
||||
import sys
|
||||
|
||||
from .getconfig import config
|
||||
|
||||
|
||||
VALIDATE = [
|
||||
"o",
|
||||
"ok",
|
||||
"OK",
|
||||
"oui",
|
||||
"OUI",
|
||||
"yes",
|
||||
"YES",
|
||||
]
|
||||
REFUSE = ["n", "non", "NON", "no", "NO"]
|
||||
CANCEL = ["a", "annuler"]
|
||||
|
||||
STYLE = Style.from_dict(
|
||||
{
|
||||
"": "#93A1A1",
|
||||
"validation": "#884444",
|
||||
"appending": "#448844",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
class CancelError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def prompt_validate(question, cancelable=False, empty_means=1, style="validation"):
|
||||
"""Prompt for validation
|
||||
|
||||
:param question: Text to print to ask the question.
|
||||
:param cancelable: enable cancel answer
|
||||
:param empty_means: result for no answer
|
||||
:return:
|
||||
0 -> Refuse
|
||||
1 -> Validate
|
||||
-1 -> cancel
|
||||
"""
|
||||
question_ = question
|
||||
choices = VALIDATE + REFUSE
|
||||
|
||||
if cancelable:
|
||||
question_ += "(a ou annuler pour sortir)"
|
||||
choices += CANCEL
|
||||
|
||||
ans = prompt(
|
||||
[
|
||||
(f"class:{style}", question_),
|
||||
],
|
||||
completer=WordCompleter(choices),
|
||||
style=STYLE,
|
||||
).lower()
|
||||
|
||||
if ans == "":
|
||||
return empty_means
|
||||
if ans in VALIDATE:
|
||||
return 1
|
||||
if cancelable and ans in CANCEL:
|
||||
return -1
|
||||
return 0
|
||||
|
||||
|
||||
def prompt_until_validate(question="C'est ok? ", cancelable=False):
|
||||
def decorator(func):
|
||||
@wraps(func)
|
||||
def wrapper(*args, **kwrd):
|
||||
ans = func(*args, **kwrd)
|
||||
|
||||
confirm = prompt_validate(question, cancelable)
|
||||
|
||||
if confirm == -1:
|
||||
raise CancelError
|
||||
|
||||
while not confirm:
|
||||
sys.stdout.flush()
|
||||
ans = func(*args, **ans, **kwrd)
|
||||
confirm = prompt_validate(question, cancelable)
|
||||
if confirm == -1:
|
||||
raise CancelError
|
||||
return ans
|
||||
|
||||
return wrapper
|
||||
|
||||
return decorator
|
||||
|
||||
|
||||
@prompt_until_validate()
|
||||
def prompt_exam(**kwrd):
|
||||
""" Prompt questions to edit an exam """
|
||||
print(HTML("<b>Nouvelle évaluation</b>"))
|
||||
exam = {}
|
||||
exam["name"] = prompt("Nom de l'évaluation: ", default=kwrd.get("name", "DS"))
|
||||
|
||||
tribes_name = [t["name"] for t in config["tribes"]]
|
||||
|
||||
exam["tribename"] = prompt(
|
||||
"Nom de la classe: ",
|
||||
default=kwrd.get("tribename", ""),
|
||||
completer=WordCompleter(tribes_name),
|
||||
validator=Validator.from_callable(lambda x: x in tribes_name),
|
||||
)
|
||||
exam["tribe"] = [t for t in config["tribes"] if t["name"] == exam["tribename"]][0]
|
||||
|
||||
exam["date"] = prompt(
|
||||
"Date de l'évaluation (%y%m%d): ",
|
||||
default=kwrd.get("date", datetime.today()).strftime("%y%m%d"),
|
||||
validator=Validator.from_callable(lambda x: (len(x) == 6) and x.isdigit()),
|
||||
)
|
||||
exam["date"] = datetime.strptime(exam["date"], "%y%m%d")
|
||||
|
||||
exam["term"] = prompt(
|
||||
"Trimestre: ",
|
||||
validator=Validator.from_callable(lambda x: x.isdigit()),
|
||||
default=kwrd.get("term", "1"),
|
||||
)
|
||||
|
||||
return exam
|
||||
|
||||
|
||||
@prompt_until_validate()
|
||||
def prompt_exercise(number=1, completer={}, **kwrd):
|
||||
exercise = {}
|
||||
try:
|
||||
kwrd["name"]
|
||||
except KeyError:
|
||||
print(HTML("<b>Nouvel exercice</b>"))
|
||||
exercise["name"] = prompt(
|
||||
"Nom de l'exercice: ", default=kwrd.get("name", f"Exercice {number}")
|
||||
)
|
||||
else:
|
||||
print(HTML(f"<b>Modification de l'exercice: {kwrd['name']}</b>"))
|
||||
exercise["name"] = kwrd["name"]
|
||||
|
||||
exercise["questions"] = []
|
||||
|
||||
try:
|
||||
kwrd["questions"][0]
|
||||
except KeyError:
|
||||
last_question_id = "1a"
|
||||
except IndexError:
|
||||
last_question_id = "1a"
|
||||
else:
|
||||
for ques in kwrd["questions"]:
|
||||
try:
|
||||
exercise["questions"].append(
|
||||
prompt_question(completer=completer, **ques)
|
||||
)
|
||||
except CancelError:
|
||||
print("Cette question a été supprimée")
|
||||
last_question_id = exercise["questions"][-1]["id"]
|
||||
|
||||
appending = prompt_validate(
|
||||
question="Ajouter un élément de notation? ", style="appending"
|
||||
)
|
||||
while appending:
|
||||
try:
|
||||
exercise["questions"].append(
|
||||
prompt_question(last_question_id, completer=completer)
|
||||
)
|
||||
except CancelError:
|
||||
print("Cette question a été supprimée")
|
||||
else:
|
||||
last_question_id = exercise["questions"][-1]["id"]
|
||||
appending = prompt_validate(
|
||||
question="Ajouter un élément de notation? ", style="appending"
|
||||
)
|
||||
|
||||
return exercise
|
||||
|
||||
|
||||
@prompt_until_validate(cancelable=True)
|
||||
def prompt_question(last_question_id="1a", completer={}, **kwrd):
|
||||
try:
|
||||
kwrd["id"]
|
||||
except KeyError:
|
||||
print(HTML("<b>Nouvel élément de notation</b>"))
|
||||
else:
|
||||
print(
|
||||
HTML(f"<b>Modification de l'élément {kwrd['id']} ({kwrd['comment']})</b>")
|
||||
)
|
||||
|
||||
question = {}
|
||||
question["id"] = prompt(
|
||||
"Identifiant de la question: ",
|
||||
default=kwrd.get("id", "1a"),
|
||||
)
|
||||
|
||||
question["competence"] = prompt(
|
||||
"Competence: ",
|
||||
default=kwrd.get("competence", list(config["competences"].keys())[0]),
|
||||
completer=WordCompleter(config["competences"].keys()),
|
||||
validator=Validator.from_callable(lambda x: x in config["competences"].keys()),
|
||||
)
|
||||
|
||||
question["theme"] = prompt(
|
||||
"Domaine: ",
|
||||
default=kwrd.get("theme", ""),
|
||||
completer=WordCompleter(completer.get("theme", [])),
|
||||
)
|
||||
|
||||
question["comment"] = prompt(
|
||||
"Commentaire: ",
|
||||
default=kwrd.get("comment", ""),
|
||||
)
|
||||
|
||||
question["is_leveled"] = prompt(
|
||||
"Évaluation par niveau: ",
|
||||
default=kwrd.get("is_leveled", "1"),
|
||||
# validate
|
||||
)
|
||||
|
||||
question["score_rate"] = prompt(
|
||||
"Barème: ",
|
||||
default=kwrd.get("score_rate", "1"),
|
||||
# validate
|
||||
)
|
||||
|
||||
return question
|
||||
@@ -1,134 +0,0 @@
|
||||
#!/usr/bin/env python
|
||||
# encoding: utf-8
|
||||
|
||||
import click
|
||||
from pathlib import Path
|
||||
import sys
|
||||
import papermill as pm
|
||||
import pandas as pd
|
||||
from datetime import datetime
|
||||
import yaml
|
||||
|
||||
from .getconfig import config, CONFIGPATH
|
||||
from .prompts import prompt_exam, prompt_exercise, prompt_validate
|
||||
from ..config import NO_ST_COLUMNS
|
||||
from .exam import Exam
|
||||
from ..dashboard.index import app as dash
|
||||
|
||||
|
||||
@click.group()
|
||||
def cli():
|
||||
pass
|
||||
|
||||
|
||||
@cli.command()
|
||||
def print_config():
|
||||
click.echo(f"Config file is {CONFIGPATH}")
|
||||
click.echo("It contains")
|
||||
click.echo(config)
|
||||
|
||||
|
||||
@cli.command()
|
||||
def setup():
|
||||
"""Setup the environnement using recoconfig.yml"""
|
||||
for tribe in config["tribes"]:
|
||||
Path(tribe["name"]).mkdir(exist_ok=True)
|
||||
if not Path(tribe["students"]).exists():
|
||||
print(f"The file {tribe['students']} does not exists")
|
||||
|
||||
|
||||
@cli.command()
|
||||
def new_exam():
|
||||
""" Create new exam csv file """
|
||||
exam = Exam(**prompt_exam())
|
||||
|
||||
if exam.path(".yml").exists():
|
||||
print(f"Fichier sauvegarde trouvé à {exam.path('.yml')} -- importation")
|
||||
with open(exam.path(".yml"), "r") as f:
|
||||
for name, questions in yaml.load(f, Loader=yaml.SafeLoader)[
|
||||
"exercices"
|
||||
].items():
|
||||
exam.add_exercise(name, questions)
|
||||
|
||||
print(exam.themes)
|
||||
# print(yaml.dump(exam.to_dict()))
|
||||
|
||||
exam.write()
|
||||
|
||||
for name, questions in exam.exercices.items():
|
||||
exam.modify_exercise(
|
||||
**prompt_exercise(
|
||||
name=name, completer={"theme": exam.themes}, questions=questions
|
||||
)
|
||||
)
|
||||
exam.write()
|
||||
|
||||
new_exercise = prompt_validate("Ajouter un exercice? ")
|
||||
while new_exercise:
|
||||
exam.add_exercise(
|
||||
**prompt_exercise(len(exam.exercices) + 1, completer={"theme": exam.themes})
|
||||
)
|
||||
exam.write()
|
||||
new_exercise = prompt_validate("Ajouter un exercice? ")
|
||||
|
||||
rows = exam.to_row()
|
||||
|
||||
base_df = pd.DataFrame.from_dict(rows)[NO_ST_COLUMNS.keys()]
|
||||
base_df.rename(columns=NO_ST_COLUMNS, inplace=True)
|
||||
|
||||
students = pd.read_csv(exam.tribe_student_path)["Nom"]
|
||||
for student in students:
|
||||
base_df[student] = ""
|
||||
|
||||
exam.tribe_path.mkdir(exist_ok=True)
|
||||
|
||||
base_df.to_csv(exam.path(".csv"), index=False)
|
||||
print(f"Le fichier note a été enregistré à {exam.path('.csv')}")
|
||||
|
||||
|
||||
@cli.command()
|
||||
@click.option("--debug", default=0, help="Debug mode for dash")
|
||||
def dashboard(debug):
|
||||
dash.run_server(debug=bool(debug))
|
||||
|
||||
|
||||
@cli.command()
|
||||
@click.argument("csv_file")
|
||||
def report(csv_file):
|
||||
csv = Path(csv_file)
|
||||
if not csv.exists():
|
||||
click.echo(f"{csv_file} does not exists")
|
||||
sys.exit(1)
|
||||
if csv.suffix != ".csv":
|
||||
click.echo(f"{csv_file} has to be a csv file")
|
||||
sys.exit(1)
|
||||
|
||||
csv_file = Path(csv_file)
|
||||
tribe_dir = csv_file.parent
|
||||
csv_filename = csv_file.name.split(".")[0]
|
||||
|
||||
assessment = str(csv_filename).split("_")[-1].capitalize()
|
||||
date = str(csv_filename).split("_")[0]
|
||||
try:
|
||||
date = datetime.strptime(date, "%y%m%d")
|
||||
except ValueError:
|
||||
date = None
|
||||
|
||||
tribe = str(tribe_dir).split("/")[-1]
|
||||
|
||||
template = Path(config["templates"]) / "tpl_evaluation.ipynb"
|
||||
|
||||
dest = Path(config["output"]) / tribe / csv_filename
|
||||
dest.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
click.echo(f"Building {assessment} ({date:%d/%m/%y}) report")
|
||||
pm.execute_notebook(
|
||||
str(template),
|
||||
str(dest / f"{assessment}.ipynb"),
|
||||
parameters=dict(
|
||||
tribe=tribe,
|
||||
assessment=assessment,
|
||||
date=f"{date:%d/%m/%y}",
|
||||
csv_file=str(csv_file.absolute()),
|
||||
),
|
||||
)
|
||||
Reference in New Issue
Block a user