person_detection/roc.py

"""
SPDX-FileCopyrightText: 2020-present Open Networking Foundation <info@opennetworking.org>
SPDX-License-Identifier: LicenseRef-ONF-Member-1.01
"""

import sys
import requests
import json
from requests.structures import CaseInsensitiveDict
import logging as log


class Roc(object):

    def __init__(self, url, keycloak, user, password, enterpriseId, siteId):
        self.url = url
        self.keycloak = keycloak
        self.user = user
        self.password = password
        self.key = self.get_key()
        self.enterpriseId = enterpriseId
        self.siteId = siteId
        self.update()

    def headers(self):
        h = CaseInsensitiveDict()
        h["Content-Type"] = "application/json"
        h["Authorization"] = "Bearer " + self.key
        return h

    def update(self):
        url = self.url + "enterprises"
        response = requests.get(url, headers=self.headers())
        if response.status_code != 200:
            log.error("get_enterprise() failed, status_code: {}".format(response.status_code))
            return None

        result = json.loads(response.text)
        enterprises = result['enterprise']

        for enterprise in enterprises:
            for k, v in enterprise.items():
                if k == 'enterprise-id' and v == self.enterpriseId:
                    self.enterprise = enterprise

    def get_mbr(self, device_group):
        url = self.url + "enterprises/enterprise/{}/site/{}/device-group/{}/mbr".format(self.enterpriseId, self.siteId, device_group)
        response = requests.get(url, headers=self.headers())
        if response.status_code != 200:
            log.error("Failed to get mbr, status_code: {}".format(response.status_code))
            return None
        mbr = json.loads(response.text)
        return mbr

    def set_mbr(self, device_group, mbr):
        log.info("Set {} mbr to {}".format(device_group, mbr))
        m = {"downlink": 10000000, "uplink": mbr}
        url = self.url + "enterprises/enterprise/{}/site/{}/device-group/{}/mbr".format(self.enterpriseId, self.siteId, device_group)
        response = requests.post(url, headers=self.headers(), json=m)

        # If error, renew api key and try again
        if response.status_code != 201:
            log.info("Renew ROC api key")
            self.key = self.get_key()
            response = requests.post(url, headers=self.headers(), json=m)
            if response.status_code != 201:
                log.error("Failed to set mbr, device_group:{}, mbr:{}, status_code: {}".format(device_group, mbr, response.status_code))
                # sys.exit()

    def get_key(self):
        headers = CaseInsensitiveDict()
        headers["Content-Type"] = "application/x-www-form-urlencoded"
        data = {
                'grant_type': 'password',
                'client_id': 'aether-roc-gui',
                'username': self.user,
                'password': self.password,
                'scope': 'openid profile email groups'
                }
        response = requests.post(self.keycloak, data, headers)
        key = json.loads(response.text)['access_token']
        return key

if __name__ == '__main__':

    log.basicConfig(
            format='%(asctime)s %(levelname)-8s %(message)s',
            level=log.DEBUG,
            datefmt='%Y-%m-%d %H:%M:%S',
            stream=sys.stdout)
    log.debug("Testing roc...")

    # use valid keycloak user/password
    url = "http://roc.aetherproject.net/aether-roc-api/aether/v2.0.0/connectivity-service-v2/"
    user = sys.argv[1]
    password = sys.argv[2]
    enterprise = 'ase-4g-enterprise'
    site = 'ase-site-4g'
    keycloak = "http://keycloak.aetherproject.net/auth/realms/aether/protocol/openid-connect/token"

    roc = Roc(url, keycloak,  user, password, enterprise, site)

    cameras = "ase-4g-device-group2"
    mbr = roc.get_mbr(cameras)
    print("uplink mbr:{}, downlink mbr: {}".format(mbr["uplink"], mbr["downlink"]))

    roc.set_mbr(cameras, 6000000)
    mbr = roc.get_mbr(cameras)
    print("uplink mbr:{}, downlink mbr: {}".format(mbr["uplink"], mbr["downlink"]))

    roc.set_mbr(cameras, 10000000)
    mbr = roc.get_mbr(cameras)
    print("uplink mbr:{}, downlink mbr: {}".format(mbr["uplink"], mbr["downlink"]))