play-daemon-threaded/pipeline/daisy.py

174 lines
6.3 KiB
Python

import logging
import requests
from collections import namedtuple
from datetime import datetime, timedelta
eduTypes = {1: {'en': 'Teaching session',
'sv': 'Lektion'},
2: {'en': 'Seminar',
'sv': 'Seminarium'},
3: {'en': 'Lecture',
'sv': 'Föreläsning'},
4: {'en': 'Laboratory session',
'sv': 'Laboration'},
5: {'en': 'Presentation',
'sv': 'Redovisning'},
6: {'en': 'Introduction',
'sv': 'Introduktion'},
7: {'en': 'Exercise',
'sv': 'Övning'},
8: {'en': 'Supervision',
'sv': 'Handledning'},
9: {'en': 'Tutorial',
'sv': 'Räknestuga'},
10: {'en': 'Project',
'sv': 'Projekt'},
11: {'en': 'Workshop',
'sv': 'Workshop'},
12: {'en': 'Q & A',
'sv': 'Frågestund'},
13: {'en': 'Deadline',
'sv': 'Deadline'},
None: None,
}
def parseSemester(semester):
halves = {'1': 'vt',
'2': 'ht'}
semester = str(semester)
year = semester[:4]
half = semester[4]
return f"{halves[half]}{year}"
class Daisy:
def __init__(self, config):
self.auth = (config['user'],
config['password'])
self.headers = {'Accept': 'application/json'}
self.url = config['url'].rstrip('/')
self.logger = logging.getLogger('play-daemon')
def _get(self, path, params={}):
"""
Make a request to the daisy API and return the result as a dict.
Throws an exception if the HTTP response indicates an error.
"""
r = requests.get(f'{self.url}{path}',
params,
headers=self.headers,
auth=self.auth)
r.raise_for_status()
return r.json()
def get_booking(self, start_time, end_time, room):
"""
Return the booking with the best overlap with the given times
for the given room.
Returned bookings are fully resolved with respect to users,
courses, etc.
"""
fmt = '%Y-%m-%d'
start_day = start_time.strftime(fmt)
end_day = (end_time + timedelta(days=1)).strftime(fmt)
params = {'start': start_day,
'end': end_day,
'room': room}
bookings = self._get('/schedule', params=params)
# Resolve all the numeric IDs in a booking:
for booking in bookings:
courses = []
for course in booking['courseSegmentInstances']:
courses.append(self.get_course(course['id']))
booking['courseSegmentInstances'] = sorted(courses)
teachers = []
for teacher in booking['teachers']:
teachers.append(self.get_person(teacher['id']))
booking['teachers'] = sorted(teachers)
if booking['bookedBy']:
booking['bookedBy'] = self.get_person(booking['bookedBy'])
# This should be replaced by an API lookup
booking['educationalType'] = eduTypes[booking['educationalType']]
# If a lecture belongs to more than one course, there will be one
# booking for each course, all with identical times.
# Sorting on courseSegmentInstance ensures that we return a consistent
# course in such cases.
def booking_sort(booking):
if booking['courseSegmentInstances']:
return booking['courseSegmentInstances'][0]['designation']
return ''
Fit = namedtuple('Fit', ['booking', 'overlap'])
best = Fit(None, timedelta())
self.logger.info('Matching bookings for room %s at %s - %s',
room, start_time, end_time)
for booking in sorted(bookings, key=booking_sort):
b_start = datetime.fromisoformat(booking['start'])
b_end = datetime.fromisoformat(booking['end'])
self.logger.debug('Booking %s for course %s: %s - %s',
booking['id'],
booking['courseSegmentInstances'],
b_start,
b_end)
range_start = max(start_time, b_start)
range_end = min(end_time, b_end)
overlap = range_end - range_start
self.logger.debug('Time ranges overlap by %s', overlap)
self.logger.debug('Previous best overlap: %s', best.overlap)
if overlap > best.overlap:
self.logger.debug('Saving %s as current best fit',
booking['id'])
best = Fit(booking, overlap)
bestfit = best.booking
if bestfit:
self.logger.info(
'Best booking fit was %s for course %s, overlap %s',
bestfit['id'],
bestfit['courseSegmentInstances'],
best.overlap)
else:
self.logger.info('No suitable booking found.')
return best.booking
def get_person(self, person_id):
"""
Return the SU.SE username associated with a user. If none exists,
return the user's first and last names instead.
Throws an exception if a user has more than one SU.SE username.
"""
usernames = [name['username']
for name in self._get(f'/person/{person_id}/usernames')
if name['realm'] == 'SU.SE']
if len(usernames) == 0:
person = self._get(f'/person/{person_id}')
return f'{person["firstName"]} {person["lastName"]}'
if len(usernames) > 1:
raise Exception(
f'More than one SU.SE username for {person_id}: {usernames}')
return usernames[0]
def get_course(self, course_id):
"""
Return the course designation for the given course ID.
"""
course = self._get(f'/courseSegment/{course_id}')
return {'designation': course['designation'],
'semester': parseSemester(course['semester'])}
def get_room_name(self, room_id):
"""
Return the name of the room with the given ID.
"""
location = self._get(f'/location/{room_id}')
return location['designation']