person_detection/roc.py

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

import requests
import json
from requests.structures import CaseInsensitiveDict


TOKEN = "### A VALID TOKEN ###"

URL = "https://roc.menlo.aetherproject.org/aether-roc-api/aether/v4.0.0/connectivity-service-v4/"
#URL = "https://roc.staging.aether.onlab.us/aether-roc-api/aether/v4.0.0/connectivity-service-v4/"

cameras = "menlo-4g-cameras"
#cameras = "cameras-4g"

MBR_HIGH = 4000000
MBR_LOW = 2000000

roc_headers = CaseInsensitiveDict()
roc_headers["Content-Type"] = "application/json"
roc_headers["Authorization"] = "Bearer " + TOKEN

def get_mbr(device_group):
    url = URL + "vcs/vcs/vcs-{}/slice/mbr".format(device_group)
    response = requests.get(url, headers=roc_headers)
    assert response.status_code == 200, "Failed to get mbr"
    mbr = json.loads(response.text)
    return mbr

def set_mbr(device_group, mbr):
    url = URL + "vcs/vcs/vcs-{}/slice/mbr".format(device_group)
    response = requests.post(url, headers=roc_headers, json=mbr)
    assert response.status_code == 201, "Failed to set mbr"

def set_uplink_mbr_high():
    mbr = {'uplink' : MBR_HIGH}
    set_mbr(cameras, mbr)

def set_uplink_mbr_low():
    mbr = {'uplink' : MBR_LOW}
    set_mbr(cameras, mbr)

if __name__ == '__main__':
    mbr = get_mbr(cameras)
    print("uplink mbr:{}, downlink mbr: {}".format(mbr["uplink"], mbr["downlink"]))
    set_uplink_mbr_low()
    mbr = get_mbr(cameras)
    print("uplink mbr:{}, downlink mbr: {}".format(mbr["uplink"], mbr["downlink"]))
    set_uplink_mbr_high()
    mbr = get_mbr(cameras)
    print("uplink mbr:{}, downlink mbr: {}".format(mbr["uplink"], mbr["downlink"]))