Merge branch 'initial-merge' into 'main'

added enrollment quantifier code

See merge request !1

.gitignore

+# Project-specific
+#
+
+# Mac file finder metadata
+.DS_Store
+# Windows file metadata
+._*
+# Old file browser metadata
+Thumbs.db
+
+# Emacs backup file
+*~
+
+# Python files
+*.pyc
+*.pyo
+__pycache__/
+.venv/
+
+# JetBrains (IntelliJ IDEA, PyCharm, etc) files
+.idea/
+*.iml
+*.iws
+*.ipr
+
+# Miscellaneous
+tmp/
+*.tmp
+*.bak
+*.swp
+

Correlate231Grades.py

+import csv
+from getpass import getpass
+from typing import Dict, List
+
+from sqlalchemy.orm import Session
+
+from StudentsDatabase import Enrollment, StudentsDatabase
+
+target_course = 'CSCE231'
+other_courses = {
+    'CS1': ['CSCE155A', 'CSCE155E', 'CSCE155H', 'CSCE155N', 'CSCE155T', 'RAIK183H', 'SOFT160', 'SOFT160H'],
+    'CS2': ['CSCE156', 'CSCE156H', 'RAIK184H', 'SOFT161', 'SOFT161H'],
+    'Discrete Math': ['CSCE235', 'CSCE235H', 'RAIK184H'],
+    'Data Structures & Algorithms': ['CSCE310', 'CSCE310H', 'RAIK283H', 'SOFT260', 'SOFT260H'],
+    'Language Concepts': ['CSCE322', 'CSCE322H']
+}
+csv_fields = ['Semester', 'Student', 'CSCE 231 Grade Category',
+              'CS1 course', 'CS1 grade', 'CS1 failures',
+              'CS2 course', 'CS2 grade', 'CS2 failures',
+              'Discrete Math course', 'Discrete Math grade', 'Discrete Math failures',
+              'Data Structures & Algorithms course', 'Data Structures & Algorithms grade', 'Data Structures & Algorithms failures',
+              'Language Concepts course', 'Language Concepts grade', 'Language Concepts failures']
+
+categorized_grades = {
+    'A+': ['A+'],
+    'A': ['A', 'A-'],
+    'B': ['B+', 'B', 'B-'],
+    'C': ['C+', 'C', 'P'],
+    'DFW': ['C-', 'D+', 'D', 'D-', 'F', 'W', 'N']
+}
+
+semester_codes = ['1218', '1221', '1228', '1231', '1238', '1241']
+# semester_codes = ['1241']
+
+
+def categorize_students(session: Session, semester_code: str) -> Dict[str, List[str]]:
+    categorized_students: Dict[str, List[str]] = {}
+    for grade_category in categorized_grades:
+        students = (session
+                    .query(Enrollment.nuid)
+                    .filter(Enrollment.semester_code == semester_code,
+                            Enrollment.course == target_course,
+                            Enrollment.grade.in_(categorized_grades[grade_category]))
+                    .all())
+        categorized_students[grade_category] = list([str(student[0]) for student in students])
+    return categorized_students
+
+
+def get_other_grades(session: Session, categorized_students: Dict[str, List[str]]) -> List[Dict[str, str]]:
+    other_grades: List[Dict[str, str]] = []
+    for course_category in other_courses:
+        for grade_category in categorized_grades:
+            print(f'\tCSCE231: {grade_category}\t\tcourse category: {course_category}')
+            grades = (session
+                      .query(Enrollment.nuid, Enrollment.course, Enrollment.grade)
+                      .filter(Enrollment.course.in_(other_courses[course_category]),
+                              Enrollment.nuid.in_(categorized_students[grade_category]),
+                              Enrollment.grade is not None, Enrollment.grade != '')
+                      .all())
+            for grade in grades:
+                other_grades.append({
+                    'student': grade[0],
+                    'course category': course_category,
+                    'course': grade[1],
+                    'grade': grade[2]
+                })
+    return other_grades
+
+
+def create_csv_rows(semester_code: str, categorized_students: Dict[str, List[str]],
+                    other_grades: List[Dict[str, str]]) -> List[Dict[str, str]]:
+    rows: List[Dict[str, str]] = []
+    for grade_category in categorized_grades:
+        for student in categorized_students[grade_category]:
+            row = {'Semester': semester_code, 'Student': student, 'CSCE 231 Grade Category': grade_category}
+            student_grades = [grade for grade in other_grades if grade['student'] == student]
+            # course, course category, grade, student -- SOFT160, CS1, A, 02200719
+            for course_category in other_courses:
+                passing_grades = [grade for grade in student_grades if grade['course category'] == course_category
+                                  and grade['grade'] not in categorized_grades['DFW']]
+                failing_grades = [grade for grade in student_grades if grade['course category'] == course_category
+                                  and grade['grade'] in categorized_grades['DFW']]
+                if len(passing_grades) > 0:
+                    passing_grade = passing_grades[0]
+                    row[f'{course_category} course'] = passing_grade['course']
+                    row[f'{course_category} grade'] = passing_grade['grade']
+                else:
+                    row[f'{course_category} course'] = 'n/a'
+                    row[f'{course_category} grade'] = 'n/a'
+                row[f'{course_category} failures'] = str(len(failing_grades))
+            rows.append(row)
+    return rows
+
+
+def main():
+    username = input('Enter your username: ')
+    password = getpass('Enter your password: ')
+    filename: str = input('What file do you want to save the data to? ')
+    session = StudentsDatabase.connect(username, password)
+    with open(filename, 'w') as csvfile:
+        writer = csv.DictWriter(csvfile, fieldnames=csv_fields)
+        writer.writeheader()
+        for semester_code in semester_codes:
+            print(f'Getting students for semester {semester_code}')
+            categorized_students = categorize_students(session, semester_code)
+            print('Getting students\' other grades')
+            other_grades = get_other_grades(session, categorized_students)
+            for row in create_csv_rows(semester_code, categorized_students, other_grades):
+                writer.writerow(row)
+
+
+if __name__ == '__main__':
+    main()

