126 lines
4.7 KiB
Python
126 lines
4.7 KiB
Python
import logging
|
|
import requests
|
|
from collections import namedtuple
|
|
from datetime import datetime, timedelta
|
|
|
|
class DaisyHandler:
|
|
def __init__(self, config):
|
|
self.auth = (config['daisy']['user'],
|
|
config['daisy']['password'])
|
|
self.headers = {'Accept': 'application/json'}
|
|
self.url = config['daisy']['url'].rstrip('/')
|
|
self.logger = logging.getLogger('play-daemon')
|
|
|
|
'''
|
|
Make a request to the daisy API and return the result as a dict.
|
|
|
|
Throws an exception if the HTTP response indicates an error.
|
|
'''
|
|
def _get(self, path, params={}):
|
|
r = requests.get(f'{self.url}{path}',
|
|
params,
|
|
headers=self.headers,
|
|
auth=self.auth)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
'''
|
|
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.
|
|
'''
|
|
def get_booking(self, start_time, end_time, room):
|
|
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'])
|
|
|
|
def booking_sort(booking):
|
|
if booking['courseSegmentInstances']:
|
|
return booking['courseSegmentInstances'][0]
|
|
return ''
|
|
|
|
Fit = namedtuple('Fit', ['booking', 'overlap'])
|
|
best = Fit(None, timedelta())
|
|
|
|
self.logger.debug('Matching bookings for room %s at %s - %s',
|
|
room, start_time, end_time)
|
|
for booking in sorted(bookings, key=booking_sort):
|
|
#b_start = datetime.fromtimestamp(int(booking['start'] / 1000))
|
|
#b_end = datetime.fromtimestamp(int(booking['end'] / 1000))
|
|
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.debug(
|
|
'Best booking fit was %s for course %s, overlap %s',
|
|
bestfit['id'],
|
|
bestfit['courseSegmentInstances'],
|
|
best.overlap)
|
|
else:
|
|
self.logger.debug('No suitable booking found.')
|
|
return best.booking
|
|
|
|
'''
|
|
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.
|
|
'''
|
|
def get_person(self, person_id):
|
|
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]
|
|
|
|
'''
|
|
Return the course designation for the given course ID.
|
|
'''
|
|
def get_course(self, course_id):
|
|
course = self._get(f'/courseSegment/{course_id}')
|
|
return course['designation']
|
|
|
|
'''
|
|
Return the name of the room with the given ID.
|
|
'''
|
|
def get_room_name(self, room_id):
|
|
location = self._get(f'/location/{room_id}')
|
|
return location['designation']
|