Source code for webapp.views

#!/usr/bin/env python
# -*- coding: utf-8 -*-

This is a collection of all required views for the web frontend.

from __future__ import print_function, division, unicode_literals, absolute_import, generators

__author__ = "Pascal Held"
__email__ = ""

from datetime import datetime, timedelta
from time import timezone

from django.shortcuts import render, get_object_or_404
from django.contrib.auth.decorators import login_required, user_passes_test
from django.contrib.auth import login, authenticate
from django.http.response import HttpResponseRedirect, HttpResponseBadRequest, HttpResponseForbidden, HttpResponse
from django.middleware.csrf import get_token
from django.views.decorators.cache import cache_control
import json
import base64
import os
import string

from six.moves.urllib.parse import quote, unquote_plus

from webapp.models import *
from webapp.processors import general_processor
import settings
import webapp.analyzer as analyzer
# Create your views here.

JSON_MIME = "application/json"

def HttpResponseJSON(content, *args, **kwargs):
    return HttpResponse(json.dumps(content), *args, content_type=JSON_MIME, **kwargs)

# Helpers

class NameMapper(object):
    id2name = None
    name2id = None

    def __init__(self):
        self.id2name = dict()
        self.name2id = dict()

    def get_id(self, name):
        if name in self.name2id:
            return self.name2id[name]

        new_id = len(self.id2name)
        self.id2name[new_id] = name
        self.name2id[name] = new_id
        return new_id

    def get_mapping(self):
        return self.id2name

# Views

