recopytex/recopytex/dashboard/exam.py

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

import dash
import dash_html_components as html
import dash_core_components as dcc
import dash_table
from dash.exceptions import PreventUpdate
import plotly.graph_objects as go
from pathlib import Path
from datetime import datetime
import pandas as pd
import numpy as np


from .. import flat_df_students, pp_q_scores
from ..config import NO_ST_COLUMNS
from ..scripts.getconfig import config, CONFIGPATH

COLORS = {
    ".": "black",
    0: "#E7472B",
    1: "#FF712B",
    2: "#F2EC4C",
    3: "#68D42F",
}

external_stylesheets = ["https://codepen.io/chriddyp/pen/bWLwgP.css"]
app = dash.Dash(__name__, external_stylesheets=external_stylesheets)
# app = dash.Dash(__name__)

app.layout = html.Div(
    children=[
        html.H1("Analyse des notes"),
        html.Div(
            [
                "Classe: ",
                dcc.Dropdown(
                    id="tribe",
                    options=[
                        {"label": t["name"], "value": t["name"]}
                        for t in config["tribes"]
                    ],
                    value=config["tribes"][0]["name"],
                ),
                "Evaluation: ",
                dcc.Dropdown(id="csv"),
            ],
            style={"columnCount": 2},
        ),
        html.Div(
            [
                dash_table.DataTable(
                    id="final_score_table",
                    columns=[
                        {"id": "Élève", "name": "Élève"},
                        {"id": "Note", "name": "Note"},
                        {"id": "Barème", "name": "Bareme"},
                    ],
                    data=[],
                    style_data_conditional=[
                        {
                            "if": {"row_index": "odd"},
                            "backgroundColor": "rgb(248, 248, 248)",
                        }
                    ],
                    style_header={
                        "backgroundColor": "rgb(230, 230, 230)",
                        "fontWeight": "bold",
                    },
                    style_data={
                        "width": "100px",
                        "maxWidth": "100px",
                        "minWidth": "100px",
                    },
                ),
                html.Div(
                    [
                        dash_table.DataTable(
                            id="final_score_describe",
                        ),
                        dcc.Graph(id="fig_assessment_hist"),
                        dcc.Graph(id="fig_competences"),
                    ]
                ),
            ],
            style={"columnCount": 2},
        ),
        html.Br(),
        html.Div(
            [
                dash_table.DataTable(
                    id="scores_table",
                    columns=[{"id": c, "name": c} for c in NO_ST_COLUMNS.values()],
                    style_cell={
                        "whiteSpace": "normal",
                        "height": "auto",
                    },
                    style_data_conditional=[],
                    editable=True,
                )
            ]
        ),
        html.P(id="lastsave"),
        dcc.Store(id="final_score"),
    ]
)


@app.callback(
    [
        dash.dependencies.Output("csv", "options"),
        dash.dependencies.Output("csv", "value"),
    ],
    [dash.dependencies.Input("tribe", "value")],
)
def update_csvs(value):
    if not value:
        raise PreventUpdate
    p = Path(value)
    csvs = list(p.glob("*.csv"))
    try:
        return [{"label": str(c), "value": str(c)} for c in csvs], str(csvs[0])
    except IndexError:
        return []


@app.callback(
    [
        dash.dependencies.Output("final_score", "data"),
    ],
    [dash.dependencies.Input("scores_table", "data")],
)
def update_final_scores(data):
    if not data:
        raise PreventUpdate
    try:
        scores = pd.DataFrame.from_records(data)
        scores = flat_df_students(scores).dropna(subset=["Score"])
        scores = pp_q_scores(scores)
        assessment_scores = scores.groupby(["Eleve"]).agg(
            {"Note": "sum", "Bareme": "sum"}
        )
        return [assessment_scores.reset_index().to_dict("records")]
    except KeyError:
        raise PreventUpdate


@app.callback(
    [
        dash.dependencies.Output("final_score_table", "columns"),
        dash.dependencies.Output("final_score_table", "data"),
    ],
    [dash.dependencies.Input("final_score", "data")],
)
def update_final_scores_table(data):
    assessment_scores = pd.DataFrame.from_records(data)
    return [
        {"id": c, "name": c} for c in assessment_scores.columns
    ], assessment_scores.to_dict("records")


@app.callback(
    [
        dash.dependencies.Output("final_score_describe", "columns"),
        dash.dependencies.Output("final_score_describe", "data"),
    ],
    [dash.dependencies.Input("final_score", "data")],
)
def update_final_scores_descr(data):
    desc = pd.DataFrame.from_records(data)["Note"].describe()
    return [{"id": c, "name": c} for c in desc.keys()], [desc.to_dict()]


