| """ |
| SPDX-FileCopyrightText: 2020-present Open Networking Foundation <info@opennetworking.org> |
| SPDX-License-Identifier: LicenseRef-ONF-Member-1.01 |
| """ |
| |
| 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) |