coldcall_lti/coldcall_lti/models.py

import datetime

from sqlalchemy import (
    Boolean,
    Date,
    DateTime,
    Float,
    ForeignKey,
    String,
    Text,
    UniqueConstraint,
    func,
    select,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship

from .db import Base


def utcnow():
    return datetime.datetime.now(datetime.UTC)


# Call outcomes. ANSWERED is the only status that counts toward a
# student's selection weight; SKIPPED calls are treated as if they never
# happened; MISSING means the student was called but absent without an
# opt-out. PENDING is used for pre-generated (printable) lists whose
# outcomes are recorded after class.
STATUS_ANSWERED = "answered"
STATUS_MISSING = "missing"
STATUS_SKIPPED = "skipped"
STATUS_PENDING = "pending"
CALL_STATUSES = (STATUS_ANSWERED, STATUS_MISSING, STATUS_SKIPPED, STATUS_PENDING)

# Default assessment vocabulary, carried over from the manual system.
ASSESSMENTS = ("GOOD", "SATISFACTORY", "POOR", "NO MEANINGFUL ANSWER")

SELECTION_WEIGHTED = "weighted"
SELECTION_CYCLE = "cycle"


class Course(Base):
    __tablename__ = "courses"

    id: Mapped[int] = mapped_column(primary_key=True)
    lti_context_id: Mapped[str] = mapped_column(String(255), unique=True)
    lti_deployment_id: Mapped[str | None] = mapped_column(String(255))
    title: Mapped[str | None] = mapped_column(String(255))

    # Per-course settings.
    weight_factor: Mapped[float] = mapped_column(Float, default=2.0)
    selection_mode: Mapped[str] = mapped_column(
        String(16), default=SELECTION_WEIGHTED
    )
    show_assessments: Mapped[bool] = mapped_column(Boolean, default=True)

    created_at: Mapped[datetime.datetime] = mapped_column(
        DateTime, default=utcnow
    )

    enrollments: Mapped[list["Enrollment"]] = relationship(back_populates="course")


class Student(Base):
    __tablename__ = "students"

    id: Mapped[int] = mapped_column(primary_key=True)
    # The LTI "sub" claim: stable, opaque, unique per Canvas user.
    canvas_user_id: Mapped[str] = mapped_column(String(255), unique=True)
    name: Mapped[str | None] = mapped_column(String(255))
    sortable_name: Mapped[str | None] = mapped_column(String(255))
    email: Mapped[str | None] = mapped_column(String(255))
    avatar_url: Mapped[str | None] = mapped_column(String(1024))

    enrollments: Mapped[list["Enrollment"]] = relationship(back_populates="student")


class Enrollment(Base):
    __tablename__ = "enrollments"
    __table_args__ = (UniqueConstraint("course_id", "student_id"),)

    id: Mapped[int] = mapped_column(primary_key=True)
    course_id: Mapped[int] = mapped_column(ForeignKey("courses.id"))
    student_id: Mapped[int] = mapped_column(ForeignKey("students.id"))
    role: Mapped[str] = mapped_column(String(32), default="student")
    # Set False when a student disappears from the NRPS roster (dropped).
    active: Mapped[bool] = mapped_column(Boolean, default=True)

    course: Mapped[Course] = relationship(back_populates="enrollments")
    student: Mapped[Student] = relationship(back_populates="enrollments")


class ClassDay(Base):
    """A day the class actually meets. When a course has these, the
    student opt-out form only offers real class days; without them it
    falls back to a free date picker."""

    __tablename__ = "class_days"
    __table_args__ = (UniqueConstraint("course_id", "date"),)

    id: Mapped[int] = mapped_column(primary_key=True)
    course_id: Mapped[int] = mapped_column(ForeignKey("courses.id"))
    date: Mapped[datetime.date] = mapped_column(Date)


class OptOut(Base):
    __tablename__ = "optouts"
    __table_args__ = (UniqueConstraint("course_id", "student_id", "date"),)

    id: Mapped[int] = mapped_column(primary_key=True)
    course_id: Mapped[int] = mapped_column(ForeignKey("courses.id"))
    student_id: Mapped[int] = mapped_column(ForeignKey("students.id"))
    date: Mapped[datetime.date] = mapped_column(Date)
    created_at: Mapped[datetime.datetime] = mapped_column(
        DateTime, default=utcnow
    )


class Call(Base):
    __tablename__ = "calls"

    id: Mapped[int] = mapped_column(primary_key=True)
    course_id: Mapped[int] = mapped_column(ForeignKey("courses.id"))
    student_id: Mapped[int] = mapped_column(ForeignKey("students.id"))
    session_date: Mapped[datetime.date] = mapped_column(Date)
    status: Mapped[str] = mapped_column(String(16), default=STATUS_PENDING)
    assessment: Mapped[str | None] = mapped_column(String(32))
    note: Mapped[str | None] = mapped_column(Text)
    created_at: Mapped[datetime.datetime] = mapped_column(
        DateTime, default=utcnow
    )

    student: Mapped[Student] = relationship()


def answered_call_counts(session, course_id):
    """Number of answered calls per student, the input to selection
    weighting. Skipped, pending, and missing calls do not count, matching
    the old system where only answered calls reduced a student's odds."""
    rows = session.execute(
        select(Call.student_id, func.count())
        .where(Call.course_id == course_id, Call.status == STATUS_ANSWERED)
        .group_by(Call.student_id)
    ).all()
    return dict(rows)


def class_days(session, course_id, start=None, end=None):
    q = select(ClassDay).where(ClassDay.course_id == course_id)
    if start is not None:
        q = q.where(ClassDay.date >= start)
    if end is not None:
        q = q.where(ClassDay.date <= end)
    return session.execute(q.order_by(ClassDay.date)).scalars().all()


def is_class_day(session, course_id, on_date):
    """True if on_date is a listed class day, or if the course has no
    class days defined at all (no schedule means no restriction)."""
    if not session.execute(
        select(ClassDay.id).where(ClassDay.course_id == course_id).limit(1)
    ).first():
        return True
    return bool(
        session.execute(
            select(ClassDay.id).where(
                ClassDay.course_id == course_id, ClassDay.date == on_date
            )
        ).first()
    )


def students_present(session, course_id, on_date):
    """Active student-role enrollees minus those opted out for the date."""
    opted_out = select(OptOut.student_id).where(
        OptOut.course_id == course_id, OptOut.date == on_date
    )
    rows = session.execute(
        select(Student)
        .join(Enrollment, Enrollment.student_id == Student.id)
        .where(
            Enrollment.course_id == course_id,
            Enrollment.role == "student",
            Enrollment.active.is_(True),
            Student.id.not_in(opted_out),
        )
    ).scalars()
    return list(rows)