"""Assessment administration interface implementation.
This module implements the Web user interface for actions relating to
scheduling and administering examinations by non-course units responsible for
those activities.
"""
import csv
from functools import partial
from io import StringIO
from itertools import count, groupby
from operator import attrgetter, itemgetter
from datetime import datetime, date, timedelta
from ll.xist.ns import html
from uw.dbtools import copy_to_csv
from uw.web.html.bootstrapform import render_bootstrap_date_range_selector, parse_bootstrap_date
from uw.web.html.form import render_hidden, render_select
from uw.web.html.format import format_return, make_table_format, format_datetime, format_date, format_time
from uw.web.wsgi import status
from uw.web.wsgi.delegate import delegate_get_post, delegate_action
from uw.web.wsgi.form import use_form_param
from uw.web.wsgi.function import return_html, return_csv, return_text_download
from uw.local.dbtools import person_delegate
from ...util.format import format_person
from ...util.identity import use_remote_identity
from ..db.cyon import UWExamsUEFFile
from ..db.exam_import_times import parse_exam_times_upload, process_exam_times_upload
from ..db.exam_import_rooms import parse_exam_rooms_upload, process_exam_rooms_upload
from .integration import format_integration
from .ui import format_duration, format_sections_multiline
[docs]def scheduled_administered_examinations_query (term_id, admin_id, admin_search_type):
query = "select exam_id, term_id, admin_id, exam_print_deadline_find_deadline_date (exam_id) as deadline, admin_sections, enrolment, quest_enrol_limit, exam_duration, primary_sitting_id, sequence_assigned, master_count, masters_uploaded, masters_upload, master_approved, master_approver_person_id, master_accepted, exam_assigned, exam_exam_full_title (exam_id) AS full_title, primary_start_time, scanning_integration, crowdmark_exam_code, markbox_exam_code " \
"from exam_exam_plus " \
"join teaching_admin_term_plus using (term_id, admin_id) " \
"left join (select exam_id, count(*) as masters_upload, count(*) filter (where master_pdf_raw is not null) as masters_uploaded from exam_exam_master group by exam_id) as eem using (exam_id) " \
"where (term_id, %s) = (%s, %s) order by deadline nulls first, primary_start_time, full_title" % (admin_search_type, term_id, admin_id)
return query
@delegate_get_post
@return_html
def exam_scheduled_index (cursor, term, admin, roles):
"""Scheduled examination list URL handler.
Displays list of scheduled examinations with links to each.
"""
if not 'ADMIN' in roles:
raise status.HTTPForbidden ()
result = [format_return ('Main Menu', None, None, 'Offering')]
result.append (html.p (html.a ('Download CSV', href="./list")))
if admin.admin_id == 20050:
result.append (html.p (html.a ('Download Cyon Export File', href="cyon-export")))
result.append (html.p (html.a ('Upload Cyon Import File', href="cyon-import")))
result.append (html.p (html.a ('Upload Room Locations File', href="rooms-import")))
result.append (html.p (html.a ('Download Final Assessment Schedule', href="schedule-publish")))
result.append (html.p (html.a ('Students with conflicts', href="conflict/")))
exams = cursor.execute_tuples (scheduled_administered_examinations_query (term_id=term.code (), admin_id=admin.admin_id, admin_search_type="schedule_admin_id"))
sequence = count (1)
table = (make_table_format (
('#', lambda r: next (sequence), lambda r: {'style': "text-align: right;"}),
('Assessment', format_exam_column),
('Sections', format_sections),
('Enrolment', format_enrolment),
('Start Time', lambda r: format_datetime (r.primary_start_time)),
('Duration', lambda r: format_duration (r.exam_duration)),
('Assigned Seating', lambda r: 'Yes' if r.exam_assigned else 'No'),
) (exams))
result.append (table)
return "%s %s Scheduled Assessments" % (admin.admin_description, term.description ()), result
@delegate_get_post
@return_csv
def exam_scheduled_csv_handler (cursor, term, admin, roles):
if not 'ADMIN' in roles:
raise status.HTTPForbidden ()
exams = cursor.execute_tuples (scheduled_administered_examinations_query (term_id=term.code (), admin_id=admin.admin_id, admin_search_type="schedule_admin_id"))
result = StringIO ()
writer = csv.writer (result)
writer.writerow (['#', 'Assessment', 'Sections', 'Enrolment', 'Capacity', 'Start Time', 'Duration', 'Assigned Seating'])
sequence = count (1)
for exam in exams:
writer.writerow ([
next (sequence),
exam.full_title,
format_sections_csv (exam.admin_sections),
exam.enrolment,
exam.quest_enrol_limit,
format_datetime (exam.primary_start_time),
format_duration (exam.exam_duration),
exam.exam_assigned
])
return [result.getvalue (), '%s (%s) Scheduled Assessments.csv' % (admin.admin_description, term.description ())]
@delegate_get_post
@return_text_download
def exam_cyon_handler (cursor, term, admin, roles):
"""Generate file for import into Cyon for examination scheduling.
"""
if admin.admin_id != 20050:
raise status.HTTPNotFound ()
result = StringIO ()
uefwriter = UWExamsUEFFile (cursor, term)
uefwriter.write (result)
return result.getvalue (), 'Odyssey Cyon Export (%s) as of %s.uef' % (term.description (), format_datetime (datetime.now ()))
@return_html
def exam_cyon_times_upload_get_handler (cursor, term, admin, roles):
"""Cyon examination times upload handler.
Shows a form to upload the Cyon times upload file.
"""
if admin.admin_id != 20050:
raise status.HTTPNotFound ()
result = [format_return ('Main Menu', None, None, 'Offering', dot='Scheduled Assessments')]
result.append (html.form (
html.p (
'Select Cyon output file to upload: ',
html.input (type="file", name="upload"),
' ',
html.input (type="submit", name="!upload", value="Upload!"),
),
method="post", enctype="multipart/form-data", action=""
))
return 'Upload Cyon Assessment Times', result
@use_form_param
@return_html
def exam_cyon_times_upload_post_handler (cursor, term, admin, roles, form):
"""Cyon examination times upload handler.
Accepts the uploaded file and processes its contents.
"""
if admin.admin_id != 20050:
raise status.HTTPNotFound ()
elif '!upload' in form:
upload = form.optional_field ('upload')
if upload is not None and (upload.filename or upload.value):
result, errors = parse_exam_times_upload (upload.value.decode ())
if result is None:
warnings = []
else:
warnings = process_exam_times_upload (cursor, term, result)
result = [format_return ('Main Menu', None, None, 'Offering', dot='Scheduled Assessments')]
result.append (html.p ('The upload has been processed.'))
if errors:
result.append (html.h2 ('Errors'))
result.append (make_table_format (
('Line', itemgetter (0)),
('Error', itemgetter (1))
) (errors))
if warnings:
result.append (html.h2 ('Warnings'))
result.append (make_table_format (
('Line', itemgetter (0)),
('Warning', itemgetter (1))
) (warnings))
return 'Upload Cyon Assessment Times', result
raise status.HTTPFound ("")
@return_html
def exam_rooms_upload_get_handler (cursor, term, admin, roles):
"""Cyon examination times upload handler.
Shows a form to upload the Cyon times upload file.
"""
if admin.admin_id != 20050:
raise status.HTTPNotFound ()
result = [format_return ('Main Menu', None, None, 'Offering', dot='Scheduled Assessments')]
result.append (html.form (
html.p (
'Select room locations file to upload: ',
html.input (type="file", name="upload"),
' ',
html.input (type="submit", name="!upload", value="Upload!"),
),
method="post", enctype="multipart/form-data", action=""
))
return 'Upload Scheduled Assessment Rooms', result
@use_form_param
@return_html
def exam_rooms_upload_post_handler (cursor, term, admin, roles, form):
"""Cyon examination times upload handler.
Accepts the uploaded file and processes its contents.
"""
if admin.admin_id != 20050:
raise status.HTTPNotFound ()
elif '!upload' in form:
upload = form.optional_field ('upload')
if upload is not None and (upload.filename or upload.value):
parsed, errors = parse_exam_rooms_upload (upload.value.decode ())
result = [format_return ('Main Menu', None, None, 'Offering', dot='Scheduled Assessments')]
if parsed is None:
warnings = None
result.append (html.p ('The upload could not be processed.'))
else:
warnings = process_exam_rooms_upload (cursor, term, parsed)
result.append (html.p ('The upload has been processed.'))
if errors:
result.append (html.h2 ('Errors'))
result.append (make_table_format (
('Line', itemgetter (0)),
('Column', itemgetter (1)),
('Error', itemgetter (2)),
) (errors))
if warnings:
result.append (html.h2 ('Warnings'))
result.append (make_table_format (
('Line', itemgetter (0)),
('Warning', itemgetter (1)),
) (warnings))
return 'Upload Scheduled Assessment Rooms', result
raise status.HTTPFound ("")
@return_html
def exam_conflict_get_handler (cursor, term, admin, roles):
"""Student schedule issue display GET handler.
Displays student with schedule issues.
"""
result = [format_return ('Main Menu', None, None, 'Offering', 'Scheduled Assessments')]
students = groupby (cursor.execute_tuples ("select *, exam_exam_full_title (exam_id) as exam_full_title, exam_exam_full_title (next_exam_id) as next_full_title from ro_exam_back_to_back natural join person_identity_complete where term_id = %(term_id)s and %(admin_id)s in (schedule_admin_id, next_schedule_admin_id) order by surname, givennames, uw_id, start_time, exam_id", term_id=term.code (), admin_id=admin.admin_id), attrgetter ("person_id"))
students = [list (g) for _, g in students]
if students:
result.append (html.p ('%d students have schedule issues related to assessments scheduled by %s:' % (len (students), admin.admin_description)))
table = html.table (
html.tr (html.th (h) for h in ['UW ID', 'Student', 'Previous', 'When', 'Next', 'When', 'Conflict'])
)
for conflicts in students:
for i, c in (enumerate (conflicts)):
row = html.tr ()
if i == 0:
row.append (html.td (html.a (c.uw_id, href="%d" % c.person_id), rowspan=len (conflicts)))
row.append (html.td (format_person (cursor, c.person_id), rowspan=len (conflicts)))
row.append (html.td (c.exam_full_title))
row.append (html.td (format_times (c.start_time, c.end_time)))
row.append (html.td (c.next_full_title))
row.append (html.td (format_times (c.next_start_time, c.next_end_time)))
row.append (html.td ('Yes' if c.conflict else None))
table.append (row)
result.append (table)
else:
result.append (html.p ('No students have schedule issues.'))
return 'Students with Assessment Schedule Issues', result
@return_html
def exam_conflict_student_get_handler (cursor, term, admin, roles, person):
"""Candidate examination sitting editor GET handler.
"""
result = [format_return ('Main Menu', None, None, 'Offering', 'Scheduled Assessments', dot='Conflicts')]
exams = cursor.execute_tuples ("select exam_id, start_time, schedule_admin_id, exam_exam_full_title (exam_id), count(*) over (partition by term_id, person_id, start_time) > 1 as conflict from exam_exam_student_sitting natural join exam_exam natural join exam_sitting where (term_id, person_id) = (%(term_id)s, %(person_id)s) and schedule_admin_id is not null order by 2, 3", term_id=term.code (), person_id=person.person_id)
sittings = cursor.execute_tuples ("select sitting_id, full_description from exam_sitting_plus join teaching_admin_contains_complete on (sitting_admin_id = inner_admin_id) where (sitting_term_id, outer_admin_id) = (%(term_id)s, %(admin_id)s) order by full_description", term_id=term.code (), admin_id=admin.admin_id)
result.append (html.form (
make_table_format (
('Exam ID', attrgetter ('exam_id')),
('Start Time', format_exam_start_time),
('Exam Title', attrgetter ('exam_exam_full_title')),
('Update Sitting', lambda r: render_select ("sitting-%d" % r.exam_id, sittings) if r.schedule_admin_id == admin.admin_id else None),
) (exams),
html.input (type="submit", name="!update", value="Update!"),
method="post", action=""
))
return 'Conflicts for %s, %s (%s)' % (person.surname, person.givennames, person.uw_id), result
@use_form_param
@use_remote_identity
@return_html
def exam_conflict_student_post_handler (cursor, term, admin, roles, person, form, remote_identity):
"""Candidate examination sitting editor POST handler.
"""
if "!update" in form:
exams = cursor.execute_values ("select exam_id from exam_exam_student_sitting natural join exam_exam natural join exam_sitting where (term_id, person_id) = (%(term_id)s, %(person_id)s) and schedule_admin_id is not null", term_id=term.code (), person_id=person.person_id)
updates = {}
for exam_id in exams:
sitting_id = form.optional_field_value ("sitting-%d" % exam_id)
if sitting_id:
updates[exam_id] = int (sitting_id)
request_id = cursor.callproc_required_value ("accom_create_person_request", term.code (), admin.admin_id, remote_identity.person_id, person.person_id, exams)
for exam_id, sitting_id in updates.items ():
cursor.execute_none ("update accom_request_candidate set sitting_id = %(sitting_id)s where (term_id, accom_admin_id, request_id, exam_id, person_id) = (%(term_id)s, %(admin_id)s, %(request_id)s, %(exam_id)s, %(person_id)s)", term_id=term.code (), admin_id=admin.admin_id, request_id=request_id, exam_id=exam_id, sitting_id=sitting_id, person_id=person.person_id)
cursor.callproc_none ("exam_exam_attach_sitting", exam_id, sitting_id)
cursor.callproc_none ("accom_request_completed", term.code (), admin.admin_id)
raise status.HTTPFound ("")
@return_csv
def exam_schedule_publish_get_handler (cursor, term, admin, roles):
"""Schedule publication GET handler.
"""
result = copy_to_csv (cursor, "select * from ro_exam_schedule_publish (%d)" % term.numericCode ())
return [result.getvalue (), '%s Final Assessment Schedule.csv' % term.description ()]
exam_scheduled_handler = delegate_action (exam_scheduled_index, {
'list': exam_scheduled_csv_handler,
'conflict': person_delegate (
delegate_get_post (exam_conflict_get_handler),
delegate_get_post (exam_conflict_student_get_handler, exam_conflict_student_post_handler)),
'cyon-export': exam_cyon_handler,
'cyon-import': delegate_get_post (exam_cyon_times_upload_get_handler, exam_cyon_times_upload_post_handler),
'rooms-import': delegate_get_post (exam_rooms_upload_get_handler, exam_rooms_upload_post_handler),
'schedule-publish': delegate_get_post (exam_schedule_publish_get_handler),
})
@delegate_get_post
@return_html
def exam_administered_index (cursor, term, admin, roles):
"""Scheduled examination list URL handler.
Displays list of scheduled examinations with links to each.
"""
if not 'ADMIN' in roles:
raise status.HTTPForbidden ()
result = [format_return ('Main Menu', None, None, 'Offering')]
result.append (html.p (html.a ('Add/edit Assessment Deadlines', href="../exams-deadline")))
result.append (html.p (html.a ('Download CSV', href="./list")))
exams = groupby (cursor.execute_tuples (scheduled_administered_examinations_query (term_id=term.code (), admin_id=admin.admin_id, admin_search_type="administer_admin_id")), attrgetter('deadline'))
for deadline, group_exams in exams:
if deadline is None:
result.append (html.h2 ("Exams with no Deadline"))
else:
result.append (html.h2 ("Deadline: %s" % format_date(deadline)))
sequence = count (1)
table = (make_table_format (
('#', lambda r: next (sequence), lambda r: {'style': "text-align: right;"}),
('Assessment', format_exam_column),
('Sections', format_sections),
('Enrolment', format_enrolment),
('Start Time', lambda r: format_datetime (r.primary_start_time)),
('Duration', lambda r: format_duration (r.exam_duration)),
('Assigned Seating', lambda r: 'Yes' if r.exam_assigned else 'No'),
('Integration', partial (format_integration, url="dashboard")),
('Finalized', lambda r: format_datetime (r.sequence_assigned)),
('Versions', attrgetter ('master_count')),
('Uploaded', format_masters_uploaded),
('Approved', lambda r: format_datetime (r.master_approved)),
('Approver', lambda r: format_person (cursor, r.master_approver_person_id)),
('Accepted', lambda r: format_datetime (r.master_accepted)),
) (group_exams))
result.append (table)
return "%s %s Administered Assessments" % (admin.admin_description, term.description ()), result
@delegate_get_post
@return_csv
def exam_administered_csv_handler (cursor, term, admin, roles):
if not 'ADMIN' in roles:
raise status.HTTPForbidden ()
exams = groupby (cursor.execute_tuples (scheduled_administered_examinations_query (term_id=term.code (), admin_id=admin.admin_id, admin_search_type="administer_admin_id")), attrgetter('deadline'))
result = StringIO ()
writer = csv.writer (result)
writer.writerow (['#', 'Assessment', 'Sections', 'Enrolment', 'Capacity', 'Start Time', 'Duration', 'Deadline', 'Assigned Seating', 'Integration', 'Finalized', 'Versions', 'Uploaded', 'Approved', 'Approver', 'Accepted'])
sequence = count (1)
for deadline, group_exams in exams:
for exam in group_exams:
writer.writerow ([
next (sequence),
exam.full_title,
format_sections_csv (exam.admin_sections),
exam.enrolment,
exam.quest_enrol_limit,
format_datetime (exam.primary_start_time),
format_duration (exam.exam_duration),
format_datetime (exam.deadline),
exam.exam_assigned,
exam.scanning_integration,
format_datetime (exam.sequence_assigned),
exam.master_count,
format_masters_uploaded (exam),
format_datetime (exam.master_approved),
format_person (cursor, exam.master_approver_person_id),
format_datetime (exam.master_accepted)
])
return [result.getvalue (), '%s (%s) Administered Assessments.csv' % (admin.admin_description, term.description ())]
exam_administered_handler = delegate_action (exam_administered_index, {
'list': exam_administered_csv_handler
})
@return_html
def exam_deadline_index_get_handler (cursor, term, admin, roles):
"""Assessment deadline list URL handler.
Displays list of deadlines for the admin and term with links to add/edit deadlines.
"""
if not 'ADMIN' in roles:
raise status.HTTPForbidden ()
result = [[format_return ('Main Menu', None, None, 'Offering')]]
result.append (html.p (html.a ('Return to administered examinations page', href="../exams-administered/")))
current_term = cursor.execute_required_value ("select term_dates @> current_date from uw_term where term_id = %(term_id)s", term_id=term.code ())
deadlines = cursor.execute_tuples ("select * from exam_print_deadline where (term_id, administer_admin_id) = (%(term_id)s, %(administer_admin_id)s) order by period_id", term_id=term.code (), administer_admin_id=admin.admin_id)
next_period_id = max ([1] + [deadline.period_id + 1 for deadline in deadlines])
result.append (html.p (html.form (html.input (type="submit", name="!edit", value="Add New Deadline"),
method="get", action="%s" % next_period_id))) if current_term else None
table = (make_table_format (
('Period', lambda r: r.period_id),
('Deadline Date', lambda r: format_date (r.deadline_date)),
('Printing Date', lambda r: format_date (r.printing_date)),
('Delivery Date', lambda r: format_date (r.delivery_date)),
('Period Begin Date', lambda r: format_date (r.period_begin_date)),
('Processed', lambda r: format_datetime (r.deadline_processed) if r.deadline_processed
else html.a ('Edit Deadline…', href="%s" % r.period_id) if current_term else None),
) (deadlines))
result.append (table)
return "%s %s Assessment Deadlines" % (admin.admin_description, term.description ()), result
@use_form_param
@return_html
def exam_deadline_index_post_handler (cursor, term, admin, roles, form):
"""Assessment deadline editor URL POST handler.
Implements an upsert action based on form results
If an error occurs, the error message is shown to the user.
"""
if not 'ADMIN' in roles:
raise status.HTTPForbidden ()
current_term = cursor.execute_required_value ("select term_dates @> current_date from uw_term where term_id = %(term_id)s", term_id=term.code ())
if not current_term:
raise status.HTTPFound (".")
period_id = form.required_field_value ("period_id")
period_begin_date = form.optional_field_value ("period_begin_date")
deadline_date = form.optional_field_value ("deadline_date")
printing_date = form.optional_field_value ("printing_date")
delivery_date = form.optional_field_value ("delivery_date")
prev_deadline_date = form.optional_field_value ("prev_deadline_date")
next_deadline_date = form.optional_field_value ("next_deadline_date")
if period_begin_date is None or deadline_date is None or printing_date is None or delivery_date is None:
return 'Error: Dates Blank', html.p ('Please fill in all dates in order to submit the deadline.')
elif (prev_deadline_date and prev_deadline_date > deadline_date) or (next_deadline_date and next_deadline_date < deadline_date):
return 'Error: Conflicting Deadline Dates', html.p ('The deadline date conflicts with current deadline dates. ', html.br ()
('Previous deadline is %s ' % (prev_deadline_date)) if prev_deadline_date else None, html.br (),
('Next deadline is %s' % (next_deadline_date)) if next_deadline_date else None, html.br (),
'Please go back and adjust the given deadline date %s.' %(deadline_date))
elif deadline_date > printing_date or printing_date > delivery_date or delivery_date > period_begin_date:
return 'Error: Date Order', html.p ('Error in deadline dates please go back and fix.',
html.br (), 'Deadline dates must be in the following order:',
html.br (), 'Period Begin Date > Delivery Date > Printing Date > Deadline Date')
if "!submit" in form:
cursor.execute_none ("insert into exam_print_deadline (term_id, administer_admin_id, period_id, period_begin_date, deadline_date, printing_date, delivery_date) values (%(term_id)s, %(administer_admin_id)s, %(period_id)s, %(period_begin_date)s, %(deadline_date)s, %(printing_date)s, %(delivery_date)s) on conflict (term_id, administer_admin_id, period_id) do update set period_begin_date = excluded.period_begin_date, deadline_date = excluded.deadline_date, printing_date = excluded.printing_date, delivery_date = excluded.delivery_date", term_id=term.code (), administer_admin_id=admin.admin_id, period_id=period_id, period_begin_date=period_begin_date, deadline_date=deadline_date, printing_date=printing_date, delivery_date=delivery_date)
raise status.HTTPFound (".")
exam_deadline_index_handler = delegate_get_post (exam_deadline_index_get_handler, exam_deadline_index_post_handler)
@return_html
def exam_deadline_edit_get_handler (cursor, term, admin, roles, period_id):
"""Assessment deadline editor URL GET handler.
Displays the editor for adding and editing deadline dates.
"""
if not 'ADMIN' in roles:
raise status.HTTPForbidden ()
current_term = cursor.execute_required_value ("select term_dates @> current_date from uw_term where term_id = %(term_id)s", term_id=term.code ())
if not current_term:
raise status.HTTPFound (".")
result = [format_return (dot='Assessment Deadlines')]
deadline = cursor.execute_optional_tuple ("select * from exam_print_deadline where (term_id, administer_admin_id, period_id) = (%(term_id)s, %(administer_admin_id)s, %(period_id)s)", term_id=term.code (), administer_admin_id=admin.admin_id, period_id=period_id)
prev_deadline_date, next_deadline_date = cursor.execute_optional_tuple ("select (select deadline_date from exam_print_deadline where (term_id, administer_admin_id, period_id) = (%(term_id)s, %(administer_admin_id)s, %(period_id)s - 1)) as prev_deadline_date, (select deadline_date from exam_print_deadline where (term_id, administer_admin_id, period_id) = (%(term_id)s, %(administer_admin_id)s, %(period_id)s + 1)) as next_deadline_date", term_id=term.code (), administer_admin_id=admin.admin_id, period_id=period_id)
select_dates = cursor.execute_required_tuple("select (lower (term_dates) + '2 months'::interval)::date as begin_date, upper (term_exam_dates) as end_date from uw_term where term_id = %(term_id)s", term_id=term.code ())
control_rows = [('Deadline Date', 'deadline_date'), ('Printing Date', 'printing_date'), ('Delivery Date', 'delivery_date'), ('Period Begin Date', 'period_begin_date')]
result.append (html.form (
html.table (
html.tr (html.th (title),
html.td (render_bootstrap_date_range_selector(controls, end=next_deadline_date - timedelta(days=1) if next_deadline_date and controls=='deadline_date' else select_dates.end_date, start=prev_deadline_date + timedelta(days=1) if prev_deadline_date else select_dates.begin_date, current= getattr(deadline, controls) if deadline else None))
) for title, controls in control_rows),
html.p (html.input (type="submit", name="!submit", value="Submit!")),
render_hidden ("period_id", period_id),
render_hidden ("prev_deadline_date", format_date (prev_deadline_date)),
render_hidden ("next_deadline_date", format_date (next_deadline_date)),
method="post", action="."
))
return "%s %s Assessment Deadlines" % (admin.admin_description, term.description ()), result