Source code for api

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

#TODO: add content descriptions

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

__author__ = "Pascal Held"
__email__ = ""

from six.moves.http_cookiejar import CookieJar
from six.moves.urllib.request import build_opener, HTTPCookieProcessor
from six.moves.urllib.parse import urlencode, quote_plus
from six.moves.urllib.error import HTTPError
import base64
import json

import settings

[docs]def normalize_url(host, path): """Adds the path to the url, to get an full uri. Multiple "/" between host and path will be normalized to host/path Parameters ========== host : string Hostname, maybe with port extension and protocoll path : string Path of the uri """ while host[-1] == "/": host = host[:-1] while path[0] == "/": path = path[1:] return "{}/{}".format(host, path)
[docs]class API(object): _csrftoken = None dataset_cache = None def __init__(self, username=settings.BACKEND_USERNAME, password=settings.BACKEND_PASSWORD, host=settings.SERVER_URI): """Initializes the API. This includes a login process. Parameters ========== username : string (default BACKEND_USERNAME from config) Selected username password : string (default BACKEND_PASSWORD from config) Password host : string (default SERVER_URI from config) The host uri of the Server Notes ===== Most request requires admin privileges. """ = host cj = CookieJar() self.cj = cj self.opener = build_opener(HTTPCookieProcessor(cj)) env = self.load("env.json/") if "csrftoken" in env: self._csrftoken = env["csrftoken"] self.login(username, password) env = self.load("env.json/") if "csrftoken" in env: self._csrftoken = env["csrftoken"]
[docs] def load(self, path, raw=False, **kwargs): """Load the required resource from the host. If a login was successful, cookie session information are attached to the request. Also the CRFS-Token is added, if data is set. All values from data are transmitted via POST command. Parameters ========== path : string path to the resource raw : boolean if True no JSON decoding is done **kwargs : dict Additional parameters which will be send as POST request. """ for cookie in self.cj: if == "csrftoken": self._csrftoken = cookie.value if self._csrftoken: kwargs["csrfmiddlewaretoken"] = self._csrftoken kwargs["api"] = 1 payload = urlencode(kwargs).encode("utf-8") else: payload = None try: response =, path), payload) except HTTPError as e: return { "status": "error", "code": e.code, "reason": e.msg, "details":"utf-8") } content = if raw: return content try: content = content.decode("utf-8") content = json.loads(content) finally: return content return json.loads("utf-8"))
[docs] def login(self, username, password): """Send login request. This is normally done by constructor. Parameters ========== username : string The username of the user password : string The password of the user Returns ======= True, if login was successful HTML response otherwise """ result = self.load("accounts/login/?next="+quote_plus("/ok.json"), username=username, password=password) try: if result["status"] == "ok": return True except (ValueError, KeyError, TypeError): return result
[docs] def add_group(self, name, public=True): """Creates a new group Parameters ========== name : string Name of the new group public : boolean Is the group joinable? Default: True Notes ===== Requires admin rights. """ return self.load("actions/add_group/", group_name=name, public=public)
[docs] def join_group(self, group_name): """Current user will be added the the selected group. Parameters ========== group_name : string Name of the selected group """ return self.load("actions/join_group/{}/".format(quote_plus(group_name)))
[docs] def list_groups(self): """List all accessible groups Returns ======= Dictionary with key is the name and value are group parameters. """ return self.load("actions/get_groups.json/")
[docs] def get_round(self, round_id): """Joins the selected round.""" return self.load("actions/get_round/{}.json/".format(quote_plus(str(round_id))))
[docs] def get_job_params(self, job_id): """Load the job parameters. This is only used for workers and need admin rights. Parameters ========== job_id : int The ID of the selected job. Returns ======= Dictionary: - algorithm - dataset - params """ return self.load("actions/get_job_params/{}.json/".format(quote_plus(str(job_id))))
[docs] def update_job_details(self, job_id, score_test, score_validation, labels_training, labels_test, labels_validation, extra_scores_test=None, extra_scores_validation=None, message=None, success=True): """Posts the job results to the server. Parameters ---------- job_id : integer The ID of the job score_test : float The score of the test set. Should be in the interval [0,1] score_validation : float The score of the validation set. Should be in the interval [0,1] labels_training : list A list of lables for the training set labels_test : list A list of lables for the test set labels_validation : list A list of labels for the validation set extra_scores_test : dict (optional) A dictionary with further scores (just for information) from the test set. extra_scores_validation : dict (optional) A dictionary with further scores (just for information) from the validation set. message : string (optional) A message from the training set. It could contain more information about the training/test. pictures : list of filenames (optional) A list of filenames with additional visualizations of the test set. success : boolean (default: True) True if the learning process was successful """ return self.load("actions/update_job_details/{}.json/".format(quote_plus(str(job_id))), score_test=score_test, score_validation=score_validation, labels_training=json.dumps(labels_training), labels_test=json.dumps(labels_test), labels_validation=json.dumps(labels_validation), extra_scores_test=json.dumps(extra_scores_test), extra_scores_validation=json.dumps(extra_scores_validation), message=message, success=success)
[docs] def add_job_picture(self, job_id, filename): """Adds a picture to the job. Parameters ========== job_id : int The ID of the job. filename : string Path to picture """ with open(filename, "rb") as file: data = base64.b64encode( return self.load("actions/add_job_picture/{}.json/".format(quote_plus(str(job_id))), filename=filename, data=data)
[docs] def get_datasets(self): """Retrieves all stored datasets. Returns ======= Dict{id: datasetinfo} """ self.dataset_cache = {int(key): val for key,val in self.load("actions/get_datasets.json/").items()} return self.dataset_cache
[docs] def get_dataset_points(self, dataset_id): """Retrieves all points from a given dataset. Parameters ========== dataset_id : int The ID of the dataset Returns ======= List of point dicts. """ dataset_id = int(dataset_id) if not self.dataset_cache or dataset_id not in self.dataset_cache: self.get_datasets() try: return self.load(self.dataset_cache[dataset_id]["url"]) except KeyError: return []
[docs] def ping(self): """Ping the server and try to retrieve a new job to be processed. Returns ======= Job ID if a new job is available or None if not. """ return self.load("actions/ping/")