Browse Source

Merge branch '6rc1' into 'master'

6rc1

See merge request sun/pyfbx!14
7rc0
Laurent 3 years ago
parent
commit
4ec493b451
  1. 172
      README.md
  2. 94
      pyfbx/__main__.py
  3. 16
      pyfbx/client.py
  4. 4
      requirements-dev.txt
  5. 2
      setup.py
  6. 5
      tests/conn_test.py

172
README.md

@ -5,72 +5,14 @@
* Full API coverage
* Https and Mdns discovery
* Thin client wrapper library (<200 lines)
* CLI script example to access any command
* Small script to perform most tasks
[![pipeline status](https://framagit.org/sun/pyfbx/badges/master/pipeline.svg)](https://framagit.org/sun/pyfbx/commits/master)
[![coverage report](https://framagit.org/sun/pyfbx/badges/master/coverage.svg)](https://framagit.org/sun/pyfbx/commits/master)
## Script
A script called pyfbx is installed and can be used like this:
To register the application:
```shell
$ pyfbx SuperAppId
```
Once registration is done on the device, write down the application token.
Using this app token, acquire a session token to execute a test command.
```shell
$ pyfbx -t '<TOKEN>' SuperAppId
$ pyfbx -t 'f:<TOKEN_FILE>' SuperAppId
```
You can run any commands on the freebox and retrieve complete result or single value
```shell
$ pyfbx -t f:/home/foo/token.txt SuperAppId -c 'Contacts.Create_a_contact(\
post_data={"display_name": "Sandy Kilo", "first_name": "Sandy", "last_name":"Kilo"})'
{ 'birthday': '',
'company': '',
'display_name': 'Sandy Kilo',
'first_name': 'Sandy',
'id': 17,
'last_name': 'Kilo',
'last_update': 1554378898,
'notes': '',
'photo_url': ''}
$ pyfbx -t f:token.txt -u https://0s2efr3i.fbxos.fr:15628 id -c "Connection.Get_the_current_Connection_status()['rate_up']"
1700
```
_Note_ : Don't forget to escape token and command with quotes.
The complete script help is:
```shell
pyfbx -h
usage: pyfbx [-h] [-c COMMAND] [-t TOKEN] [-v] [-n] [-u URL] app_id
positional arguments:
app_id application identifier
optional arguments:
-h, --help show this help message and exit
-c COMMAND, --command COMMAND
command, defaults to
System.Get_the_current_system_info()
-t TOKEN, --token TOKEN
token (or f:<filename>)
-v, --verbose increase verbosity to INFO, use twice for DEBUG
-n, --http Disable MDNS and use http known address
-u URL, --url URL specific url to query
```
## Library usage
The library is available to create your own scripts.
The REST API is generated at runtime with the creation of class attributes for all Freebox subsystems.
```python
@ -120,6 +62,116 @@ Access_a_given_contact_entry(id) method of pyfbx.client.Contacts instance
>>> f.mksession()
```
## Script
The included script can perform a lot of tasks.
### Basics
```shell
$ pyfbx -h
usage: pyfbx [-h] [-a APP_ID] [-t TOKEN] [-v] [-n] [-j] [-d DELAY] [-u URL]
[-s SEND] [-c COMMAND]
optional arguments:
-h, --help show this help message and exit
-a APP_ID, --app_id APP_ID
application identifier
-t TOKEN, --token TOKEN
token (or f:<filename>)
-v, --verbose increase verbosity to INFO, use twice for DEBUG
-n, --http disable MDNS and use http known address
-j, --json json output
-d DELAY, --delay DELAY
cylically send command (number of seconds)
-u URL, --url URL specific url to query
-s SEND, --send SEND url to send json to
-c COMMAND, --command COMMAND
command, defaults to
System.Get_the_current_system_info()
```
First, register the application:
```shell
$ pyfbx SuperAppId
```
Once registration is done on the device, write down the application token.
You can also create a file with token as first line and application Id as second.
Using this app token, acquire a session token to execute a test command.
```shell
$ pyfbx -t '<TOKEN>' SuperAppId
$ pyfbx -t 'f:<TOKEN_FILE>' SuperAppId
```
### Advanced
You can run any commands on the freebox and retrieve complete result or single value
```shell
$ pyfbx -t f:/home/foo/token.txt SuperAppId -c 'Contacts.Create_a_contact(\
post_data={"display_name": "Sandy Kilo", "first_name": "Sandy", "last_name":"Kilo"})'
{ 'birthday': '',
'company': '',
'display_name': 'Sandy Kilo',
'first_name': 'Sandy',
'id': 17,
'last_name': 'Kilo',
'last_update': 1554378898,
'notes': '',
'photo_url': ''}
$ pyfbx -u https://0s2efr3i.fbxos.fr:15628 id -c "Connection.Get_the_current_Connection_status()['rate_up']" -t f:token.txt
1700
```
With the delay option, commands can be sent cyclically:
```shell
$ pyfbx -d 1 -c "Connection.Get_the_current_Connection_status()['rate_up']" -t f:token.txt
42460
50710
58890
53120
53400
```
With the send option, result can be sent to a remote URL.
In the following, the result is sent cyclically to example.com in json format.
```shell
pyfbx -j -s http://example.com/values -d 10 -c "Connection.Get_the_current_Connection_status()" -t f:token.txt
```
It's possible to execute several commands.
Here, two commands are sent cyclically and results sent to an URL.
```shell
pyfbx -c 'System.Get_the_current_system_info()' -c 'Connection.Get_the_current_Connection_status()' -d 10 -j -s http://localhost:8182/telegraf -t f:token.txt
```
Telegraf http listener v2 input plugin with json format can be used to plot data in realtime.
_Note_ : Don't forget to escape token and command with quotes.
### Telegraf plots
You can use a telegraph configuration in /etc/telegraf/telegraf.d/freebox:
```
[[inputs.http_listener_v2]]
service_address = ":8182"
path = "/telegraf"
methods = ["POST"]
read_timeout = "10s"
write_timeout = "10s"
max_body_size = "5KB"
data_format = "json"
name_override = "pyfbx"
```
## Developpment
### Testing

94
pyfbx/__main__.py

@ -2,9 +2,12 @@
Console script
"""
import os
import time
import logging
import json
import argparse
import pprint
import requests
from pyfbx import Fbx
log_level = (logging.WARNING, logging.INFO, logging.DEBUG)
@ -12,48 +15,85 @@ log_level = (logging.WARNING, logging.INFO, logging.DEBUG)
def console(log, level):
log.setLevel(level)
formatter = logging.Formatter('%(asctime)s - %(name)14s - %(levelname)s - %(message)s')
if level == logging.DEBUG:
formatter = logging.Formatter('%(asctime)s - %(name)14s - %(levelname)7s - %(funcName)10s:%(lineno)3d - %(message)s')
else:
formatter = logging.Formatter('%(asctime)s - %(name)14s - %(levelname)7s - %(message)s')
stream_handler = logging.StreamHandler()
stream_handler.setFormatter(formatter)
log.addHandler(stream_handler)
def output(fbx, command, json_output, send_url):
try:
res = eval("fbx.{}".format(command))
except Exception as exc:
if "403 Client Error" in str(exc):
fbx.log.info("Got 403, refreshing token")
fbx.mksession()
res = eval("fbx.{}".format(command))
else:
raise(exc)
if send_url:
try:
r = requests.post(send_url, json=res)
except BaseException as exc:
print("While sending to {}, got exception {}".format(send_url, exc))
elif json_output:
print(json.dumps(res))
else:
pp = pprint.PrettyPrinter(indent=4)
pp.pprint(res)
def main():
log = logging.getLogger("pyfbx")
parser = argparse.ArgumentParser()
parser.add_argument("app_id", type=str, help="application identifier")
parser.add_argument("-c", "--command", type=str,
help="command, defaults to System.Get_the_current_system_info()",
default="System.Get_the_current_system_info()")
parser.add_argument("-t", "--token", type=str, help="token (or f:<filename>)")
parser.add_argument("-a", "--app_id", type=str,
help="application identifier")
parser.add_argument("-t", "--token", type=str,
help="token (or f:<filename>)")
parser.add_argument("-v", "--verbose", action="count", default=0,
help="increase verbosity to INFO, use twice for DEBUG")
parser.add_argument("-n", "--http", action="store_true",
help="disable MDNS and use http known address")
parser.add_argument("-j", "--json", action="store_true", default=False,
help="json output")
parser.add_argument("-d", "--delay", type=int,
help="cylically send command (number of seconds)")
parser.add_argument("-u", "--url", type=str,
help="specific url to query",
default=None)
help="specific url to query")
parser.add_argument("-s", "--send", type=str,
help="url to send json to")
parser.add_argument("-c", "--command", action='append',
help="command, defaults to System.Get_the_current_system_info()")
args = parser.parse_args()
console(log, log_level[min(2, args.verbose)])
try:
myfb = Fbx(nomdns=args.http, url=args.url)
token = args.token
if token:
if token.startswith('f:'):
with open(args.token[2:]) as tok_file:
token = tok_file.read().strip()
else:
log.warning("Registering app {}, id {}, Press button".format(__name__, args.app_id))
token = myfb.register(app_id=args.app_id, app_name=__name__,
device=os.uname().nodename)
log.warning("Save your application token: {}".format(token))
myfb.mksession(app_id=args.app_id, token=token)
pp = pprint.PrettyPrinter(indent=4)
pp.pprint(eval("myfb.{}".format(args.command)))
return 0
except BaseException as err:
print(err)
return 2
token = args.token
app_id = args.app_id
if not args.command:
args.command = ["System.Get_the_current_system_info()"]
myfb = Fbx(nomdns=args.http, url=args.url)
if token:
if token.startswith('f:'):
with open(args.token[2:]) as tok_file:
token, app_id = tok_file.read().splitlines()
else:
log.warning("Registering app {}, id {}, Press button".format(__name__, app_id))
token = myfb.register(app_id=app_id, app_name=__name__,
device=os.uname().nodename)
log.warning("Save your application token: {}".format(token))
myfb.mksession(app_id=app_id, token=token)
while True:
log.debug("\n")
for command in args.command:
output(myfb, command, args.json, args.send)
if not args.delay:
break
time.sleep(args.delay)
return 0
if __name__ == "__main__":

16
pyfbx/client.py

@ -42,14 +42,14 @@ class FbxTransport():
def set_url(self, url, nomdns):
if url is None:
if nomdns:
self._url = self.get_local_base()
self._url = self.local_base()
else:
self._url = mdns.FbxMDNS().search() or self.get_local_base()
self._url = mdns.FbxMDNS().search() or self.local_base()
else:
if re.search("https?://", url) is None:
url = "http://" + url
if "/api/" not in url:
self._url = self.get_local_base(url)
self._url = self.local_base(url)
else:
self._url = url
@ -71,7 +71,7 @@ class FbxTransport():
else:
raise FbxErrorResponse(response['error_code'], response['msg'])
def get_local_base(self, url=api._DISC_HTTP_URL):
def local_base(self, url=api._DISC_HTTP_URL):
response = self._session.get("{}/api_version".format(url)).json()
self.log.debug("<< Detected api {}".format(response['api_version']))
return "%s%sv%s" % (url, response['api_base_url'],
@ -86,6 +86,8 @@ class Fbx():
def __init__(self, url=None, nomdns=False, session=None):
self.log = logging.getLogger("pyfbx.fbx")
self._trn = FbxTransport(url, nomdns=nomdns, session=session)
self.app_id = None
self.token = None
# Create on the fly attributes to classes
_globals = globals()
@ -112,10 +114,14 @@ class Fbx():
return s == "granted" and self.token
def mksession(self, app_id=None, token=None):
self.log.debug("Making session with token={}[{}], app_id={}[{}]".format(
token, self.token, self.app_id, app_id))
if token: # Don't overwrite previous token (used for refresh)
self.token = token
if app_id:
self.app_id = app_id
elif not self.app_id:
raise Exception("Missing app_id")
login = self._trn.api_exec("GET", "login/")
if not login['logged_in']:
data = {
@ -128,7 +134,7 @@ class Fbx():
session_token = resp["session_token"]
self.app_id = app_id
self._trn.set_session_header(session_token)
self.log.info("Authenticated".format())
self.log.info("Authenticated. Storing token={}, app_id={}".format(self.token, self.app_id))
return resp["permissions"]

4
requirements-dev.txt

@ -1,5 +1,5 @@
mock
pytest
pytest-cov
pytest>=4.4.1
pytest-cov>=2.6.1
twine
wheel

2
setup.py

@ -3,7 +3,7 @@ from setuptools import setup, find_packages
setup(
name="pyfbx",
version="0.0.5",
version="0.0.6rc1",
description="Freebox thin client",
long_description=(pathlib.Path(__file__).parent / "README.md").read_text(),
long_description_content_type='text/markdown',

5
tests/conn_test.py

@ -14,6 +14,8 @@ from pyfbx.mdns import FbxMDNS
def test_fbx_register():
f = Fbx("http://192.168.1.254/api/v6")
with pytest.raises(Exception):
f.mksession()
with patch('pyfbx.client.FbxTransport.api_exec',
side_effect=iter([
{"track_id": "id", "app_token": "tok"},
@ -27,8 +29,7 @@ def test_fbx_session_local():
tokname = os.path.join(os.path.dirname(__file__), 'token.txt')
with open(tokname) as tokfile:
token = tokfile.readline().rstrip()
app_id = tokfile.readline().rstrip()
token, app_id = tokfile.read().splitlines()
print("token:{}, app_id:{}".format(token, app_id))
f = Fbx()
f.mksession(app_id=app_id, token=token)

Loading…
Cancel
Save