coldcall_lti/coldcall_lti/models.py

import datetime

from sqlalchemy import (
    Boolean,
    Date,
    DateTime,
    Float,
    ForeignKey,
    String,
    Text,
    Time,
    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 and points, carried over from the
# manual system: compute_final_case_grades.R scored GOOD=100,
# SATISFACTORY and POOR one and two UW-4.0-scale steps down
# (100 - n*50/(4-0.7)), and NO MEANINGFUL ANSWER zero.
DEFAULT_ASSESSMENT_LEVELS = (
    ("GOOD", 100.0),
    ("SATISFACTORY", 84.85),
    ("POOR", 69.7),
    ("NO MEANINGFUL ANSWER", 0.0),
)

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))

    # Enough platform detail to call NRPS outside a launch, for the
    # on-demand roster refresh. Captured/updated on each launch.
    lti_issuer: Mapped[str | None] = mapped_column(String(255))
    lti_client_id: Mapped[str | None] = mapped_column(String(255))
    nrps_url: Mapped[str | None] = mapped_column(String(1024))

    # Course term dates from Canvas ($Canvas.course.startAt/endAt
    # custom parameters), used to bound date pickers. Nullable: not
    # every course has them set in Canvas.
    start_date: Mapped[datetime.date | None] = mapped_column(Date)
    end_date: Mapped[datetime.date | None] = mapped_column(Date)

    # 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)

    # Grading parameters (participation_grades.R defaults, expressed in
    # points out of 100; 65.15 pts and 4.55 pts are the UW-4.0 values
    # 1.7 and 0.3 under the linear mapping).
    allowance_sd_units: Mapped[float] = mapped_column(Float, default=0.25)
    passing_points: Mapped[float] = mapped_column(Float, default=65.15)
    form_penalty_points: Mapped[float] = mapped_column(Float, default=4.55)
    n_sims: Mapped[int] = mapped_column(default=4000)
    sim_seed: Mapped[int] = mapped_column(default=20260606)

    # Display scale for grades: "uw4" (linear UW 4.0 map), "table"
    # (threshold rows in scale_config JSON: [[min_points, label], ...]),
    # or "none" (points only).
    scale_type: Mapped[str] = mapped_column(String(16), default="uw4")
    scale_config: Mapped[str | None] = mapped_column(Text)
    # The course's grading scheme as Canvas reports it at launch
    # (JSON [[min_points, label], ...]); import copies it into
    # scale_config.
    canvas_grading_scheme: Mapped[str | None] = mapped_column(Text)
    publish_grade_reports: Mapped[bool] = mapped_column(Boolean, default=False)

    # Gradebook passback: off by default; the push pages only exist
    # when enabled. Endpoint details are captured at launch regardless,
    # so enabling later needs no re-launch.
    ags_enabled: Mapped[bool] = mapped_column(Boolean, default=False)
    ags_lineitems_url: Mapped[str | None] = mapped_column(String(1024))
    ags_scopes: Mapped[str | None] = mapped_column(Text)

    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 LMS 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))
    pronouns: Mapped[str | None] = mapped_column(String(64))
    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 AssessmentLevel(Base):
    """One rung of a course's assessment scale: a label and its points
    out of 100. Calls reference levels by id, so renaming a level
    renames it everywhere, past calls included."""

    __tablename__ = "assessment_levels"
    __table_args__ = (UniqueConstraint("course_id", "label"),)

    id: Mapped[int] = mapped_column(primary_key=True)
    course_id: Mapped[int] = mapped_column(ForeignKey("courses.id"))
    label: Mapped[str] = mapped_column(String(64))
    points: Mapped[float] = mapped_column(Float)
    position: Mapped[int] = mapped_column(default=0)


def ensure_default_levels(session, course):
    existing = session.execute(
        select(AssessmentLevel.id)
        .where(AssessmentLevel.course_id == course.id)
        .limit(1)
    ).first()
    if existing:
        return
    for i, (label, points) in enumerate(DEFAULT_ASSESSMENT_LEVELS):
        session.add(
            AssessmentLevel(
                course_id=course.id, label=label, points=points, position=i
            )
        )
    session.flush()


def assessment_levels(session, course_id):
    return (
        session.execute(
            select(AssessmentLevel)
            .where(AssessmentLevel.course_id == course_id)
            .order_by(AssessmentLevel.position, AssessmentLevel.id)
        )
        .scalars()
        .all()
    )


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)
    # When class begins; bounds opt-out withdrawal. Optional: with no
    # time recorded, same-day withdrawal is simply not allowed.
    start_time: Mapped[datetime.time | None] = mapped_column(Time)


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
    )
    # Withdrawals are recorded, never erased: an opt-out with
    # withdrawn_at set no longer counts anywhere, but the audit trail
    # of the change survives.
    withdrawn_at: Mapped[datetime.datetime | None] = mapped_column(DateTime)


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_id: Mapped[int | None] = mapped_column(
        ForeignKey("assessment_levels.id")
    )
    note: Mapped[str | None] = mapped_column(Text)
    created_at: Mapped[datetime.datetime] = mapped_column(
        DateTime, default=utcnow
    )
    # When the outcome (status/assessment) was last recorded or
    # changed. Distinguishes assessed-in-class from assessed-later;
    # null for pending calls and for imported legacy data.
    assessment_entered_at: Mapped[datetime.datetime | None] = mapped_column(
        DateTime
    )

    student: Mapped[Student] = relationship()
    level: Mapped["AssessmentLevel | None"] = relationship()

    @property
    def assessment(self):
        return self.level.label if self.level else None


class GradeRun(Base):
    """One computed grading result: the parameter snapshot and the full
    per-student output, stored as JSON. Grade pages always show a run,
    never a live computation, so what you reviewed is what you push."""

    __tablename__ = "grade_runs"

    id: Mapped[int] = mapped_column(primary_key=True)
    course_id: Mapped[int] = mapped_column(ForeignKey("courses.id"))
    created_at: Mapped[datetime.datetime] = mapped_column(
        DateTime, default=utcnow
    )
    params: Mapped[str] = mapped_column(Text)
    results: Mapped[str] = mapped_column(Text)


def latest_grade_run(session, course_id):
    return session.execute(
        select(GradeRun)
        .where(GradeRun.course_id == course_id)
        .order_by(GradeRun.created_at.desc(), GradeRun.id.desc())
    ).scalars().first()


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 class_has_begun(session, course_id, day, now=None):
    """Whether the class on `day` has started, for bounding opt-out
    withdrawal. Without a recorded start time, the whole class day
    counts as begun (the conservative reading)."""
    now = now or datetime.datetime.now()
    if day < now.date():
        return True
    if day > now.date():
        return False
    row = session.execute(
        select(ClassDay).where(
            ClassDay.course_id == course_id, ClassDay.date == day
        )
    ).scalar_one_or_none()
    if row is None or row.start_time is None:
        return True
    return now.time() >= row.start_time


def nonskipped_call_counts(session, course_id):
    """Calls per student excluding skipped ones — a student's position
    in the cycle-mode pass. Pending list slots count: a printed line
    is a claim on being called."""
    rows = session.execute(
        select(Call.student_id, func.count())
        .where(Call.course_id == course_id, Call.status != STATUS_SKIPPED)
        .group_by(Call.student_id)
    ).all()
    return dict(rows)


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,
        OptOut.withdrawn_at.is_(None),
    )
    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)