EnrollmentQuantifier.py

+import re
+from getpass import getpass
+from typing import List
+
+from sqlalchemy import distinct
+from sqlalchemy.orm import Session
+
+from StudentsDatabase import Enrollment, StudentsDatabase
+
+
+class Course:       # TODO: replace this with a dict
+    def __init__(self, prefix, number, suffix):
+        self.prefix = prefix
+        self.number = number
+        self.suffix = suffix
+        self.processed = False
+
+    def __str__(self):
+        return f'{self.prefix}{self.number}{self.suffix}'
+
+
+def get_courses(session: Session, semester_code: str) -> List[List[str]]:
+    # noinspection PyTypeChecker
+    database_courses = (session
+                        .query(distinct(Enrollment.course))
+                        .filter(Enrollment.semester_code == semester_code)
+                        .all())
+    courses = []
+    combined_courses = []
+    for course in database_courses:
+        match = re.match(r'([a-zA-Z]+)(\d*)([a-zA-Z]?)', str(course[0]))
+        if match:
+            try:
+                courses.append(Course(match.group(1), int(match.group(2)), match.group(3)))
+            except ValueError as error:
+                print(f'Skipping {course[0]} because "{error}"')
+        else:
+            print(f'Skipping {course[0]} because it does not match the regex')
+    courses.sort(key=lambda c: f'{c.number}{c.suffix}')
+    for course in courses:
+        if not course.processed:
+            combined_course = [str(course)]
+            course.processed = True
+            courses_with_same_number = \
+                [other_course for other_course in courses if course != other_course and not other_course.processed
+                 and (course.number == other_course.number or (400 <= course.number <= 499  # TODO prepare for CSCE 377/877
+                                                               and 800 <= other_course.number <= 899
+                                                               and other_course.number == course.number + 400))]
+            for other_course in courses_with_same_number:
+                should_crosslist = input(f'Crosslist {other_course} with {course}? [Y] ')
+                if len(should_crosslist) == 0:
+                    should_crosslist = 'y'
+                if should_crosslist[0].lower() == 'y':
+                    combined_course.append(str(other_course))
+                    other_course.processed = True
+            print(f'Including {combined_course}')
+            combined_courses.append(combined_course)
+    return combined_courses
+
+
+def get_enrollment(session: Session, semester_code: str, courses: List[str]):
+    enrollment = 0
+    for course in courses:
+        # noinspection PyTypeChecker
+        enrollment += len(session
+                          .query(distinct(Enrollment.nuid))
+                          .filter(Enrollment.semester_code == semester_code,
+                                  Enrollment.course == course,
+                                  Enrollment.dropped != 1)
+                          .all())
+    return enrollment
+
+
+def main():
+    username = input('Enter your username: ')
+    password = getpass('Enter your password: ')
+    semester_code = input('Enter semester code: ')
+    session = StudentsDatabase.connect(username, password)
+    courses = get_courses(session, semester_code)
+    for course in courses:
+        enrollment = get_enrollment(session, semester_code, course)
+        for index, subcourse in enumerate(course):
+            if index != 0:
+                print('/', end='')
+            print(subcourse, end='')
+        print(f', {enrollment}')
+
+
+if __name__ == '__main__':
+    main()

