AdvLabDB/advlabdb/models.py

"""
See the file DB.drawio for the design of the database. It can be opened in the internet browser with the website:
https://app.diagrams.net

For more information about the implementation, see the part to Models in the documentation of Flask-SQLAlchemy:
https://flask-sqlalchemy.palletsprojects.com/en/2.x/models/
"""

from decimal import ROUND_HALF_UP, Decimal

from flask import flash
from flask_login import current_user
from flask_security.models import fsqla_v2 as fsqla
from flask_security.models.fsqla_v2 import FsRoleMixin, FsUserMixin
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import func, select

from .exceptions import DatabaseException

MIN_MARK = 0
MAX_MARK = 15
MIN_YEAR = 22
MAX_YEAR = 99
MIN_STUDENT_NUMBER = 0
MIN_EXPERIMENT_NUMBER = 1
MIN_GROUP_NUMBER = 1
MIN_DURATION_IN_DAYS = 1
MIN_PART_NUMBER = 1

db = SQLAlchemy()

# For Flask-Security-Too
fsqla.FsModels.set_db_info(db)


def get_count(table):
    return db.session.scalar(select(func.count()).select_from(table))


def get_first(table):
    return db.session.execute(table.limit(1)).scalars().first()


def roundHalfUpToInt(number):
    return int(Decimal(number).quantize(Decimal(0), rounding=ROUND_HALF_UP))


