Source code for webapp.models

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

This is a collection of all required models.

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

__author__ = "Pascal Held"
__email__ = ""

import json

from django.db import models
from django.contrib.auth.models import User
from django.core.urlresolvers import reverse

from datetime import datetime

from six.moves.urllib.parse import quote_plus

from import timestamp

[docs]class Worker(models.Model): """A worker is the backend host, which does the calculations""" hostname = models.CharField(max_length=128) """Hostname of the worker""" last_heartbeat = models.DateTimeField(null=True) """Is used to check if worker is up""" last_job = models.DateTimeField(null=True) """Last Job Timestamp, will be used to select worker""" jobs_done = models.IntegerField(default=0) """Number of finished jobs."""
[docs]class Group(models.Model): """The Group class represents a group of people. There could be multiple groups. A group could be a lecture, an event or any other user group. """ name = models.CharField(max_length=128, unique=True) """The name of the group. Must be unique""" free_to_enter = models.BooleanField(default=True) """Is it possible for new users to enter this group?""" users = models.ManyToManyField(User) """Set of all users, which are actually in this group""" def __str__(self): return
[docs] def url(self): """Generate the url to the group page. """ return "/{}/".format(quote_plus(
[docs] def join_url(self): """Generate the url to join the group. """ return reverse("join_group", args=(quote_plus(,))
def get_round_ids(self): return list(self.round_set.all().values_list('id', flat=True)) def get_jsonable(self, user=None): return { "id":, "pk":, "name":, "free_to_enter": self.free_to_enter, }
[docs]class Dataset(models.Model): """Represents a dataset object. It contains of multiple DatasetPoints. """ name = models.CharField(max_length=128, unique=True) """Name of the Dataset""" dimensions = models.IntegerField(null=True) """Number of dimensions, where -1 means variable.""" nr_training_points = models.IntegerField(null=True) """Number of training points.""" nr_test_points = models.IntegerField(null=True) """Number of test points.""" nr_validation_points = models.IntegerField(null=True) """Number of validation points.""" nr_classes = models.IntegerField(null=True) """Number of different classes""" class_labels = models.TextField(null=True, default="{}") """class_id to label map in JSON format.""" default_visualization = models.CharField(max_length=128, default="scatter") """Default visualization type.""" default_dimensions = models.TextField(null=True, default="[]") """Default scatter dimensions.""" def get_dimension_str(self): if self.dimensions is None: return "unknown" if self.dimensions == -1: return "variable" else: return "{}D".format(self.dimensions) def get_size_str(self): return "{}/{}/{}".format(self.nr_training_points, self.nr_test_points, self.nr_validation_points) def __str__(self): return "{} - {} ({})".format(, self.get_dimension_str(), self.get_size_str()) def get_jsonable(self, user=None): return { "name":, "id":, "pk":, "dimensions": self.dimensions, "nr_training_points": self.nr_training_points, "nr_test_points": self.nr_test_points, "nr_validation_points": self.nr_validation_points, "nr_classes": self.nr_classes, "class_labels": json.loads(self.class_labels), "default_visualization": self.default_visualization, "default_dimensions": json.loads(self.default_dimensions) if self.default_dimensions else [], "get_dimension_str": self.get_dimension_str(), "get_size_str": self.get_size_str(), "str": self.__str__(), "url": self.get_json_url(user), "url_training_arff": self.get_arff_url("training"), "url_test_arff": self.get_arff_url("test"), "url_validation_arff": self.get_arff_url("validation"), } def get_json_url(self, user=None): if user and user.is_superuser: return reverse("full_json_dataset", args=(quote_plus(,)) return reverse("json_dataset", args=(quote_plus(,)) def get_arff_url(self, subset): return reverse("get_dataset_arff", args=(quote_plus(, subset))
[docs]class DatasetPoint(models.Model): """A single Point of a dataset.""" dataset = models.ForeignKey(Dataset) """The related dataset.""" coords = models.TextField() """The coordinates as JSON array""" class_id = models.IntegerField() """The class as integer.""" point_type = models.SmallIntegerField() """This pont is part of which part of the dataset (training, test, validation)?""" def get_jsonable(self): return { "pk":, "coords": json.loads(self.coords), "class_id": self.class_id, "point_type": self.point_type } def get_coords_seperated(self, sep=","): s = sep.join([str(x) for x in json.loads(self.coords)]) print(s) return s def __str__(self): return "{}: {} {} ({})".format(, self.coords, self.class_id, self.point_type)
[docs]class Round(models.Model): """A Round is a single event for a group. Each group can define multiple rounds. """ group = models.ForeignKey(Group) """Group assigned to this round""" name = models.CharField(max_length=128) """Name representation of this round""" start_time = models.DateTimeField(null=True) """The start of the round. If the start time is in future or not set, this round is not active""" end_time = models.DateTimeField(null=True) """The end of the round. If the end is in the past, this round is closed. If the end is not set, the round will be open.""" datasets = models.ManyToManyField(Dataset) """All active datasets for this round.""" algorithms = models.TextField() """Selected algorithms as JSON list.""" limit_jobs = models.IntegerField(default=15) """Number of allowed jobs.""" limit_submit = models.IntegerField(default=3) """Number of submissions.""" def __str__(self): return "{}: {}".format(, def get_jsonable(self): return { "id":, "group":, "name":, "start_time": timestamp(self.start_time), "end_time": timestamp(self.end_time), "datasets": [ for ds in self.datasets.all()], "algorithms": json.loads(self.algorithms), "url": self.url() }
[docs] def url(self): """Generate the url to the group page. """ return "{}{}/".format(,
[docs] def is_active(self): """Checks if the current round is active. """ t = datetime.utcnow() if not self.start_time or self.start_time > t: return False if not self.end_time or self.end_time > t: return True return False
[docs] def get_free_jobs(self, user, dataset_id): """Return the number of open jobs. Admin user always get the full number of jobs. Failed jobs are ignored. Parameters ========== user : user object The requesting user dataset_id : int The PK of the selected dataset. """ if user.is_superuser: return self.limit_jobs return max(0, self.limit_jobs-Job.objects .filter(,, dataset_id=dataset_id) .exclude(success=False) .count())
[docs] def get_free_submits(self, user, dataset_id): """Return the number of open submits. Admin user always get the full number of submits. Parameters ========== user : user object The requesting user dataset_id : int The PK of the selected dataset. """ if user.is_superuser: return self.limit_submit return max(0, self.limit_submit-Job.objects .filter(,, dataset_id=dataset_id, is_selected=True).count())
[docs]class Job(models.Model): """Jobs are the units create by user to work in a specific round.""" created = models.DateTimeField(auto_now_add=True) """Create time of the job.""" modified = models.DateTimeField(auto_now=True) """last modified""" user = models.ForeignKey(User) """Owner of the Job""" round = models.ForeignKey(Round) """Related Round""" dataset = models.ForeignKey(Dataset) """Related Dataset""" algorithm = models.CharField(max_length=128) """selected algorithm""" params = models.TextField(default="{}") """Algorithms parameters as JSON dict""" score_test = models.FloatField(null=True) """The score of the test set. Should be in the interval [0,1]""" score_validation = models.FloatField(null=True) """The score of the validation set. Should be in the interval [0,1]""" labels_training = models.TextField(null=True) """A list of lables for the training set in JSON notation""" labels_test = models.TextField(null=True) """A list of lables for the test set in JSON notation""" labels_validation = models.TextField(null=True) """A list of labels for the validation set in JSON notation""" extra_scores_test = models.TextField(null=True) """A dictionary with further scores (just for information) from the test set in JSON notation""" extra_scores_validation = models.TextField(null=True) """A dictionary with further scores (just for information) from the validation set in JSON notation.""" message = models.TextField(null=True) """A message from the training set. It could contain more information about the training/test.""" success = models.NullBooleanField(null=True) """True if the learning process was successful""" worker = models.ForeignKey(Worker, null=True) """Worker which handles this job.""" fetch_params_ts = models.DateTimeField(null=True) """Timestamp when a worker fetches the params.""" finished_ts = models.DateTimeField(null=True) """Timestamp when the calculation are finished.""" is_selected = models.BooleanField(default=False) """Is this job marked as a selected result?""" def get_jsonable(self, user): # public values data = { "algorithm": self.algorithm, "dataset": self.dataset_id, "round": self.round_id, "id":, "pk":, "user": self.user.username, "finished": self.finished_ts is not None } # owners values if user.is_superuser or user == self.user: extra_data = { "extra_scores_test": self.extra_scores_test, "is_selected": self.is_selected, "labels_training": self.labels_training, "labels_test": self.labels_test, "message": self.message, "params": self.params, "score_test": round(self.score_test,3) if self.score_test else self.score_test, "success": self.success, "images": ["/media/{}/{}".format(, img.filename) for img in self.jobpicture_set.all()] } else: extra_data = {} for key, val in extra_data.items(): data[key] = val # public selected values if self.is_selected: data["score_validation"] = round(self.score_validation,3) if self.score_validation else self.score_validation # owner selected values if (user.is_superuser or user == self.user) and self.is_selected: data["extra_scores_validation"] = self.extra_scores_validation, data["labels_validation"] = self.labels_validation return data def __str__(self): return "Job #{} ({}, {}, {}, {})".format(, self.user.username, self.round_id,, self.algorithm)
[docs]class JobPicture(models.Model): """Pictures to illustrate job details""" job = models.ForeignKey(Job) """Related Job""" filename = models.CharField(max_length=128) """Filename of the image"""
[docs]class DatasetResultPoints(models.Model): """Best result with points for a given dataset in a specific round.""" user = models.ForeignKey(User) """User of this result""" round = models.ForeignKey(Round) """The round""" dataset = models.ForeignKey(Dataset) """The selected Dataset""" job = models.ForeignKey(Job) """The related job""" score = models.FloatField() """The Validation Score""" points = models.IntegerField() """The earned Points""" def __str__(self): return "{} # {} # {} # {} # {} # {}".format(self.round, self.dataset, self.user, self.job, self.score, self.points) def get_jsonable(self): return { "user": self.user.username, "realname": self.user.first_name, "roundid": self.round_id, "jobid": self.job_id, "datasetid": self.dataset_id, "score": round(self.score,3), "points": self.points }
[docs]class RoundResultPoints(models.Model): """Round results.""" user = models.ForeignKey(User) """related user""" round = models.ForeignKey(Round) """related round""" points = models.IntegerField() """Sum of points of all datasets.""" stars = models.IntegerField() """Stars / Big Points for this round""" def get_jsonable(self): return { "user": self.user.username, "realname": self.user.first_name, "roundid": self.round_id, "stars": self.stars, "points": self.points }