roc.py

"""
SPDX-FileCopyrightText: 2022-present Intel Corporation
SPDX-FileCopyrightText: 2020-present Open Networking Foundation <info@opennetworking.org>
SPDX-License-Identifier: Apache-2.0
"""

import sys
import requests
import json
# noinspection PyUnresolvedReferences
from requests.structures import CaseInsensitiveDict
import logging as log
from datetime import datetime

from device import Device


class Roc(object):

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

        self.enterprise = None
        self.update()

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

    def get_key(self):
        url = "https://keycloak.opennetworking.org/auth/realms/master/protocol/openid-connect/token"
        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(url, data, headers)
        key = json.loads(response.text)['access_token']
        return key

    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_site(self):
        sites = self.enterprise['site']
        for site in sites:
            for k, v in site.items():
                if k == 'site-id' and v == self.siteId:
                    return site
        return None

    def devices(self):
        site = self.get_site()
        devices = site['device']
        for device in devices:
            yield device

    def get_imsi(self, simId):
        site = self.get_site()
        sims = site['sim-card']
        for sim in sims:
            if sim['sim-id'] == simId:
                return str(sim['imsi'])
        return None

    def update_devices(self, devices):
        self.update()
        new = {}
        for d in self.devices():
            imsi_id = str(d['device-id'])
            last_reachable = datetime.min
            imsi = self.get_imsi(d['sim-card'])
            if imsi_id in devices:
                last_reachable = devices[imsi_id].last_reachable
            new[imsi_id] = Device(imsi_id, imsi, last_reachable)
        return new


if __name__ == '__main__':
    url = "https://roc.menlo.aetherproject.org/aether-roc-api/aether/v2.0.0/connectivity-service-v2/"
    user = sys.argv[1]
    password = sys.argv[2]
    enterprise = 'aether-onf'
    site = 'menlo-4g'
    roc = Roc(url, user, password, enterprise, site)
    devices = roc.update_devices({})
    for device in devices.values():
        print(device)