[docs]def admin_only(user): """Test function to check if the user is an admin, and allowed to manage things. """ if user and user.is_superuser: return True return False
def webapp(request): if request.user.is_authenticated(): return render(request, "webapp.html") else: return render(request, "login.html") @user_passes_test(admin_only)
[docs]def add_group(request): """Creates a new group. The Group parameter are read from the request.REQUEST field. So you can use it with POST and GET. Parameters ========== group_name : str The name of the new group public : str Default is public. Private on the values ('0', 'no', 'private', 'false') """ if not "group_name" in request.REQUEST: return HttpResponseBadRequest group_name = request.REQUEST["group_name"] public = True try: if request.REQUEST["public"] in ('0', 'no', 'private', 'false'): public = False except KeyError: pass if not Group.objects.filter(name=group_name).count(): g=Group(name=group_name, free_to_enter=public) if "api" in request.REQUEST: return HttpResponseJSON({ "status": "ok", "group_id": }) return HttpResponseRedirect(g.url())
@login_required() def get_groups(request): context = general_processor(request) return HttpResponseJSON(context["groups"]) @login_required() def get_round(request, pk): round = get_object_or_404(Round, pk=pk) return HttpResponseJSON(round.get_jsonable()) @user_passes_test(admin_only) def add_dataset(request): if not "name" in request.REQUEST or not "points" in request.REQUEST or not request.REQUEST["name"] or not request.REQUEST["points"]: return HttpResponseBadRequest() points = json.loads(request.REQUEST['points']) ds = Dataset(name=request.REQUEST["name"]) mapper = NameMapper() dimension = None cnt = [0, 0, 0] for point in points: p = DatasetPoint( dataset=ds, coords=json.dumps(point["coords"]), class_id=mapper.get_id(point["class_id"]), point_type=point["point_type"] ) if dimension is None: dimension = len(point["coords"]) elif dimension != len(point["coords"]): dimension = -1 cnt[point["point_type"]] += 1 ds.nr_training_points = cnt[settings.TYPE_TRAINING] ds.nr_test_points = cnt[settings.TYPE_TEST] ds.nr_validation_points = cnt[settings.TYPE_VALIDATION] mapping = mapper.get_mapping() ds.nr_classes = len(mapping) ds.class_labels = json.dumps(mapping) ds.dimensions = dimension if "api" in request.REQUEST: return HttpResponseJSON({ "status": "ok", "dataset": ds.get_jsonable(request.user) }) return HttpResponseRedirect("/datasets/") @user_passes_test(admin_only)
[docs]def edit_round(request): """updates the round parameters.""" required_fields = ["name", "group", "algorithms", "datasets"] if not all([field in request.REQUEST for field in required_fields]): return HttpResponseBadRequest() group = get_object_or_404(Group, name=request.REQUEST["group"]) if "round-pk" in request.REQUEST and int(request.REQUEST["round-pk"]) != -1: round = get_object_or_404(Round, pk=int(request.REQUEST["round-pk"])) else: round = Round() = request.REQUEST["name"] = group round.algorithms = request.REQUEST["algorithms"] try: offset = timedelta(minutes=int(request.REQUEST["timeoffset"])) except: offset = 0 try: start_datestring = "{} {}".format(request.REQUEST["startDate"], request.REQUEST["startTime"]) round.start_time = datetime.strptime(start_datestring, "%Y-%m-%d %H:%M") + offset except: round.start_time = None try: end_datestring = "{} {}".format(request.REQUEST["endDate"], request.REQUEST["endTime"]) round.end_time = datetime.strptime(end_datestring, "%Y-%m-%d %H:%M") + offset except: round.end_time = None round.datasets.clear() for ds_pk in json.loads(request.REQUEST["datasets"]): round.datasets.add(Dataset.objects.get(pk=int(ds_pk))) settings.msg_queue.put({ "msg": "update_round", "roundid":, "data": round.get_jsonable() }) if "api" in request.REQUEST: return HttpResponseJSON({ "status": "ok", "round": round.get_jsonable() }) return HttpResponseRedirect(group.url()+"settings/")
[docs]def join_group(request, group_name): """Add the current user to the selected group. """ group = get_object_or_404(Group, name=group_name) if not group.free_to_enter: return HttpResponseForbidden() group.users.add(request.user) if "api" in request.REQUEST: return HttpResponseJSON({ "status": "ok" }) return HttpResponseRedirect(group.url())
@login_required() def get_datasets(request): return HttpResponseJSON({ ds.get_jsonable(request.user) for ds in Dataset.objects.all()}) @cache_control(max_age=86400, public=True) # cache one day @login_required() def get_dataset_json(request, name="", pk=None): name = unquote_plus(name) if pk: dataset = get_object_or_404(Dataset, pk=pk) else: dataset = get_object_or_404(Dataset, name=name) points = [p.get_jsonable() for p in dataset.datasetpoint_set.exclude(point_type=settings.TYPE_VALIDATION)] return HttpResponseJSON(points) @user_passes_test(admin_only) def get_full_dataset_json(request, name="", pk=None): name = unquote_plus(name) if pk: dataset = get_object_or_404(Dataset, pk=pk) else: dataset = get_object_or_404(Dataset, name=name) points = [p.get_jsonable() for p in dataset.datasetpoint_set.all()] return HttpResponseJSON(points) def get_env(request): return HttpResponseJSON({ "csrftoken": get_token(request) }) @login_required() def get_dataset_arff(request, name="", pk=None, subset=""): name = unquote_plus(name) if pk: dataset = get_object_or_404(Dataset, pk=pk) else: dataset = get_object_or_404(Dataset, name=name) types = { "training": settings.TYPE_TRAINING, "test": settings.TYPE_TEST, "validation": settings.TYPE_VALIDATION } if subset == "validation" and not request.user.is_superuser: return HttpResponseForbidden() context = { "points": dataset.datasetpoint_set.filter(point_type=types.get(subset, -1)), "subset": subset, "dataset": dataset, "attributes": range(dataset.dimensions), "classes_str": ",".join([str(x) for x in json.loads(dataset.class_labels).keys()]) } return render(request, "dataset.arff", context, content_type="text/plain") def api_ok(request): return HttpResponseJSON({ "status": "ok" }) @login_required() def create_job(request): for field in ["round", "algorithm", "dataset", "params"]: if not field in request.POST: return HttpResponseJSON({ "status": "error", "msg": "{} is required in POST data.".format(field) }) try: round = Round.objects.get(pk=request.POST["round"]) except: return HttpResponseJSON({ "status": "error", "msg": "Round not found" }) if not round.is_active(): return HttpResponseJSON({ "status": "error", "msg": "Round not active" }) try: ds = round.datasets.get(pk=request.POST["dataset"]) except: return HttpResponseJSON({ "status": "error", "msg": "Dataset not in round" }) if not round.get_free_jobs(request.user, return HttpResponseJSON({ "status": "error", "msg": "Job limit reached" }) alg = request.POST["algorithm"] if not alg in json.loads(round.algorithms): return HttpResponseJSON({ "status": "error", "msg": "Algorithm not allowed" }) sendparams = json.loads(request.POST["params"]) jobparams = {} def check_params(parameters, prefix=""): print(parameters) for pid, pdata in parameters.items(): value = None fullpid = prefix+pid if fullpid in sendparams: value = sendparams[fullpid] if "parser" in pdata: value = pdata["parser"](value) if "validators" in pdata: for validator in pdata["validators"]: res = validator(value, dataset=ds, **pdata) if res: return "{}: {}".format(pdata["name"], res) jobparams[fullpid] = value if pdata["type"] == "dict" and "values" in pdata and isinstance(pdata["values"], dict): if value in pdata["values"]: info = pdata["values"][value] if "parameters" in info: res = check_params(info["parameters"], prefix+pid+"-") if res: return res return False msg = check_params(settings.algorithms[alg]["parameters"]) if msg: return HttpResponseJSON({ "status": "error", "msg": msg, }) job = Job() job.user = request.user job.round = round job.dataset = ds job.algorithm = alg job.params = json.dumps(jobparams) settings.msg_queue.put({ "msg": "update_job", "jobid":, "user": job.user.username }) return HttpResponseJSON({ "status": "ok", "jobid":, "free_jobs": { "dataset_id":, "round_id":, "total_jobs": round.limit_jobs, "total_submits": round.limit_submit, "free_jobs": round.get_free_jobs(request.user,, "free_submits": round.get_free_submits(request.user, } }) @user_passes_test(admin_only) def get_job_params(request, jobid): worker = get_worker(request) job = get_object_or_404(Job, pk=jobid) job.fetch_params_ts = datetime.utcnow() job.worker = worker worker.last_job = datetime.utcnow() return HttpResponseJSON({ "algorithm": job.algorithm, "dataset_id": job.dataset_id, "dataset":, "params": json.loads(job.params) }) @login_required() def get_job_details(request, jobid): job = get_object_or_404(Job, pk=jobid) return HttpResponseJSON(job.get_jsonable(request.user)) @login_required() def submit_job(request, jobid): job = get_object_or_404(Job, pk=jobid) round=job.round if not round.get_free_submits(request.user, job.dataset_id): return HttpResponseForbidden("Submit limited reached.") if job.user == request.user or request.user.is_superuser: if job.round.is_active(): job.is_selected = True settings.msg_queue.put({ "msg": "update_job", "jobid":, "user": job.user.username }) analyzer.update_round_dataset_ranking(job.round_id, job.dataset_id) return HttpResponseJSON({ "job": job.get_jsonable(request.user), "free_jobs": { "dataset_id": job.dataset_id, "round_id": job.round_id, "total_jobs": round.limit_jobs, "total_submits": round.limit_submit, "free_jobs": round.get_free_jobs(request.user, job.dataset_id), "free_submits": round.get_free_submits(request.user, job.dataset_id) } }) else: return HttpResponseForbidden("Related round is not active") return HttpResponseForbidden("You are not job owner.") @login_required() def load_job_list(request): if not request.user.is_superuser: jobs = [job.get_jsonable(request.user) for job in Job.objects.filter(user=request.user)] else: jobs = [job.get_jsonable(request.user) for job in Job.objects.all()] return HttpResponseJSON(jobs) @user_passes_test(admin_only) def update_job_details(request, jobid): job = get_object_or_404(Job, pk=jobid) try: if request.POST["score_test"] != "None": job.score_test = float(request.POST["score_test"]) else: job.score_test = None except KeyError: pass except ValueError: return HttpResponseJSON({ "status": "error", "msg": "score_test has an invalid format" }) try: if request.POST["score_validation"] != "None": job.score_validation = float(request.POST["score_validation"]) else: job.score_validation = None except KeyError: pass except ValueError: return HttpResponseJSON({ "status": "error", "msg": "score_validation has an invalid format" }) try: if request.POST["labels_training"] != "null" and request.POST["labels_training"] is not None: job.labels_training = json.dumps([int(x) for x in json.loads(request.POST["labels_training"])]) else: job.labels_training = None except KeyError: pass except ValueError: return HttpResponseJSON({ "status": "error", "msg": "labels_training has an invalid format" }) try: if request.POST["labels_test"] != "null" and request.POST["labels_test"] is not None: job.labels_test = json.dumps([int(x) for x in json.loads(request.POST["labels_test"])]) else: job.labels_test = None except KeyError: pass except ValueError: return HttpResponseJSON({ "status": "error", "msg": "labels_test has an invalid format" }) try: if request.POST["labels_validation"] != "null" and request.POST["labels_validation"] is not None: job.labels_validation = json.dumps([int(x) for x in json.loads(request.POST["labels_validation"])]) else: job.labels_validation = None except KeyError: pass except ValueError: return HttpResponseJSON({ "status": "error", "msg": "labels_validation has an invalid format" }) try: if request.POST["extra_scores_test"] != "null": job.extra_scores_test = json.dumps({x:y for x,y in json.loads(request.POST["extra_scores_test"]).items()}) else: job.extra_scores_test = None except KeyError: pass except ValueError: return HttpResponseJSON({ "status": "error", "msg": "extra_scores_test has an invalid format" }) try: if request.POST["extra_scores_validation"] != "null": job.extra_scores_validation = json.dumps({x:y for x,y in json.loads(request.POST["extra_scores_validation"]).items()}) else: job.extra_scores_validation = None except KeyError: pass except ValueError: return HttpResponseJSON({ "status": "error", "msg": "extra_scores_validation has an invalid format" }) try: if request.POST["message"] != "null": job.message = request.POST["message"] else: job.message = None except KeyError: pass try: if request.POST["message"] != "null": success_text = request.POST["success"] job.success = success_text.lower() in ("1", "true", "yes") else: job.success = None except KeyError: pass job.finished_ts = datetime.utcnow() worker = job.worker worker.jobs_done += 1 settings.msg_queue.put({ "msg": "update_job", "jobid":, "user": job.user.username }) return HttpResponseJSON({ "status": "ok", }) @user_passes_test(admin_only) def add_job_picture(request, jobid): job = get_object_or_404(Job, pk=jobid) valid_chars = "-_.() %s%s" % (string.ascii_letters, string.digits) filename = ''.join(c for c in request.POST["filename"] if c in valid_chars) try: path = settings.DJANGO_MEDIA_ROOT+os.path.sep+str(jobid) try: os.makedirs(path) except OSError: pass with open(path+os.path.sep+filename, "wb") as file: file.write(base64.b64decode(request.POST["data"].encode())) JobPicture.objects.create(job=job, filename=filename) except KeyError: return HttpResponseJSON({ "status": "error", "msg": "filename and data required" }) settings.msg_queue.put({ "msg": "update_job", "jobid":, "user": job.user.username }) return HttpResponseJSON({ "status": "ok", }) def create_user(request): for param in ["username", "realname", "password1", "password2"]: if param not in request.POST or not request.POST[param]: return HttpResponseRedirect("/#"+quote("{} is required".format(param))) if request.POST["password1"] != request.POST["password2"]: return HttpResponseRedirect("/#"+quote("The given passwords are not equal.")) if not request.POST["password1"]: return HttpResponseRedirect("/#"+quote("You must set a password")) if User.objects.filter(username=request.POST["username"]).count(): return HttpResponseRedirect("/#"+quote("User already exists. Please choose another name.")) User.objects.create_user(request.POST["username"], password=request.POST["password1"], first_name=request.POST["realname"]) user = authenticate(username=request.POST["username"], password=request.POST["password1"]) login(request, user) return HttpResponseRedirect("/") def get_worker(request): hostname = request.META["REMOTE_HOST"] if "REMOTE_HOST" in request.META and request.META["REMOTE_HOST"] else request.META["REMOTE_ADDR"] return Worker.objects.get_or_create(hostname=hostname)[0] @user_passes_test(admin_only) def ping(request): worker = get_worker(request) job = Job.objects.filter(worker=None).first() if not job: return HttpResponseJSON(None) worker.last_heartbeat = datetime.utcnow() job.worker = worker return HttpResponseJSON( @login_required() def score_list(request, round_id): scores = DatasetResultPoints.objects.filter(round_id=round_id) objs = [obj.get_jsonable() for obj in scores] scores = RoundResultPoints.objects.filter(round_id=round_id) for obj in scores: objs.append(obj.get_jsonable()) return HttpResponseJSON(objs) @login_required() def score_list_total(request): scores = RoundResultPoints.objects.all() objs = [obj.get_jsonable() for obj in scores] return HttpResponseJSON(objs) def get_time(request): time = datetime.utcnow() return HttpResponseJSON(timestamp(time)) @user_passes_test(admin_only) def start_round(request, round_id): round = get_object_or_404(Round, pk=round_id) t = datetime.utcnow() if not round.is_active(): round.start_time = t if round.end_time < t: round.end_time = None settings.msg_queue.put({ "msg": "update_round", "roundid":, "data": round.get_jsonable() }) return HttpResponseJSON({ "status": "ok", "round": round.get_jsonable() }) return HttpResponseJSON({ "status": "ok", }) @user_passes_test(admin_only) def stop_round(request, round_id): round = get_object_or_404(Round, pk=round_id) t = datetime.utcnow() if round.end_time is None or round.end_time > t: round.end_time = t settings.msg_queue.put({ "msg": "update_round", "roundid":, "data": round.get_jsonable() }) return HttpResponseJSON({ "status": "ok", "round": round.get_jsonable() }) return HttpResponseJSON({ "status": "ok", }) @login_required() def get_free_jobs(request, round_id, dataset_id): round = get_object_or_404(Round, pk=round_id) return HttpResponseJSON({ "dataset_id": dataset_id, "round_id": round_id, "total_jobs": round.limit_jobs, "total_submits": round.limit_submit, "free_jobs": round.get_free_jobs(request.user, dataset_id), "free_submits": round.get_free_submits(request.user, dataset_id) }) @user_passes_test(admin_only) def update_dataset_default_dimension(request): ds = get_object_or_404(Dataset, pk=request.POST["dataset_id"]) data = [int(x) for x in json.loads(request.POST["dimensions"])] ds.default_dimensions = json.dumps(data) return HttpResponseJSON({ "status": "ok" })