#!/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__ = "paheld@gmail.com"
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.
"""
self.host = 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 cookie.name == "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 = self.opener.open(normalize_url(self.host, path), payload)
except HTTPError as e:
return {
"status": "error",
"code": e.code,
"reason": e.msg,
"details": e.fp.read().decode("utf-8")
}
content = response.read()
if raw:
return content
try:
content = content.decode("utf-8")
content = json.loads(content)
finally:
return content
return json.loads(response.read().decode("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(file.read())
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/")