@app.callback(
    [
        dash.dependencies.Output("fig_assessment_hist", "figure"),
    ],
    [dash.dependencies.Input("final_score", "data")],
)
def update_final_scores_hist(data):
    assessment_scores = pd.DataFrame.from_records(data)

    ranges = np.linspace(
        0, assessment_scores.Bareme.max(), int(assessment_scores.Bareme.max() * 2 + 1)
    )
    bins = pd.cut(assessment_scores["Note"], ranges)
    assessment_scores["Bin"] = bins
    assessment_grouped = (
        assessment_scores.reset_index()
        .groupby("Bin")
        .agg({"Bareme": "count", "Eleve": lambda x: "\n".join(x)})
    )
    assessment_grouped.index = assessment_grouped.index.map(lambda i: i.right)
    fig = go.Figure()
    fig.add_bar(
        x=assessment_grouped.index,
        y=assessment_grouped.Bareme,
        text=assessment_grouped.Eleve,
        textposition="auto",
        hovertemplate="",
        marker_color="#4E89DE",
    )
    # fig = go.Figure(
    #     data=go.Histogram(
    #         x=assessment_scores["Note"],
    #         xbins={"start": 0, "end": assessment_scores["Bareme"].max(), "size": 0.25},
    #     ),
    # )
    return [fig]


@app.callback(
    [
        dash.dependencies.Output("fig_competences", "figure"),
    ],
    [dash.dependencies.Input("scores_table", "data")],
)
def update_competence_fig(data):
    scores = pd.DataFrame.from_records(data)
    scores = flat_df_students(scores).dropna(subset=["Score"])
    scores = pp_q_scores(scores)
    pt = pd.pivot_table(
        scores,
        index=["Exercice", "Question", "Commentaire"],
        columns="Score",
        aggfunc="size",
        fill_value=0,
    )
    for i in {i for i in pt.index.get_level_values(0)}:
        pt.loc[(str(i), "", ""), :] = ""
    pt.sort_index(inplace=True)
    index = (
        pt.index.get_level_values(0)
        + ":"
        + pt.index.get_level_values(1)
        + " "
        + pt.index.get_level_values(2)
    )

    fig = go.Figure()
    bars = [
        {"score": -1, "name": "Pas de réponse", "color": COLORS["."]},
        {"score": 0, "name": "Faut", "color": COLORS[0]},
        {"score": 1, "name": "Peu juste", "color": COLORS[1]},
        {"score": 2, "name": "Presque juste", "color": COLORS[2]},
        {"score": 3, "name": "Juste", "color": COLORS[3]},
    ]
    for b in bars:
        try:
            fig.add_bar(
                x=index, y=pt[b["score"]], name=b["name"], marker_color=b["color"]
            )
        except KeyError:
            pass
    fig.update_layout(barmode="relative")
    return [fig]


@app.callback(
    [dash.dependencies.Output("lastsave", "children")],
    [
        dash.dependencies.Input("scores_table", "data"),
        dash.dependencies.State("csv", "value"),
    ],
)
def save_scores(data, csv):
    scores = pd.DataFrame.from_records(data)
    print(f"save at {csv} ({datetime.today()})")
    scores.to_csv(csv, index=False)
    return [datetime.today()]


def highlight_value(df):
    """ Cells style  """
    hight = []
    for v, color in COLORS.items():
        hight += [
            {
                "if": {"filter_query": "{{{}}} = {}".format(col, v), "column_id": col},
                "backgroundColor": color,
                "color": "white",
            }
            for col in df.columns
            if col not in NO_ST_COLUMNS.values()
        ]
    return hight


@app.callback(
    [
        dash.dependencies.Output("scores_table", "columns"),
        dash.dependencies.Output("scores_table", "data"),
        dash.dependencies.Output("scores_table", "style_data_conditional"),
    ],
    [dash.dependencies.Input("csv", "value")],
)
def update_scores_table(value):
    if not value:
        raise PreventUpdate
    stack = pd.read_csv(value, encoding="UTF8")
    # try:
    #     stack = stack.drop(columns=["Nom", "Trimestre", "Date", "Competence", "Domaine", "Est_nivele", "Bareme"])
    # except KeyError:
    #     stack = stack
    return (
        [{"id": c, "name": c} for c in stack.columns],
        stack.to_dict("records"),
        highlight_value(stack),
    )