-
Notifications
You must be signed in to change notification settings - Fork 7
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Fix library structure and some typo name
- Loading branch information
1 parent
d8bbfcc
commit 6d12cbb
Showing
6 changed files
with
258 additions
and
258 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,254 @@ | ||
from .bboxConstant import BboxConstant | ||
from .bboxAuth import BboxAuth | ||
from .bboxApiURL import BboxAPIUrl | ||
from .bboxApiCall import BboxApiCall | ||
|
||
|
||
class Bbox: | ||
""" | ||
Class to interact with Bouygues Bbox Modem Routeur | ||
API Reference used for this : https://api.bbox.fr/doc/apirouter/ | ||
""" | ||
|
||
def __init__(self, ip=BboxConstant.DEFAULT_LOCAL_IP): | ||
""" | ||
Initiate a Bbox instance with a default local ip (192.168.1.254) | ||
:param ip: Ip of the box if you do not want the default one | ||
:type ip: str | ||
:return: A Bbox Api Instance | ||
:rtype: None | ||
""" | ||
self.bbox_url = BboxAPIUrl(None, None, ip) | ||
self.bbox_auth = BboxAuth(None, None, False, self.bbox_url.authentication_type) | ||
|
||
@property | ||
def get_access_type(self): | ||
""" | ||
Return if the access is made on the local network or remotely | ||
:return: AUTHENTICATION_TYPE_LOCAL (0) or AUTHENTICATION_TYPE_REMOTE (1) | ||
:rtype: int | ||
""" | ||
return self.bbox_url.authentication_type | ||
|
||
""" | ||
USEFUL FUNCTIONS | ||
""" | ||
|
||
""" | ||
DEVICE API | ||
""" | ||
|
||
def get_bbox_info(self): | ||
""" | ||
This API returns Bbox information | ||
:return: numbers of info about the box itself (see API doc) | ||
:rtype: dict | ||
""" | ||
self.bbox_auth.set_access(BboxConstant.AUTHENTICATION_LEVEL_PUBLIC, BboxConstant.AUTHENTICATION_LEVEL_PRIVATE) | ||
self.bbox_url.set_api_name(BboxConstant.API_DEVICE, None) | ||
api = BboxApiCall(self.bbox_url, BboxConstant.HTTP_METHOD_GET, None, | ||
self.bbox_auth) | ||
resp = api.execute_api_request() | ||
return resp.json()[0] | ||
|
||
def set_display_luminosity(self, luminosity): | ||
""" | ||
Change the intensity of light of the front panel of the box | ||
:param luminosity: must be between 0 (light off) and 100 | ||
:type luminosity: int | ||
""" | ||
if (luminosity < 0) or (luminosity > 100): | ||
raise ValueError("Luminosity must be between 0 and 100") | ||
self.bbox_auth.set_access(BboxConstant.AUTHENTICATION_LEVEL_PRIVATE, | ||
BboxConstant.AUTHENTICATION_LEVEL_PRIVATE) | ||
self.bbox_url.set_api_name(BboxConstant.API_DEVICE, "display") | ||
data = {'luminosity': luminosity} | ||
api = BboxApiCall(self.bbox_url, BboxConstant.HTTP_METHOD_PUT, data, | ||
self.bbox_auth) | ||
api.execute_api_request() | ||
|
||
def reboot(self): | ||
""" | ||
Reboot the device | ||
Useful when trying to get xDSL sync | ||
""" | ||
token = self.get_token() | ||
self.bbox_auth.set_access(BboxConstant.AUTHENTICATION_LEVEL_PRIVATE, BboxConstant.AUTHENTICATION_LEVEL_PRIVATE) | ||
url_suffix = "reboot?btoken={}".format(token) | ||
self.bbox_url.set_api_name(BboxConstant.API_DEVICE, url_suffix) | ||
api = BboxApiCall(self.bbox_url, BboxConstant.HTTP_METHOD_POST, None, | ||
self.bbox_auth) | ||
api.execute_api_request() | ||
|
||
def get_token(self): | ||
""" | ||
Return a string which is a token, needed for some API calls | ||
:return: Token (can be used with some API call | ||
:rtype: str | ||
.. todo:: make a token class to be able to store date of expiration | ||
""" | ||
self.bbox_auth.set_access(BboxConstant.AUTHENTICATION_LEVEL_PRIVATE, BboxConstant.AUTHENTICATION_LEVEL_PRIVATE) | ||
self.bbox_url.set_api_name(BboxConstant.API_DEVICE, "token") | ||
api = BboxApiCall(self.bbox_url, BboxConstant.HTTP_METHOD_GET, None, | ||
self.bbox_auth) | ||
resp = api.execute_api_request() | ||
return resp.json()[0]['device']['token'] | ||
|
||
""" | ||
LAN API | ||
""" | ||
|
||
def get_all_connected_devices(self): | ||
""" | ||
Get all info about devices connected to the box | ||
:return: a list with each device info | ||
:rtype: list | ||
""" | ||
self.bbox_auth.set_access(BboxConstant.AUTHENTICATION_LEVEL_PUBLIC, BboxConstant.AUTHENTICATION_LEVEL_PRIVATE) | ||
self.bbox_url.set_api_name(BboxConstant.API_HOSTS, None) | ||
api = BboxApiCall(self.bbox_url, BboxConstant.HTTP_METHOD_GET, None, | ||
self.bbox_auth) | ||
resp = api.execute_api_request() | ||
return resp.json()[0]["hosts"]["list"] | ||
|
||
def is_device_connected(self, ip): | ||
""" | ||
Check if a device identified by it IP is connected to the box | ||
:param ip: IP of the device you want to test | ||
:type ip: str | ||
:return: True is the device is connected, False if it's not | ||
:rtype: bool | ||
""" | ||
all_devices = self.get_all_connected_devices() | ||
for device in all_devices: | ||
if ip == device['ipaddress']: | ||
return device['active'] == 1 | ||
return False | ||
|
||
""" | ||
USER ACCOUNT | ||
""" | ||
|
||
def login(self, password): | ||
""" | ||
Authentify yourself against the box, | ||
:param password: Admin password of the box | ||
:type password: str | ||
:return: True if your auth is successful | ||
:rtype: bool | ||
""" | ||
self.bbox_auth.set_access(BboxConstant.AUTHENTICATION_LEVEL_PUBLIC, BboxConstant.AUTHENTICATION_LEVEL_PUBLIC) | ||
self.bbox_url.set_api_name("login", None) | ||
data = {'password': password} | ||
api = BboxApiCall(self.bbox_url, BboxConstant.HTTP_METHOD_POST, data, | ||
self.bbox_auth) | ||
response = api.execute_api_request() | ||
if response.status_code == 200: | ||
self.bbox_auth.set_cookie_id(response.cookies["BBOX_ID"]) | ||
return self.bbox_auth.is_authentified() | ||
|
||
def logout(self): | ||
""" | ||
Destroy the auth session against the box | ||
:return: True if your logout is successful | ||
:rtype: bool | ||
""" | ||
self.bbox_auth.set_access(BboxConstant.AUTHENTICATION_LEVEL_PUBLIC, BboxConstant.AUTHENTICATION_LEVEL_PUBLIC) | ||
self.bbox_url.set_api_name("logout", None) | ||
api = BboxApiCall(self.bbox_url, BboxConstant.HTTP_METHOD_POST, None, | ||
self.bbox_auth) | ||
response = api.execute_api_request() | ||
if response.status_code == 200: | ||
self.bbox_auth.set_cookie_id(None) | ||
return not self.bbox_auth.is_authentified() | ||
|
||
""" | ||
WAN API | ||
""" | ||
|
||
def get_xdsl_info(self): | ||
""" | ||
Get all data about your xDSL connection | ||
:return: A dict with all data about your xdsl connection (see API doc) | ||
:rtype: dict | ||
""" | ||
self.bbox_auth.set_access(BboxConstant.AUTHENTICATION_LEVEL_PUBLIC, BboxConstant.AUTHENTICATION_LEVEL_PRIVATE) | ||
self.bbox_url.set_api_name(BboxConstant.API_WAN, "xdsl") | ||
api = BboxApiCall(self.bbox_url, BboxConstant.HTTP_METHOD_GET, None, | ||
self.bbox_auth) | ||
resp = api.execute_api_request() | ||
return resp.json()[0]["wan"]["xdsl"] | ||
|
||
def get_xdsl_stats(self): | ||
""" | ||
Get all stats about your xDSL connection | ||
:return: A dict with all stats about your xdsl connection (see API doc) | ||
:rtype: dict | ||
""" | ||
self.bbox_auth.set_access(BboxConstant.AUTHENTICATION_LEVEL_PUBLIC, BboxConstant.AUTHENTICATION_LEVEL_PRIVATE) | ||
self.bbox_url.set_api_name(BboxConstant.API_WAN, "xdsl/stats") | ||
api = BboxApiCall(self.bbox_url, BboxConstant.HTTP_METHOD_GET, None, | ||
self.bbox_auth) | ||
resp = api.execute_api_request() | ||
return resp.json()[0]["wan"]["xdsl"]["stats"] | ||
|
||
def get_ip_stats(self): | ||
""" | ||
Get all stats about your Wan ip connection | ||
:return: A dict with all stats about your Wan ip connection (see API doc) | ||
:rtype: dict | ||
""" | ||
self.bbox_auth.set_access(BboxConstant.AUTHENTICATION_LEVEL_PUBLIC, BboxConstant.AUTHENTICATION_LEVEL_PRIVATE) | ||
self.bbox_url.set_api_name(BboxConstant.API_WAN, "ip/stats") | ||
api = BboxApiCall(self.bbox_url, BboxConstant.HTTP_METHOD_GET, None, | ||
self.bbox_auth) | ||
resp = api.execute_api_request() | ||
return resp.json()[0]["wan"]["ip"]["stats"] | ||
|
||
def is_bbox_connected(self): | ||
""" | ||
Check if your xDsl connection is ok | ||
:return: True is the box has an xdsl connection | ||
:rtype: bool | ||
""" | ||
xdsl_info = self.get_xdsl_info() | ||
return xdsl_info["state"] == "Connected" | ||
|
||
def get_up_bitrates(self): | ||
""" | ||
:return: the upload bitrates of the xdsl connectionbitrates in Mbps | ||
:rtype: float | ||
""" | ||
xdsl_info = self.get_xdsl_info() | ||
return xdsl_info["up"]["bitrates"] / 1000 | ||
|
||
def get_down_bitrates(self): | ||
""" | ||
:return: the download bitrates of the xdsl connectionbitrates in Mbps | ||
:rtype: float | ||
""" | ||
xdsl_info = self.get_xdsl_info() | ||
return xdsl_info["down"]["bitrates"] / 1000 | ||
|
||
def get_up_used_bandwith(self): | ||
""" | ||
Return a percentage of the current used xdsl upload bandwith | ||
Instant measure, can be very different from one call to another | ||
:return: 0 no bandwith is used, 100 all your bandwith is used | ||
:rtype: int | ||
""" | ||
ip_stats_up = self.get_ip_stats()['tx'] | ||
percent = ip_stats_up['bandwidth']*100/ip_stats_up['maxBandwidth'] | ||
return int(percent) | ||
|
||
def get_down_used_bandwith(self): | ||
""" | ||
Return a percentage of the current used xdsl download bandwith | ||
Instant measure, can be very different from one call to another | ||
:return: 0 no bandwith is used, 100 all your bandwith is used | ||
:rtype: int | ||
""" | ||
ip_stats_up = self.get_ip_stats()['rx'] | ||
percent = ip_stats_up['bandwidth']*100/ip_stats_up['maxBandwidth'] | ||
return int(percent) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.