FindTAs.py

+from getpass import getpass
+from typing import List, Dict
+
+from sqlalchemy import distinct
+from sqlalchemy.orm import Session
+
+from StudentsDatabase import StudentsDatabase, Enrollment, semester_code_is_well_formed, CourseSchedule, Student
+
+
+def get_semester_codes(current_semester_code: str) -> List[str]:
+    initial_semester_code = ''
+    semester_code_is_valid = False
+    while not semester_code_is_valid:
+        initial_semester_code = input('What is the earliest semester code to examine? ')
+        semester_code_is_valid = semester_code_is_well_formed(initial_semester_code)
+        if semester_code_is_valid and initial_semester_code >= current_semester_code:
+            print(f'{initial_semester_code} must be earlier than {current_semester_code}.')
+            semester_code_is_valid = False
+    semester_codes = []
+    next_semester_code = initial_semester_code
+    while next_semester_code < current_semester_code:
+        semester_codes.append(next_semester_code)
+        century_year = int(next_semester_code[:-1])
+        month = int(next_semester_code[-1])
+        match month:
+            case 1:
+                month = 5
+            case 5:
+                month = 8
+            case 8:
+                month = 1
+            case _:
+                print('Reached unreachable code!')
+                exit(1)
+        if month == 1:
+            century_year += 1
+        next_semester_code = f'{century_year}{month}'
+    return semester_codes
+
+
+def get_allowable_grades() -> List[str]:
+    grade = ''
+    grade_is_well_formed = False
+    while not grade_is_well_formed:
+        grade = input('What is the minimum grade to consider? ').upper()
+        grade_is_well_formed = (1 <= len(grade) <= 2 and grade[0] in {'A', 'B', 'C', 'D', 'F'}
+                                and (grade[1] in {'+', '-'} if len(grade) == 2 else True))
+        if not grade_is_well_formed:
+            print(f'{grade} is not a valid grade.')
+    grades = [grade]
+    while grade != 'A+':
+        if len(grade) == 1:
+            grade = f'{grade}+'
+        elif grade[1] == '-':
+            grade = grade[0]
+        else:
+            grade = f'{chr(ord(grade[0]) - 1)}-'
+        grades.append(grade)
+    return grades
+
+
+# noinspection PyPep8Naming
+def get_candidate_TAs(session: Session, course_code: str,
+                      semester_codes: List[str], allowable_grades: List[str]) -> List[Dict[str, str]]:
+    # noinspection PyTypeChecker
+    students = (session
+                .query(Enrollment)
+                .filter(Enrollment.course == course_code,
+                        Enrollment.semester_code.in_(semester_codes),
+                        Enrollment.grade.in_(allowable_grades))
+                .all())
+    return [dict(nuid=str(student.nuid), grade=str(student.grade), semester_code=str(student.semester_code))
+            for student in students]
+
+
+# noinspection DuplicatedCode
+def filter_on_time_window(session: Session, semester_code: str,
+                          candidates: List[Dict[str, str]]) -> List[Dict[str, str]]:
+    day = ''
+    while day == '':
+        day = input('Which day are you looking for (M/T/W/R/F)? ').upper()
+        match day:
+            case 'M' | 'T' | 'W' | 'R' | 'F':
+                day = day
+            case 'MONDAY' | 'TUESDAY' | 'WEDNESDAY' | 'FRIDAY':
+                day = day[0]
+            case 'THURSDAY':
+                day = 'R'
+            case _:
+                print(f'{day} is not a valid day.')
+                day = ''
+    start_time = ''
+    while start_time == '':
+        start_time = input('What is the start time (use a 24-hour clock)? ').replace(':', '')
+        if not start_time.isdigit() or len(start_time) < 3 or len(start_time) > 4:
+            print(f'{start_time} is not a valid 24-hour time.')
+            start_time = ''
+        elif len(start_time) == 3:
+            start_time = f'0{start_time}'
+        if start_time and (int(start_time[0]) > 2 or int(start_time[2]) > 5):
+            print(f'{start_time} is not a valid 24-hour time.')
+            start_time = ''
+    end_time = ''
+    while end_time == '':
+        end_time = input('What is the ending time (use a 24-hour clock)? ').replace(':', '')
+        if not end_time.isdigit() or len(end_time) < 3 or len(end_time) > 4:
+            print(f'{end_time} is not a valid 24-hour time.')
+            end_time = ''
+        elif len(end_time) == 3:
+            end_time = f'0{end_time}'
+        if end_time and (int(end_time[0]) > 2 or int(end_time[2]) > 5):
+            print(f'{end_time} is not a valid 24-hour time.')
+            end_time = ''
+        if end_time < start_time:
+            print(f'The ending time must be later than the start time.')
+            end_time = ''
+    print(
+        'NOTE: Filtering will only check for School of Computing courses; there may yet be conflicts with other departments\' courses.')
+    # noinspection PyTypeChecker
+    candidate_enrollments = (session
+                             .query(Enrollment.nuid,
+                                    CourseSchedule.start_time,
+                                    CourseSchedule.end_time)
+                             .join(CourseSchedule,
+                                   (Enrollment.course == CourseSchedule.course) &
+                                   (Enrollment.section == CourseSchedule.section))
+                             .filter(Enrollment.nuid.in_([candidate['nuid'] for candidate in candidates]),
+                                     Enrollment.semester_code == semester_code,
+                                     CourseSchedule.days.like(f'%{day}%'))
+                             .distinct(CourseSchedule.start_time, CourseSchedule.end_time)
+                             .all())
+    # TODO: add filter for dropped != 1
+    filtered_candidates = []
+    for candidate in candidates:
+        has_schedule_conflict = False
+        enrollments = [enrollment for enrollment in candidate_enrollments if enrollment.nuid == candidate['nuid']]
+        for enrollment in enrollments:
+            enrollment_start_time = enrollment.start_time.replace(':', '')
+            enrollment_end_time = enrollment.end_time.replace(':', '')
+            if (enrollment_start_time <= start_time < enrollment_end_time
+                    or enrollment_start_time < end_time <= enrollment_end_time):
+                # TODO: troubleshoot -- this allowed at least one CSCE course 1230-1320 through
+                has_schedule_conflict = True
+        if not has_schedule_conflict:
+            filtered_candidates.append(candidate)
+    return filtered_candidates
+
+
+# noinspection DuplicatedCode
+def main():
+    username = input('Enter your username: ')
+    password = getpass('Enter your password: ')
+    session = StudentsDatabase.connect(username, password)
+    course_code = input('Which course are you searching? ').upper().replace(' ', '')
+    current_semester_code = ''
+    semester_code_is_valid = False
+    while not semester_code_is_valid:
+        current_semester_code = input('What is current semester code? ')
+        semester_code_is_valid = semester_code_is_well_formed(current_semester_code)
+    semester_codes = get_semester_codes(current_semester_code)
+    allowable_grades = get_allowable_grades()
+    # TODO: determine if we need to further restrict the semesters and/or grades
+    candidates = get_candidate_TAs(session, course_code, semester_codes, allowable_grades)
+    print('Number of matching candidates found...')
+    for semester_code in semester_codes:
+        if len([candidate for candidate in candidates if candidate['semester_code'] == semester_code]) > 0:
+            print(f'{semester_code} -- ', end='\t')
+            for grade in allowable_grades:
+                print(f'{grade} {len([candidate for candidate in candidates
+                                      if candidate['semester_code'] == semester_code and candidate['grade'] == grade])}',
+                      end='\t')
+            print()
+    looking_for_time_window = input('Are you looking for a specific time window? [N] ').upper()
+    if len(looking_for_time_window) > 0 and looking_for_time_window[0] == 'Y':
+        candidates = filter_on_time_window(session, current_semester_code, candidates)
+    # print('Number of matching candidates who have no known course conflicts...')
+    # for semester_code in semester_codes:
+    #     if len([candidate for candidate in candidates if candidate['semester_code'] == semester_code]) > 0:
+    #         print(f'{semester_code} -- ', end='\t')
+    #         for grade in allowable_grades:
+    #             print(f'{grade} {len([candidate for candidate in candidates
+    #                                   if candidate['semester_code'] == semester_code and candidate['grade'] == grade])}',
+    #                   end='\t')
+    #         print()
+    students = (session
+                .query(Student)
+                .filter(Student.nuid.in_([candidate['nuid'] for candidate in candidates]))
+                .distinct(Student.nuid)
+                .all())
+    # for student in students:
+    #     other_data = [candidate for candidate in candidates if candidate['nuid'] == student.nuid][0]
+    #     semester_code = other_data['semester_code']
+    #     grade = other_data['grade']
+    for candidate in candidates:
+        semester_code = candidate['semester_code']
+        grade = candidate['grade']
+        student = [student for student in students if student.nuid == candidate['nuid']][0]
+        print(f'{semester_code} {grade}\t{student.first_name} {student.last_name}')
+
+
+if __name__ == '__main__':
+    main()

