"""Automatic creation of print jobs for examinations.
Routines for creating print requisitions for individual examinations and
entire Registrar's Office print periods.
"""
from datetime import date
from fnmatch import fnmatch
from math import ceil
from operator import attrgetter
from os import listdir
from pathlib import Path
from subprocess import check_output, STDOUT
from tempfile import mkdtemp
import os.path
import traceback
from uw.emailutils import emailaddr
from uw.web.html.join import english_join
from ...print_.db.nms_requisition import nms_requisition
from ...util.mail import odyssey_email, ro_email
from ...outgoing.make_mail import mailout
from .crowdmark import crowdmark_exam
from .markbox import markbox_exam
odyssey_ro_email = emailaddr ('RO Final Assessment Processing', 'instruct-ro-exams@odyssey.uwaterloo.ca')
paper_colours = {
'A': 'White',
'Q': 'Orchid',
'R': 'Yellow',
}
[docs]def message_to_course_staff (cursor, message, term_id, admin_id, role_codes, recipient_type='To'):
for s in cursor.execute_values ("select distinct person_id from auth_offering_personnel_complete where (term_id, admin_id) = (%(term_id)s, %(admin_id)s) and role_code = any (%(role_codes)s) and role_current and not auth_backup", term_id=term_id, admin_id=admin_id, role_codes=role_codes):
message.add_recipient_person (recipient_type, s)
[docs]def format_pages (master, include_duplex, include_size):
"""Format a brief description of the pages required for the given master.
The "master" parameter is a row of exam_exam_master containing at least the
fields master_pages_gross, and size_code.
The output is a brief English description of the number of pages in the
master, and optionally the duplexing and paper size.
"""
items = ['%d pages' % master.master_pages_gross]
if include_duplex:
items.append (format_duplex (master.master_duplex))
if include_size:
items.append (format_size (master.size_code))
return ' '.join (items)
[docs]def find_paper_colour (documents):
"""Format a brief description of the pages required for the given masters.
:param documents: A database row with information about a master type.
:return: A brief English description of the number of pages in the
masters, together with duplexing and page size information if it
differs between the masters.
This routine describes the paper required to print the masters in the event
the specific fields on the print requisition are insufficient to describe
the situation.
"""
documents = [document for document in documents if document is not None]
include_duplex = len (set (map (attrgetter ('master_duplex'), documents))) > 1
include_size = len (set (map (attrgetter ('size_code'), documents))) > 1
return ', '.join ('%s %s' % (format_pages (document, include_duplex, include_size), paper_colours[document.type_code].lower ()) for document in documents)
[docs]def append_file (files, java_dir, filename):
"""Append contents of a file to the specified list of files.
Parameters:
files -- list of files to which to append file contents;
java_dir -- directory in which to find file;
filename -- filename within java_dir.
This routine reads the contents of [java_dir]/[filename] and appends a
tuple of (filename, contents) to the provided list. This is a utility
routine for use by submit_exam_print_job.
"""
files.append ((filename, (java_dir / filename).read_bytes ()))
[docs]def submit_exam_print_job (cursor, exam):
"""Submit a print job for the specified examination.
:param cursor: DB cursor.
:param exam: examination to print (exam_exam row).
Invokes the Java prepareExamPrintJob class to create the output files for
the given examination, then creates a print job within the Print
application to print the papers.
If the examination is RO-administered, a blank account number is used, the
contact person is hardcoded to an RO staff member, no delivery information
is provided, the box-labelling and OPD/Relief pulls are included in the
instructions, and a copy of each Answers master PDF is included for the
emergency duplication file folder.
**TODO: Should do more checking of status (nobody unseated, etc.)**
"""
if exam.sequence_assigned is None:
raise
if exam.master_approved is None:
raise
if exam.master_accepted is not None:
raise
# Create PDF files
java_dir = Path (mkdtemp ())
check_output (['java', '-mx2G', 'ca.uwaterloo.odyssey.exams.prepareExamPrintJob', str (exam.exam_id), java_dir], stderr=STDOUT)
# Determine files to include in print job
versioned = []
reference = None
duplex = set ()
paper_sizes = set ()
pages = 0
files = []
has_aas = False
type_codes = set ()
for master in cursor.exam_master_types_by_exam (exam_id=exam.exam_id):
type_codes.add (master.type_code)
pages += master.master_pages_gross
paper_sizes.add (master.size_code)
if master.type_code in ['A', 'Q']:
# Answers or Questions pages
versioned.append (master)
for filename in listdir (java_dir):
if fnmatch (filename, '%s*.pdf' % master.type_description):
append_file (files, java_dir, filename)
has_aas = has_aas or filename == '%s-AAS.pdf' % master.type_description
elif master.type_code == 'R':
# Reference sheet
reference = master
append_file (files, java_dir, 'reference.pdf')
else:
# Unknown master type
raise
duplex.add (master.master_duplex)
if versioned == [] and reference is None:
raise
ro_print = exam.administer_admin_id == 20050
# Find Customer Information
if ro_print:
# TODO: Should not hardcode to Annette Ertel
person_id = cursor.execute_required_value ("select person_id from person_identity_complete where userid = 'a3ertel'")
else:
person_id = cursor.callproc_optional_value ("exam_print_contact", exam.exam_id)
if person_id is None:
raise
account = cursor.callproc_optional_value ("exam_print_account", exam.exam_id)
if account is None:
raise
# Find Job Information
deadline = cursor.callproc_optional_value ("exam_print_deadline", exam.exam_id)
if deadline is None:
raise
# Create print job
job = nms_requisition (cursor)
# Set Customer Information
job.set_contact (person_id)
if account is not None:
job.set_flexfield (account)
# Set Job Information
job.set_title (exam.full_title)
job.set_date_required (deadline)
job.set_end_use ('teaching')
# Shipping Information
if not ro_print:
if exam.administer_admin_id is None:
job.set_courier (exam.exam_author_person_id)
else:
job.set_courier (person_id)
# Copying/Printing
total_copies = exam.count_occupied_seats + exam.count_used_rush_seats + exam.spare_count
job.set_copyprint (
pages=pages,
copies=total_copies,
simplex=False in duplex,
duplex=True in duplex,
paper_bond=True,
size_letter='LTR' in paper_sizes,
size_legal='LGL' in paper_sizes,
paper_white='A' in type_codes,
paper_other=find_paper_colour (versioned + [reference]) if type_codes - set (['A']) else None,
ink_black=True,
)
# Binding/Finishing
job.set_binding (stapling='corner')
# Special Instructions
# None needed for midterms, typically
# Need location etc. (box label) for RO finals
# Need something else (?) for CEL finals
instructions = []
for document in versioned:
if document.master_duplex:
sheets = ceil (document.master_pages_gross / 2.0)
else:
sheets = document.master_pages_gross
instructions += [
"Print %(document)s*.pdf %(plex)s on %(colour)s, stapling every %(sheets)d sheets." % {'document': document.type_description, 'plex': format_duplex (document.master_duplex), 'colour': paper_colours[document.type_code].lower (), 'sheets': sheets},
"",
]
if reference:
instructions += [
"Print stapled copies of reference.pdf %(plex)s on yellow." % {'plex': format_duplex (reference.master_duplex)},
"",
]
if not ro_print and has_aas:
instructions += [
"Deliver *-AAS.pdf to AAS Reception, NH 1401. Deliver other papers according to Shipping Information above.",
"",
]
if ro_print:
divisions = cursor.callproc_tuples ("ro_print_divisions", exam.exam_id)
for division in divisions:
if division.room_delivery != 'RO':
details = division.admin_description
else:
details = "%s %s %s" % (division.room_depot, format_ro_times (exam.primary_start_time, exam.exam_duration), division.rooms)
instructions.append ("%d copies: %s" % (division.copies, details))
total_copies -= division.copies
# Total number of copies added up over print divisions must equal
# total obtained directly from examination
if total_copies:
raise
# Include emergency duplication master copies
for document in versioned:
for suffix, content in cursor.execute_tuples ("select exam_exam_master_suffix (exam_id, type_code, version_seq) AS suffix, master_pdf from exam_exam_master where (exam_id, type_code) = (%(exam_id)s, %(type_code)s)", exam_id=exam.exam_id, type_code=document.type_code):
suffix = '' if suffix is None else '-%s' % suffix
filename = '%s-master%s.pdf' % (document.type_description, suffix)
job.attach_file (filename, content)
job.set_instructions ('\n'.join (instructions))
for file, content in files:
job.attach_file (file, content)
job.ready ()
cursor.execute_none ("update exam_exam set master_accepted = now () where exam_id = %(exam_id)s and master_accepted is null", exam_id=exam.exam_id)
return job.job_id, java_dir
[docs]def print_examination (cursor, exam):
try:
if exam.scanning_integration == 'M' and exam.markbox_answer_page_pdf is None:
print ("%s %s submitting to Markbox" % (exam.primary_start_time, exam.full_title))
markbox_exam (cursor, exam)
elif exam.scanning_integration != 'C' or exam.crowdmark_qrcodes_pdf is not None:
print ("%s %s printing" % (exam.primary_start_time, exam.full_title))
(job_id, output_dir) = submit_exam_print_job (cursor, exam)
print ("Job ID %d, Output in %s" % (job_id, output_dir))
elif cursor.execute_required_value ("select exists (select * from crowdmark_create_request where exam_id = %(exam_id)s and response_status = 201)", exam_id=exam.exam_id):
print ("%s %s already submitted to Crowdmark, no QR Codes" % (exam.primary_start_time, exam.full_title))
else:
print ("%s %s submitting to Crowdmark" % (exam.primary_start_time, exam.full_title))
crowdmark_exam (cursor, exam)
cursor.connection.commit ()
except Exception as x:
cursor.connection.rollback ()
print (traceback.format_exc ())
contact_info_text = [
'',
'Please make note of the following emergency telephone numbers:',
' Examinations Office, ext. 49982 or 42711 (Monday - Friday, 8:30 a.m. - 4:30 p.m.)',
' Annette Ertel (Evenings and Saturday only 226-600-5951)',
'If you have any questions about this matter, please give me a call at ext. 48619.',
''
]
important_note_text = [
'Important Notes:',
' 1. Students are prohibited from consuming food and drinks (with the exception of water in a clear bottle with no label) during their final examinations. Students can, however arrange for a short nutrition break with a Proctor outside the exam venue during the final exam.',
' If a single, short break is not sufficient because a student is medically required to consume food/drinks regularly during a final exam, he/she must register for special accommodations with AccessAbility Services and must submit appropriate documentation from a recognized professional at least three weeks prior to the start of the final examination period.',
' Please remind your students of this regulation in any final exam-related communication.',
' 2. The consumption of food is also prohibited for those administering final examinations however, drinks are permitted (coffee, juice, pop, water etc.).',
''
]
emergency_procedures_link = 'https://uwaterloo.ca/registrar/final-examinations/fire-alarm-evacuation-procedures-final-examinations'
[docs]def ro_depot_instructions (cursor, exam):
depot_locs = cursor.execute_optional_tuple ("select array_agg (depot_loc) filter (where depot_loc <> 'PAC' and depot_loc <> 'CIF') as depot_room_locs, array_agg (depot_loc) filter (where depot_loc = 'PAC' or depot_loc = 'CIF') as gym_locs from (select distinct ro_print_depot (room_id) as depot_code from exam_exam_sitting_room where exam_id = %(exam_id)s) t natural join ro_depot", exam_id=exam.exam_id)
locs_text = []
if depot_locs.depot_room_locs is not None:
locs_text.append ('in %s' % english_join (*depot_locs.depot_room_locs))
if depot_locs.gym_locs is not None:
locs_text.append ('at the front of %s' % english_join (*depot_locs.gym_locs))
result = [
'Instructors must pick up their examination papers/supplies %s on the day of the examination at least 30 minutes prior to the start of the examination. Registrar’s Office delegates will be on duty to dispense the examination papers and supplies 45 minutes to 1 hour before the examination period begins and will remain there until 10 minutes after the examination begins to dispense extra supplies if necessary. %s' % (english_join (*locs_text) if locs_text else '', '%s will be closed and locked; please knock on the door.' % english_join (*depot_locs.depot_room_locs) if depot_locs.depot_room_locs is not None else ''),
'',
'In order to ensure the secure release of examinations, all instructors/proctors who pick up printed examinations will be required to identify themselves with photo identification (e.g. Watcard, driver’s license, etc.) to the Registrar’s Office staff. Exam papers will not be released to instructors/proctors who do not appear on the proctor list or do not have photo identification. Please inform those who will pick up your printed examination to have photo identification with them.'
]
return result
[docs]def render_sittings_rooms_depot (cursor, exam, by_depot):
sittings_sql = "select start_time, exam_time_format_times (start_time, case when sitting_duration is null then exam_duration else sitting_duration end) as times_formatted, ro_print_format_rooms (exam_id, array_agg (room_id)) as rooms %s from exam_exam natural join exam_exam_sitting_room natural join exam_sitting where exam_id = %%(exam_id)s and sitting_admin_id <> 20060 group by exam_id, start_time, times_formatted" % (", (SELECT depot_loc FROM ro_depot WHERE depot_code = ro_print_depot (room_id)) as depot_loc" if by_depot else "")
if by_depot:
sittings_sql = sittings_sql + ", depot_loc"
result = cursor.execute_tuples (sittings_sql + " order by start_time", exam_id=exam.exam_id)
return result
[docs]def ro_pickup_instructions (cursor, exam):
sittings_by_depot = render_sittings_rooms_depot (cursor, exam, by_depot=True)
sittings_by_time = render_sittings_rooms_depot (cursor, exam, by_depot=False)
content_lines = [
'As indicated on the Final Examination Schedule, the examination for %s (%s) will be written on' % (exam.full_title, exam.admin_sections),
]
for sitting in sittings_by_time:
content_lines.append (' %s in %s' % (sitting.times_formatted, sitting.rooms))
content_lines.append ('')
for sitting in sittings_by_depot:
if sitting.depot_loc is not None:
content_lines.append ('Assessments written in %s can be picked up in %s.' % (sitting.rooms, sitting.depot_loc))
content_lines.extend (ro_depot_instructions (cursor, exam))
content_lines.extend (contact_info_text)
content_lines.extend (important_note_text)
content_lines.extend ([
'Please print the fire evacuation procedure related to your exam location in the link below.'
'',
emergency_procedures_link
])
return content_lines
[docs]def ro_admin_ineligible_email (cursor, exam):
content_lines = [
'If you have not delivered or emailed your exam directly to the Registrar’s Office for printing, you are responsible to print and deliver your papers to your exam.',
'',
'If you have delivered or emailed your exam directly to the Registrar’s Office for printing, follow the directions below for exam pick up and supplies:',
]
content_lines.extend (ro_pickup_instructions (cursor, exam))
content = '\n'.join (content_lines)
message = mailout (cursor, ro_email, 'RO Administered Deadline and Instructions for %s on %s' % (exam.full_title, exam.times_formatted), content)
message_to_course_staff (cursor, message, exam.term_id, exam.admin_id, ['ISC', 'INST'])
message.add_recipient ('Cc', odyssey_email)
message.add_recipient ('Cc', ro_email)
message.done ()
[docs]def ro_admin_accepted_email (cursor, exam):
content_lines = []
content_lines.extend (ro_pickup_instructions (cursor, exam))
content = '\n'.join (content_lines)
message = mailout (cursor, ro_email, 'RO Administered Assessment Information for %s on %s' % (exam.full_title, exam.times_formatted), content)
message_to_course_staff (cursor, message, exam.term_id, exam.admin_id, ['ISC', 'INST'])
message.add_recipient ('Cc', odyssey_email)
message.add_recipient ('Cc', ro_email)
message.done ()
[docs]def submit_period_print_jobs (cursor, period):
"""Submit print jobs for all appropriate examinations in a print period.
:param cursor: DB cursor.
:param period: Administered examination deadline (exam_print_deadline row).
Creates print jobs for all eligible examinations in the specified print
period that are ready to print. Also prints an indication on stdout of
what it did for each examination, and uses Outgoing to email the
Registrar's Office a list of all examinations accepted.
"""
exams = cursor.execute_values ("select exam_id from exam_print_deadline_find_exams (%(term_id)s, %(admin_id)s, %(period_id)s) order by primary_start_time", term_id=period.term_id, admin_id=period.administer_admin_id, period_id=period.period_id)
accepted_exams = []
for exam_id in exams:
exam = cursor.execute_required_tuple ("select * from exam_exam_plus where exam_id = %(exam_id)s", exam_id=exam_id)
if exam.master_approved is None:
status = 'Ineligible'
ro_admin_ineligible_email (cursor, exam)
cursor.execute_none ("update exam_exam set administer_admin_id = null where exam_id = %(exam_id)s", exam_id=exam.exam_id)
else:
status = 'Accepted'
ro_admin_accepted_email (cursor, exam)
accepted_exams.append (exam)
print('%s %s' % (status, exam.full_title))
begin_date = period.period_begin_date.strftime ('%B %d')
content = ['For examinations during the period beginning %s, the following courses have submitted electronic PDF masters for printing:' % begin_date, '']
content += ['%s: %s' % (exam.full_title, exam.admin_sections) for exam in accepted_exams]
message = mailout (cursor, odyssey_email, 'Final Assessment Print Period Starting %s' % begin_date, '\n'.join (content))
message.add_recipient ('To', odyssey_ro_email)
message.add_recipient ('Cc', odyssey_email)
message.done ()
cursor.callproc_none ("exam_print_deadline_processed", period.term_id, period.administer_admin_id, period.period_id)
cursor.connection.commit ()
# Produce print jobs
for exam in accepted_exams:
if exam.master_accepted is not None:
status = 'Already printed'
elif exam.sequence_assigned is None:
status = 'Seating not ready'
else:
status = 'Printing'
print_examination (cursor, exam)
print('%s %s' % (status, exam.full_title))