Homec4science
Diffusion Slides (master) Browse

R3683/scripts/validate_groups
c7e05dd9fa90master

Last Change
Raw File

validate_groups
View Options

#!/usr/bin/env python3
import os
import subprocess
import mailer
import argparse
import getpass
import random
import Slides.class_helper as ch
import pandas as pd
config = ch.get_class_config()
parser = argparse.ArgumentParser(
description='Validate groups for homeworks')
parser.add_argument('-n', '--homework', type=str,
help='specify the homework label',
required=True)
parser.add_argument('-t', '--target', type=str,
help='Specify the name of a mail box'
' to send test messages, '
f'default : {config["teachers"][0]}',
default=config['teachers'][0])
parser.add_argument('-r', '--report', action='store_true',
help='Generate a org-mode file ready to grade')
parser.add_argument('-u', '--username', type=str,
help='Username to log to the SMTP server',
default=None)
args = parser.parse_args()
students_list = config['students']
students_list.rename(columns={"e-Mail": "email"},
inplace=True)
students_list = students_list.applymap(
lambda x: x.strip() if isinstance(x, str) else x)
students_list["in_group"] = False
students_list.set_index("email", inplace=True)
homework = config['homeworks'][args.homework]
group_list = homework['groups']
group_list.rename(columns={"Student #1 EPFL email": "email1",
"Student #2 EPFL email": "email2",
"Homeworks repository": "repository"
},
inplace=True)
group_list = group_list.applymap(
lambda x: x.strip() if isinstance(x, str) else x)
target = args.target
username = args.username
password = None
if username:
print('login:', username)
password = getpass.getpass()
# ###############################################################
# generate group keys
def make_group_keys(group):
emails = [group.email1, group.email2]
emails.sort()
if len(emails) < 2:
raise RuntimeError(
'invalid group: ' + str(emails))
print(emails)
key = '_'.join(emails)
return key
group_keys = group_list.apply(make_group_keys, axis=1)
group_list['group_key'] = group_keys
group_list.set_index('group_key', inplace=True)
# ###############################################################
def run_command(cmd, cwd=None):
if cwd is None:
cwd = os.getcwd()
process = subprocess.Popen(cmd, shell=True,
cwd=cwd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
output = process.communicate()
output = [i.decode().strip() for i in output if i.decode().strip() != '']
output = '\n'.join(output)
ret = process.returncode
return ret, output
# ###############################################################
# search for failing to clone groups
try:
os.mkdir(os.path.join(config['git_root'], 'homeworks'))
except Exception:
pass
def clone_group_repo(group):
dirname = os.path.join(
config['git_root'], 'homeworks', args.homework+".repositories", group.name)
if os.path.isdir(dirname):
ret = 0
else:
clone_cmd = group.repository + ' ' + dirname
ret, out = run_command(clone_cmd)
if ret == 0:
print('Cloning_Success: group ' + group.name)
else:
print('Cloning_Failure: group ' + group.name)
cmd = 'git rev-parse --abbrev-ref origin/HEAD'
ret, out = run_command(cmd, cwd=dirname)
remote, master = out.split('/')
cmd = ('git rev-list -n 1 --first-parent'
f' --before="{homework["deadline"]}" {master}')
ret, rev = run_command(cmd, cwd=dirname)
cmd = f'git checkout {rev}'
ret, out = run_command(cmd, cwd=dirname)
if ret != 0:
print(f"Failed to get revision: \n\n{out}")
return ret == 0, master, rev
res = group_list.apply(
clone_group_repo, axis=1, result_type='expand')
group_list['clone_success'] = res[0]
group_list['master'] = res[1]
group_list['revision'] = res[2]
# ###############################################################
# construct invalid groups
def construct_valid_groups(group):
# check if in student list
count1 = students_list.loc[group.email1].size
count2 = students_list.loc[group.email2].size
is_valid = (count1 > 0 and count2 > 0)
if is_valid:
students_list.loc[group.email1, 'in_group'] = True
students_list.loc[group.email2, 'in_group'] = True
return is_valid
valid_group = group_list.apply(construct_valid_groups, axis=1)
group_list['valid_group'] = valid_group
################################################################
# generate org report file
def append_group_in_report(f, group):
f.write(f"* {group.name}\n")
f.write(f"** {group.revision}\n\n")
if 'report_template' in homework:
f.write(homework['report_template'])
elif 'report_template' in config:
f.write(config['report_template'])
f.write("\n")
if args.report:
fname = os.path.join(config['git_root'],
'homeworks', args.homework+'.report.org')
if os.path.exists(fname):
print(
f"Warning: will not overwrite the report file: delete it manually ({fname})")
else:
f = open(fname, 'w')
group_list.apply(lambda g: append_group_in_report(f, g), axis=1)
f.close()
################################################################
# make random groups
unregistered_list = [
e.name for k, e in students_list.query('in_group==False').iterrows()]
random.shuffle(unregistered_list)
if len(unregistered_list):
if len(unregistered_list) % 2 == 0:
random_groups = [e for e in zip(
unregistered_list[::2], unregistered_list[1::2])]
else:
random_groups = [e for e in zip(
unregistered_list[:-1:2], unregistered_list[1:-1:2])]
random_groups[-1] = random_groups[-1][0], random_groups[-1][1], unregistered_list[-1]
random_groups = [" - {0}".format(", ".join(b))
for b in random_groups]
random_groups = '\n'.join(random_groups)
################################################################
# output info
print('\ntotal_students:', len(students_list))
print('registered_students:', len(students_list.query("in_group==True")))
print('unregistered_students:', len(students_list.query("in_group==False")))
print('invalid_groups:', len(group_list.query("valid_group==False")))
################################################################
# sending emails
# send emails o failed clones
def send_failed_clone_email(group):
mailer.mail(
username=username,
password=password,
sender_email=target,
subject='SP4E homeworks: error in cloning your project',
copy_mail=config['teachers']+config['assistants'],
message=f"""
Dear Students,
Apparently we cannot clone your repository:
{group.repository}
Please fix the permissions.
Best regards,
The teaching team.
""",
target_emails=[group.email1, group.email2]
)
def send_invalid_groups(group):
mailer.mail(
username=username,
password=password,
sender_email=target,
subject=f'{config["acronym"]} homeworks: invalid group',
message="""
Dear Students,
Your group is composed with at least a student not officially registered
for the class. Therefore I have to ask you to change the composition of
the group.
You have to understand that the grading of so many projects is a lot of work.
Therefore we will do it only for the registered students.
With my best regards,
The teaching team.
""",
copy_mail=config['teachers']+config['assistants'],
target_emails=[group.email1, group.email2]
)
# send email to unregistered people
def send_unregistered():
if 'group_form' in homework:
group_form = homework["group_form"]
else:
group_form = config["group_form"]
mailer.mail(
username=username,
password=password,
sender_email=target,
subject=f'{config["acronym"]} homeworks: not registered in a group',
message=f"""
Dear Students,
Apparently you did not register yet to any group.
Several reasons might explain this.
If it happens that you need to find a pair, please find below the
automatically created groups.
{random_groups}
If the situation does not suit you, please inform us as quickly as
possible.
You still have to create a repository to store your homework.
Please inform us of your repository URI, for instance by filling the
[form]({group_form})
With my best regards,
The teaching team.
""",
copy_mail=config['teachers']+config['assistants'],
target_emails=unregistered_list,
markdown=True
)
if username:
group_list.query("clone_success==False").apply(
send_failed_clone_email, axis=1)
group_list.query("valid_group==False").apply(
send_invalid_groups, axis=1)
if len(unregistered_list) > 0:
send_unregistered()
c4science · Help