class Student(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    student_number = db.Column(
        db.Integer, db.CheckConstraint(f"student_number >= {MIN_STUDENT_NUMBER}"), nullable=False, unique=True
    )
    first_name = db.Column(db.String(100), nullable=False)
    last_name = db.Column(db.String(100), nullable=False)
    uni_email = db.Column(db.String(200), nullable=False, unique=True)
    contact_email = db.Column(db.String(200), nullable=True, unique=True)
    bachelor_thesis = db.Column(db.String, nullable=True)
    bachelor_thesis_work_group = db.Column(db.String, nullable=True)
    note = db.Column(db.Text, nullable=True)

    part_students = db.relationship("PartStudent", back_populates="student", lazy=True)

    def str(self):
        return f"{self.first_name} {self.last_name}"

    def __str__(self):
        return self.str()


class PartStudent(db.Model):
    # A student doing a specific part
    id = db.Column(db.Integer, primary_key=True)
    final_part_mark = db.Column(
        db.Integer,
        db.CheckConstraint(f"final_part_mark BETWEEN {MIN_MARK} AND {MAX_MARK}"),
        nullable=True,
    )

    student_id = db.Column(db.Integer, db.ForeignKey("student.id"), nullable=False)
    student = db.relationship("Student", back_populates="part_students")
    part_id = db.Column(db.Integer, db.ForeignKey("part.id"), nullable=False)
    part = db.relationship("Part", back_populates="part_students")
    group_id = db.Column(db.Integer, db.ForeignKey("group.id"), nullable=True)
    group = db.relationship("Group", back_populates="part_students")

    experiment_marks = db.relationship("ExperimentMark", back_populates="part_student", lazy=True)

    __table_args__ = (
        db.UniqueConstraint(student_id, part_id),
        db.UniqueConstraint(student_id, group_id),
    )

    def str(self):
        return f"{self.student.str()} {self.part}"

    def __str__(self):
        return f"<{self.str()}>"

    @staticmethod
    def check(part, group):
        if group is not None and group.program != part.program:
            raise DatabaseException(
                f"Group's program {group.program} and student part's program {part.program} do not match!"
            )

    def __init__(self, student, part, group=None):
        PartStudent.check(part, group)

        super().__init__(student=student, part=part, group=group)

    def checkThenSetFinalPartMark(self):
        """
        Return True if final_part_mark changed, False otherwise.
        """
        finalWeightingSum = 0
        finalMarkSum = 0
        groupExperiments = []

        for experimentMark in self.experiment_marks:
            if None in (experimentMark.oral_mark, experimentMark.protocol_mark):
                # Not all marks are set!
                try:
                    self.final_part_mark = None

                    db.session.commit()
                    return True
                except Exception as ex:
                    flash(str(ex), "error")

                    db.session.rollback()
                    return False

            groupExperiment = experimentMark.group_experiment
            groupExperiments.append(groupExperiment)

            semesterExperiment = groupExperiment.semester_experiment

            finalWeighting = semesterExperiment.final_weighting
            finalWeightingSum += finalWeighting

            # Not using final_experiment_mark to avoid rounding two times
            finalMarkSum += finalWeighting * (
                semesterExperiment.protocol_weighting * experimentMark.protocol_mark
                + semesterExperiment.oral_weighting * experimentMark.oral_mark
            )

        if set(groupExperiments) != set(self.group.group_experiments):
            flash(f"{self} does not have an experiment mark for every group experiment in his group!", "warning")
            return False

        oldFinalPartMark = self.final_part_mark

        try:
            self.final_part_mark = roundHalfUpToInt(finalMarkSum / finalWeightingSum)

            db.session.commit()
        except Exception as ex:
            flash(str(ex), "error")

            db.session.rollback()
            return False

        # Inform admin about what changed
        if current_user.has_role("admin"):
            if oldFinalPartMark != self.final_part_mark:
                category = "danger" if oldFinalPartMark and oldFinalPartMark > self.final_part_mark else "info"

                flash(
                    f"Final part mark changed for {self} from {oldFinalPartMark} to {self.final_part_mark}.",
                    category,
                )
            else:
                flash(f"Final part mark did not change for {self} from {oldFinalPartMark}.", "warning")

        return True


class Group(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    number = db.Column(db.Integer, db.CheckConstraint(f"number >= {MIN_GROUP_NUMBER}"), nullable=False)

    semester_id = db.Column(db.Integer, db.ForeignKey("semester.id"), nullable=False)
    semester = db.relationship("Semester", back_populates="groups")
    program_id = db.Column(db.Integer, db.ForeignKey("program.id"), nullable=False)
    program = db.relationship("Program", back_populates="groups")

    part_students = db.relationship("PartStudent", back_populates="group", lazy=True)
    group_experiments = db.relationship("GroupExperiment", back_populates="group", lazy=True)

    __table_args__ = (db.UniqueConstraint(number, semester_id, program_id),)

    def str_without_semester(self):
        return f"{self.number} {self.program.str()}"

    def str(self):
        return f"{self.str_without_semester()} {self.semester.str()}"

    def __str__(self):
        return f"<{self.str()}>"

    @staticmethod
    def check(part_students, program=None):
        commonProgram = part_students[0].part.program

        if program and program != commonProgram:
            raise DatabaseException("Group's program and students' program mismatch!")

        for partStudent in part_students[1:]:
            if partStudent.part.program != commonProgram:
                raise DatabaseException(f"Part Students {part_students} are not in the same program!")

    @staticmethod
    def customInit(part_students):
        Group.check(part_students)

        semester = part_students[0].part.semester
        program = part_students[0].part.program

        highestGroupNumber = get_first(
            select(Group.number)
            .where(Group.semester == semester, Group.program == program)
            .order_by(Group.number.desc())
        )

        if highestGroupNumber is not None:
            number = highestGroupNumber + 1
        else:
            number = 1

        return Group(
            program=program,
            number=number,
            part_students=part_students,
            semester=semester,
        )


class GroupExperiment(db.Model):
    # An experiment specified to a group
    id = db.Column(db.Integer, primary_key=True)
    note = db.Column(db.Text, nullable=True)
    experiment_marks_missing = db.Column(db.Boolean, default=True, nullable=False)

    semester_experiment_id = db.Column(db.Integer, db.ForeignKey("semester_experiment.id"), nullable=False)
    semester_experiment = db.relationship("SemesterExperiment", back_populates="group_experiments")
    group_id = db.Column(db.Integer, db.ForeignKey("group.id"), nullable=False)
    group = db.relationship("Group", back_populates="group_experiments")

    appointments = db.relationship("Appointment", back_populates="group_experiment", lazy=True)
    experiment_marks = db.relationship("ExperimentMark", back_populates="group_experiment", lazy=True)

    __table_args__ = (db.UniqueConstraint(semester_experiment_id, group_id),)

    def str(self):
        return f"<SemExp: {self.semester_experiment.str()} | Gr: {self.group.str()}>"

    def __str__(self):
        return self.str()

    @staticmethod
    def check(semester_experiment, group):
        for partStudent in group.part_students:
            student = partStudent.student
            for partStudent in student.part_students:
                for experimentMark in partStudent.experiment_marks:
                    if experimentMark.group_experiment.semester_experiment.experiment == semester_experiment.experiment:
                        raise DatabaseException(
                            f"{student} has already done {semester_experiment.experiment} in {partStudent.part} and had {experimentMark}!"
                        )

    def __init__(self, semester_experiment, group):
        GroupExperiment.check(semester_experiment, group)

        super().__init__(semester_experiment=semester_experiment, group=group)

        for partStudent in group.part_students:
            db.session.add(ExperimentMark(part_student=partStudent, group_experiment=self))

    def update_experiment_marks_missing(self):
        for experiment_mark in self.experiment_marks:
            if experiment_mark.final_experiment_mark is None:
                self.experiment_marks_missing = True
                return

        self.experiment_marks_missing = False


class Experiment(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    number = db.Column(db.Integer, db.CheckConstraint(f"number >= {MIN_EXPERIMENT_NUMBER}"), nullable=False)
    title = db.Column(db.String(200), nullable=False)
    description = db.Column(db.Text, nullable=True)
    wiki_link = db.Column(db.String(300), nullable=True)
    building = db.Column(db.String(100), nullable=False)
    room = db.Column(db.String(100), nullable=False)
    responsibility = db.Column(db.String(200), nullable=True)
    duration_in_days = db.Column(
        db.Integer, db.CheckConstraint(f"duration_in_days >= {MIN_DURATION_IN_DAYS}"), nullable=False
    )
    active = db.Column(db.Boolean, default=True, nullable=False)

    program_id = db.Column(db.Integer, db.ForeignKey("program.id"), nullable=False)
    program = db.relationship("Program", back_populates="experiments")

    semester_experiments = db.relationship("SemesterExperiment", back_populates="experiment", lazy=True)

    __table_args__ = (db.UniqueConstraint(number, program_id),)

    def str(self):
        return f"{self.number} {self.program.str()}"

    def __str__(self):
        return f"<{self.str()}>"


# Helper table for the many to many relationship between Assistant and SemesterExperiment
experiment_assistant = db.Table(
    "experiment_assistant",
    db.Column("semester_experiment_id", db.Integer, db.ForeignKey("semester_experiment.id"), primary_key=True),
    db.Column("assistant_id", db.Integer, db.ForeignKey("assistant.id"), primary_key=True),
)


class SemesterExperiment(db.Model):
    # An experiment in a specific semester
    id = db.Column(db.Integer, primary_key=True)
    oral_weighting = db.Column(
        db.Float,
        db.CheckConstraint("oral_weighting BETWEEN 0 AND 1"),
        default=0.5,
        nullable=False,
    )
    protocol_weighting = db.Column(
        db.Float,
        db.CheckConstraint("protocol_weighting BETWEEN 0 AND 1"),
        default=0.5,
        nullable=False,
    )
    final_weighting = db.Column(
        db.Float,
        db.CheckConstraint("final_weighting BETWEEN 0 AND 1"),
        default=1.0,
        nullable=False,
    )

    experiment_id = db.Column(db.Integer, db.ForeignKey("experiment.id"), nullable=False)
    experiment = db.relationship("Experiment", back_populates="semester_experiments")
    semester_id = db.Column(db.Integer, db.ForeignKey("semester.id"), nullable=False)
    semester = db.relationship("Semester", back_populates="semester_experiments")

    assistants = db.relationship(
        "Assistant", secondary=experiment_assistant, back_populates="semester_experiments", lazy=True
    )
    group_experiments = db.relationship("GroupExperiment", back_populates="semester_experiment", lazy=True)

    __table_args__ = (db.UniqueConstraint(experiment_id, semester_id),)

    def str_without_semester(self):
        return f"{self.experiment.str()}"

    def str(self):
        return f"{self.str_without_semester()} {self.semester.str()}"

    def __str__(self):
        return f"<{self.str()}>"

    def checkAndRoundWeightings(self):
        roundedOralWeighting = round(self.oral_weighting, 2)
        roundedProtocolWeighting = round(self.protocol_weighting, 2)

        weightingSum = round(roundedOralWeighting + roundedProtocolWeighting, 2)

        if weightingSum != 1:
            raise DatabaseException(
                f"Oral and protocol weightings (rounded to 2 decimal digits) sum to {weightingSum} and not 1.00!"
            )

        self.oral_weighting = roundedOralWeighting
        self.protocol_weighting = roundedProtocolWeighting

    def updateFinalExperimentAndPartMarks(self):
        for groupExperiment in self.group_experiments:
            for experimentMark in groupExperiment.experiment_marks:
                experimentMark.update_final_experiment_mark()

        for groupExperiment in self.group_experiments:
            for partStudent in groupExperiment.group.part_students:
                partStudent.checkThenSetFinalPartMark()


class Assistant(db.Model):
    id = db.Column(db.Integer, primary_key=True)

    user_id = db.Column(db.Integer, db.ForeignKey("user.id"), nullable=False, unique=True)
    user = db.relationship("User", back_populates="assistant")

    semester_experiments = db.relationship(
        "SemesterExperiment", secondary=experiment_assistant, back_populates="assistants", lazy=True
    )
    appointments = db.relationship("Appointment", back_populates="assistant", lazy=True)
    experiment_marks = db.relationship("ExperimentMark", back_populates="assistant", lazy=True)

    def str(self):
        return f"{self.user}"

    def __str__(self):
        return self.str()


class Admin(db.Model):
    id = db.Column(db.Integer, primary_key=True)

    user_id = db.Column(db.Integer, db.ForeignKey("user.id"), nullable=False, unique=True)
    user = db.relationship("User", back_populates="admin")

    experiment_marks = db.relationship("ExperimentMark", back_populates="admin", lazy=True)

    def str(self):
        return f"{self.user}"

    def __str__(self):
        return self.str()


class Appointment(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    date = db.Column(
        db.Date, db.CheckConstraint(f"date BETWEEN '20{MIN_YEAR}-01-01' AND '20{MAX_YEAR}-01-01'"), nullable=False
    )  # To be specified with the python package "datetime"
    special = db.Column(db.Boolean, default=False, nullable=False)  # In the break or not

    group_experiment_id = db.Column(db.Integer, db.ForeignKey("group_experiment.id"), nullable=False)
    group_experiment = db.relationship("GroupExperiment", back_populates="appointments")
    assistant_id = db.Column(db.Integer, db.ForeignKey("assistant.id"), nullable=False)
    assistant = db.relationship("Assistant", back_populates="appointments")

    def str(self):
        return f"{self.date} {self.group_experiment.str()}"

    def __str__(self):
        return f"<{self.str()}>"

    @staticmethod
    def checkAndGetAssistant(groupExperiment, assistant=None):
        semesterExperiment = groupExperiment.semester_experiment
        semesterExperimentAssistants = semesterExperiment.assistants

        if semesterExperimentAssistants is None:
            raise DatabaseException(f"{semesterExperiment} does not have assistants yet!")

        if assistant is not None:
            if assistant not in semesterExperimentAssistants:
                raise DatabaseException(f"{assistant} is not responsible for {semesterExperiment}!")
        else:
            if len(semesterExperimentAssistants) == 1:
                assistant = semesterExperimentAssistants[0]
            else:
                raise DatabaseException(
                    f"Experiment {semesterExperiment} has more than one assistant. You have to assign one of these assistants: {semesterExperimentAssistants}"
                )

        return assistant

    def __init__(self, date, special, group_experiment, assistant=None):
        assistant = Appointment.checkAndGetAssistant(group_experiment, assistant)

        super().__init__(
            date=date,
            special=special,
            group_experiment=group_experiment,
            assistant=assistant,
        )

    def custom_update(self, date, special, group_experiment, assistant=None):
        self.assistant = Appointment.checkAndGetAssistant(group_experiment, assistant)

        self.date = date
        self.special = special
        self.group_experiment = group_experiment


class Part(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    number = db.Column(db.Integer, db.CheckConstraint(f"number >= {MIN_PART_NUMBER}"), nullable=False)

    program_id = db.Column(db.Integer, db.ForeignKey("program.id"), nullable=False)
    program = db.relationship("Program", back_populates="parts")
    semester_id = db.Column(db.Integer, db.ForeignKey("semester.id"), nullable=False)
    semester = db.relationship("Semester", back_populates="parts")

    part_students = db.relationship("PartStudent", back_populates="part", lazy=True)

    __table_args__ = (db.UniqueConstraint(program_id, number, semester_id),)

    def str_without_semester(self):
        return f"{self.program.str()}{self.number}"

    def str(self):
        return f"{self.str_without_semester()} {self.semester.str()}"

    def __str__(self):
        return f"<{self.str()}>"


class Semester(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    label = db.Column(db.String(10), db.CheckConstraint("label IN ('WS', 'SS')"), nullable=False)
    year = db.Column(db.Integer, db.CheckConstraint(f"year BETWEEN {MIN_YEAR} AND {MAX_YEAR}"), nullable=False)

    parts = db.relationship("Part", back_populates="semester", lazy=True)
    semester_experiments = db.relationship("SemesterExperiment", back_populates="semester", lazy=True)
    active_users = db.relationship("User", back_populates="active_semester", lazy=True)
    groups = db.relationship("Group", back_populates="semester", lazy=True)

    __table_args__ = (db.UniqueConstraint(label, year),)

    def str(self):
        return f"{self.label}{self.year}"

    def __str__(self):
        return self.str()

    def __init__(self, label, year):
        last_semester = Semester.lastSemester()

        if last_semester is not None:
            if year < last_semester.year or (year == last_semester.year and label == "SS"):
                raise DatabaseException(f"You can only create semesters older than the last semester {last_semester}!")

        super().__init__(label=label, year=year)

    @staticmethod
    def initFromOldSemester(label, year, oldSemester, transferParts, transferAssistants):
        semester = Semester(label=label, year=year)

        if transferParts:
            semester.transferPartsFrom(oldSemester)

        for experiment in db.session.execute(select(Experiment).where(Experiment.active == True)).scalars():
            newSemesterExperiment = SemesterExperiment(experiment=experiment, semester=semester)

            if transferAssistants:
                for oldSemesterExperiment in oldSemester.semester_experiments:
                    if oldSemesterExperiment.experiment == experiment:
                        newSemesterExperiment.assistants = oldSemesterExperiment.assistants

            db.session.add(newSemesterExperiment)

        return semester

    def transferPartsFrom(self, oldSemester):
        for part in oldSemester.parts:
            db.session.add(Part(program=part.program, number=part.number, semester=self))

    @staticmethod
    def sortedSemestersStartingWithNewest(limit=0):
        # Inserting an older semester is not allowed!
        # Therefore, the id is enough.
        stmt = select(Semester).order_by(Semester.id.desc())

        if limit > 0:
            stmt = stmt.limit(limit)

        return db.session.execute(stmt).scalars()

    @staticmethod
    def lastSemester():
        return Semester.sortedSemestersStartingWithNewest(limit=1).first()


class ExperimentMark(db.Model):
    # A mark for a student after a specific experiment

    id = db.Column(db.Integer, primary_key=True)

    oral_mark = db.Column(
        db.Integer,
        db.CheckConstraint(f"oral_mark BETWEEN {MIN_MARK} AND {MAX_MARK}"),
        nullable=True,
    )
    protocol_mark = db.Column(
        db.Integer,
        db.CheckConstraint(f"oral_mark BETWEEN {MIN_MARK} AND {MAX_MARK}"),
        nullable=True,
    )

    final_experiment_mark = db.Column(
        db.Integer,
        db.CheckConstraint(f"oral_mark BETWEEN {MIN_MARK} AND {MAX_MARK}"),
        nullable=True,
    )

    part_student_id = db.Column(db.Integer, db.ForeignKey("part_student.id"), nullable=False)
    part_student = db.relationship("PartStudent", back_populates="experiment_marks")
    group_experiment_id = db.Column(db.Integer, db.ForeignKey("group_experiment.id"), nullable=False)
    group_experiment = db.relationship("GroupExperiment", back_populates="experiment_marks")
    assistant_id = db.Column(
        db.Integer, db.ForeignKey("assistant.id"), nullable=True
    )  # The last assistant who edited the mark
    assistant = db.relationship("Assistant", back_populates="experiment_marks")
    admin_id = db.Column(db.Integer, db.ForeignKey("admin.id"), nullable=True)  # The last admin who edited the mark
    admin = db.relationship("Admin", back_populates="experiment_marks")

    __table_args__ = (db.UniqueConstraint(part_student_id, group_experiment_id),)

    def str(self):
        return (
            f"<Final Exp Mark: {self.final_experiment_mark} | Exp: {self.group_experiment.semester_experiment.str()}>"
        )

    def __str__(self):
        return self.str()

    @staticmethod
    def check(part_student, group_experiment):
        if not part_student.group:
            raise DatabaseException("The part student does not have a group yet!")
        else:
            if group_experiment not in part_student.group.group_experiments:
                raise DatabaseException("The group of the part student does not have the given group experiment!")

    def __init__(self, part_student, group_experiment):
        ExperimentMark.check(part_student, group_experiment)

        super().__init__(part_student=part_student, group_experiment=group_experiment)

    def update_final_experiment_mark(self):
        """
        Return True if final_experiment_mark changed, False otherwise.
        """
        old_final_experiment_mark = self.final_experiment_mark
        oral_mark = self.oral_mark
        protocol_mark = self.protocol_mark

        if None in (oral_mark, protocol_mark):
            self.final_experiment_mark = None
        else:
            semesterExperiment = self.group_experiment.semester_experiment
            self.final_experiment_mark = roundHalfUpToInt(
                semesterExperiment.oral_weighting * oral_mark + semesterExperiment.protocol_weighting * protocol_mark
            )

        final_experiment_mark_changed = self.final_experiment_mark != old_final_experiment_mark
        return final_experiment_mark_changed

    def set_oral_protocol_mark(self, oral_mark: int, protocol_mark: int, call_update_experiment_marks_missing=True):
        """
        Return True if final_experiment_mark changed, False otherwise.
        """
        self.oral_mark = oral_mark
        self.protocol_mark = protocol_mark

        final_experiment_mark_changed = self.update_final_experiment_mark()
        if final_experiment_mark_changed:
            self.part_student.checkThenSetFinalPartMark()
            if call_update_experiment_marks_missing:
                self.group_experiment.update_experiment_marks_missing()

        return final_experiment_mark_changed


class User(db.Model, FsUserMixin):
    first_name = db.Column(db.String(100), nullable=False)
    last_name = db.Column(db.String(100), nullable=False)
    phone_number = db.Column(db.String(50), nullable=True)
    mobile_phone_number = db.Column(db.String(50), nullable=True)
    building = db.Column(db.String(100), nullable=True)
    room = db.Column(db.String(100), nullable=True)

    active_semester_id = db.Column(db.Integer, db.ForeignKey("semester.id"), nullable=False)
    active_semester = db.relationship("Semester", back_populates="active_users")

    admin = db.relationship("Admin", back_populates="user", lazy=False, uselist=False)
    assistant = db.relationship("Assistant", back_populates="user", lazy=True, uselist=False)

    def str(self):
        return f"{self.first_name} {self.last_name}"

    def __str__(self):
        return self.str()

    def setActiveSemester(self, semester):
        if semester is None:
            flash("Can not change the active semester to None!", "error")
            return

        if self.active_semester != semester:
            try:
                self.active_semester = semester
                db.session.commit()
            except Exception as ex:
                flash(str(ex), "error")

                db.session.rollback()
            else:
                flash(f"Active semester changed to {semester}!", "warning")


class Role(db.Model, FsRoleMixin):
    def str(self):
        return f"{self.name}"

    def __str__(self):
        return self.str()


class Program(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    label = db.Column(db.String(25), nullable=False, unique=True)

    parts = db.relationship("Part", back_populates="program", lazy=True)
    experiments = db.relationship("Experiment", back_populates="program", lazy=True)
    groups = db.relationship("Group", back_populates="program", lazy=True)

    def str(self):
        return f"{self.label}"

    def __str__(self):
        return self.str()