StudentsDatabase.py

+from typing import Union
+
+from _mysql_connector import MySQLInterfaceError
+from mysql.connector import errors
+from sqlalchemy import create_engine, Column, Integer, String, Date, exc, ForeignKey
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker, Session #, relationship
+
+
+def semester_code_is_well_formed(current_semester_code: str) -> bool:
+    is_well_formed = (len(current_semester_code) == 4 and current_semester_code.isdigit()
+                      and current_semester_code[0] == '1' and current_semester_code[-1] in {'1', '5', '8'})
+    # we'll assume we don't want to go back to the 20th century
+    # we'll assume semesters only start in January, May, or August
+    if not is_well_formed:
+        print(f'{current_semester_code} is not a valid semester code. '
+              f'See https://registrar.unl.edu/academic-standards/policies/year-term-identifier/')
+    return is_well_formed
+
+
+Base = declarative_base()
+
+
+class Student(Base):
+    __tablename__ = 'students'
+    id = Column(Integer, primary_key=True)
+    semester_code = Column(String(4))
+    nuid = Column(String(11), unique=True)
+    first_name = Column(String(50))
+    middle_name = Column(String(255))
+    last_name = Column(String(50))
+    rest_name = Column(String(15))
+    enrolled = Column(Integer)
+    college = Column(String(15))
+    major = Column(String(15))
+    student_year = Column(String(10))
+    gpa = Column(String(15))
+    raiks_student = Column(Integer)
+    email = Column(String(255))
+    unl_uid = Column(String(255))
+    private = Column(Integer)
+    dropped = Column(Integer)
+    insert_date = Column(Date)
+    # Relationship to Enrollments
+    # enrollments = relationship("Enrollment", back_populates="student")
+
+
+class Enrollment(Base):
+    __tablename__ = 'enrollments'
+    id = Column(Integer, primary_key=True)
+    semester_code = Column(String(4))
+    nuid = Column(String(11), ForeignKey('students.nuid'))
+    course = Column(String(11))
+    section = Column(String(11))
+    grade = Column(String(5))
+    override = Column(String(25))
+    dropped = Column(Integer)
+    insert_date = Column(Date)
+    # Relationship to Student
+    # student = relationship("Student", back_populates="enrollments")
+    # Relationship to CourseSchedule
+    # noinspection PyTypeChecker
+    # course_schedule = relationship("CourseSchedule", back_populates="enrollments", foreign_keys=[course, section])
+
+
+class CourseSchedule(Base):
+    __tablename__ = 'course_schedules'
+    id = Column(Integer, primary_key=True)
+    academic_session_id = Column(Integer)
+    semester_code = Column(String(4))
+    call_number = Column(String(11))
+    course = Column(String(11))
+    section = Column(String(15))
+    title = Column(String(50))
+    topic = Column(String(255))
+    department = Column(String(15))
+    credits = Column(String(4))
+    enrollment = Column(Integer)
+    instructor = Column(String(50))
+    instructor_nuid = Column(String(12))
+    activity = Column(String(5))
+    building = Column(String(15))
+    room = Column(String(10))
+    days = Column(String(7))
+    start_time = Column(String(10))
+    end_time = Column(String(10))
+    active = Column(Integer)
+    hidden = Column(Integer)
+    override = Column(Integer)
+    insert_date = Column(Date)
+    # Relationship to Enrollments
+    # enrollments = relationship("Enrollment", back_populates="course_schedule")
+
+
+class StudentsDatabase(object):
+    @staticmethod
+    def connect(username: str, password: str,
+                host: str = 'cse-apps.unl.edu', port: int = 3306, database: str = 'students') -> Session:
+        url = StudentsDatabase.construct_mysql_url(host, port, database, username, password)
+        try:
+            students_database = StudentsDatabase(url)
+            session = students_database.create_session()
+        except exc.ProgrammingError | errors.ProgrammingError | MySQLInterfaceError as error:
+            # we don't seem to actually be able to catch the exception!
+            print('Could not connect to the database.')
+            print(error)
+            exit(1)
+        return session
+
+    @staticmethod
+    def construct_mysql_url(authority: str, port: Union[int, str], database: str, username: str, password: str) -> str:
+        return f'mysql+mysqlconnector://{username}:{password}@{authority}:{port}/{database}'
+
+    @staticmethod
+    def construct_in_memory_url() -> str:
+        return 'sqlite:///'
+
+    def __init__(self, url):
+        self.engine = create_engine(url)
+        self.Session = sessionmaker()
+        self.Session.configure(bind=self.engine)
+
+    def ensure_tables_exist(self) -> None:
+        Base.metadata.create_all(self.engine)
+
+    def create_session(self) -> Session:
+        return self.Session()