Source code for labgrid.driver.power.digitalloggers_restapi

'''
Driver for Digital Loggers PDU that use the REST API.
Tested with Ethernet Power Controller 7.

Based on https://www.digital-loggers.com/restapi.pdf

By default, only an authenticated user is allowed by REST API.

NetworkPowerPort:
    model: 'digitalloggers_restapi'
    host: 'http://admin:1234@192.168.0.100'
    index: 0
'''
from urllib.parse import urlparse

import requests
from requests.auth import HTTPDigestAuth
from requests.packages import urllib3




def extract_user_password_from_host(host):
    url = urlparse(host)
    if '@' in url.netloc:
        user=url.username
        password=url.password
        _host= f'{url.scheme}://{url.netloc.split("@")[1]}'
    else:
        user = None
        password = None
        _host= f'{url.scheme}://{url.netloc}'
    return user, password, _host




def power_set(host, port, index, value):
    # curl -u admin:1234 -v -X PUT -H "X-CSRF: x" --data 'value=true' --digest http://192.168.0.100/restapi/relay/outlets/=0/state/
    # curl -u admin:1234 -v -X PUT -H "X-CSRF: x" --data 'value=false' --digest http://192.168.0.100/restapi/relay/outlets/=0/state/
    # curl --insecure -u admin:1234 -v -X PUT -H "X-CSRF: x" --data 'value=true' --digest https://192.168.0.100/restapi/relay/outlets/=0/state/
    # curl --insecure -u admin:1234 -v -X PUT -H "X-CSRF: x" --data 'value=false' --digest https://192.168.0.100/restapi/relay/outlets/=0/state/
    assert port is None

    index = int(index)
    value = 'true' if value else 'false'
    payload = {'value' : value }
    headers = {'X-CSRF': 'x', 'Accept': 'application/json'}
    user, password, url = extract_user_password_from_host(host)
    host = f'{url}/restapi/relay/outlets/={index}/state/'
    with urllib3.warnings.catch_warnings():
        urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
        if user and password:
            r = requests.put(host, data=payload, auth=HTTPDigestAuth(user, password), headers=headers, verify=False)
        else:
            r = requests.put(host, data=payload, headers=headers, verify=False)
        r.raise_for_status()




def power_get(host, port, index):
    # curl -u admin:1234 -v -X GET -H "X-CSRF: x" --digest http://192.168.0.100/restapi/relay/outlets/=0/state/
    # curl --insecure -u admin:1234 -v -X GET -H "X-CSRF: x" --digest https://192.168.0.100/restapi/relay/outlets/=0/state/
    assert port is None

    index = int(index)
    user, password, url = extract_user_password_from_host(host)
    headers = {'X-CSRF': 'x', 'Accept': 'application/json'}
    host = f'{url}/restapi/relay/outlets/={index}/state/'
    with urllib3.warnings.catch_warnings():
        urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
        if user and password:
            r = requests.get(host, auth=HTTPDigestAuth(user, password), headers=headers, verify=False)
        else:
            r = requests.get(host, headers=headers, verify=False)
        r.raise_for_status()
    statuses = r.json()
    return statuses[0]