Skip to content

Instantly share code, notes, and snippets.

@indisoluble

indisoluble/connman_agent.py

Last active November 29, 2016 14:45
Show Gist options
  • Save indisoluble/cf4956ae65e855a308ba8e82a7ea9b4f to your computer and use it in GitHub Desktop.
Save indisoluble/cf4956ae65e855a308ba8e82a7ea9b4f to your computer and use it in GitHub Desktop.
Download ZIP
Simple Web Server (based on CherryPy) to configure Internet Connection (based on Connman + Pyconnman)
Raw
connman_agent.py
#!/usr/bin/python
# Imports
from argparse import ArgumentParser
from gi.repository import GObject
import connman_shared as shared
import dbus.mainloop.glib
import pyconnman
# Constants
ARG_AGENT_PATH = 'agent_path'
ARG_PASSPHRASE = 'passphrase'
ARG_SERV_NAME = 'service_name'
ARG_SERV_PATH = 'service_path'
# Functions
# Classes
# Main
if __name__ == '__main__':
parser = ArgumentParser(description = 'Connman DBus Agent')
requiredArgs = parser.add_argument_group('required arguments')
requiredArgs.add_argument('--' + ARG_AGENT_PATH, required = True)
requiredArgs.add_argument('--' + ARG_SERV_PATH, required = True)
requiredArgs.add_argument('--' + ARG_SERV_NAME, required = True)
requiredArgs.add_argument('--' + ARG_PASSPHRASE, required = True)
args = vars(parser.parse_args())
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
manager = pyconnman.ConnManager()
agent = pyconnman.SimpleWifiAgent(args[ARG_AGENT_PATH])
agent.set_service_params(service = args[ARG_SERV_PATH],
name = args[ARG_SERV_NAME],
passphrase = args[ARG_PASSPHRASE])
manager.register_agent(args[ARG_AGENT_PATH])
mainloop = GObject.MainLoop()
mainloop.run()
Raw
connman_server.py
#!/usr/bin/python
# Imports
from argparse import ArgumentParser
from cgi import escape
from dbus.exceptions import DBusException
from os import getpgid, killpg, setsid
from signal import SIGTERM
from subprocess import Popen
from sys import exc_info
from time import sleep
import cherrypy
import connman_agent as agent
import pyconnman
# Constants
AGENT_APP = './connman_agent.py'
AGENT_PATH = '/comodoo_pos/agent'
ARGUMENT_HOST = 'host'
ARGUMENT_PORT = 'port'
CONNECTED_SERVICE_STATE = ['online', 'ready']
KEY_NAME = 'Name'
KEY_STATE = 'State'
SLEEP_TIME = 1
TECH_PROPERTY_POWERED = 'Powered'
# Functions
def webpage_with_body(body):
return '''
<html>
<head></head>
<body>
%s
</body>
</html>
''' % (body)
def webpage_with_orderedlist(ol):
body = '<ol>'
for element in ol:
body+= '<li>' + element + '</li>'
body += '</ol>'
return webpage_with_body(body)
def html_link_with_text(text, link):
return '''<a href="%s">%s</a>''' % (link, text)
def html_paragraph_with_text(text):
return '''<p>%s</p>''' % (text)
def html_section_with_html_element(element):
return '''
<div>
%s
</div>
''' % (element)
def html_inline_button_with_text(text, action,
param1 = None, value1 = None,
param2 = None, value2 = None):
element = '<form method="post" style="display:inline-block" '
element += 'action="' + action + '" '
element += '>'
if param1 and value1:
element += '<input type="hidden" '
element += 'name="' + param1 + '" '
element += 'value="' + value1 + '" '
element += '/>'
if param2 and value2:
element += '<input type="hidden" '
element += 'name="' + param2 + '" '
element += 'value="' + value2 + '" '
element += '/>'
element += '<button type="submit">' + text + '</button>'
element += '</form>'
return element
def html_button_with_text(text, action,
inputParam,
hiddenParam1, hiddenValue1,
hiddenParam2, hiddenValue2):
return '''
<form method="post" action="%s">
<input type="hidden" name="%s" value="%s" />
<input type="hidden" name="%s" value="%s" />
<input type="text" name="%s" />
<button type="submit">%s</button>
</form>
''' % (action,
hiddenParam1, hiddenValue1,
hiddenParam2, hiddenValue2,
inputParam,
text)
# Classes
class Server(object):
@cherrypy.expose
def index(self):
lt = html_link_with_text('List Technologies',
link = 'list_technologies')
ls = html_link_with_text('List Services',
link = 'list_services')
lcs = html_link_with_text('List Connected Services',
link = 'list_connected_services')
return webpage_with_orderedlist([lt, ls, lcs])
@cherrypy.expose
def list_technologies(self):
manager = pyconnman.ConnManager()
technologies = manager.get_technologies()
ol = []
for (path, params) in technologies:
text = (params[KEY_NAME]
if (KEY_NAME in params)
else path)
link = 'technology_info?tech_path=' + escape(path)
element = html_link_with_text(text, link = link)
element += html_inline_button_with_text('On',
action = 'power_technology',
param1 = 'tech_path',
value1 = escape(path),
param2 = 'on',
value2 = '1')
element += html_inline_button_with_text('Off',
action = 'power_technology',
param1 = 'tech_path',
value1 = escape(path),
param2 = 'on',
value2 = '0')
ol.append(html_section_with_html_element(element))
return webpage_with_orderedlist(ol)
@cherrypy.expose
def technology_info(self, tech_path):
tech = pyconnman.ConnTechnology(tech_path)
return webpage_with_body(tech)
@cherrypy.expose
def power_technology(self, tech_path, on):
try:
tech = pyconnman.ConnTechnology(tech_path)
tech.set_property(TECH_PROPERTY_POWERED, int(on))
except DBusException:
print 'Error: ', exc_info()
raise cherrypy.HTTPRedirect('list_services')
@cherrypy.expose
def list_services(self):
manager = pyconnman.ConnManager()
services = manager.get_services()
ol = []
for (path, params) in services:
text = (params[KEY_NAME]
if (KEY_NAME in params)
else path)
link = 'service_info?service_path=' + escape(path)
element = html_link_with_text(text, link = link)
if KEY_NAME in params:
element += html_inline_button_with_text('Configure',
action = 'get_service_passphrase',
param1 = 'service_name',
value1 = escape(params[KEY_NAME]),
param2 = 'service_path',
value2 = escape(path))
ol.append(html_section_with_html_element(element))
return webpage_with_orderedlist(ol)
@cherrypy.expose
def service_info(self, service_path):
service = pyconnman.ConnService(service_path)
return webpage_with_body(service)
@cherrypy.expose
def get_service_passphrase(self, service_name, service_path):
text = '''Passphrase for %s:''' % (service_name)
element = html_paragraph_with_text(text)
element += html_button_with_text('Connect',
action = 'connect_service',
inputParam = 'service_passphrase',
hiddenParam1 = 'service_path',
hiddenValue1 = escape(service_path),
hiddenParam2 = 'service_name',
hiddenValue2 = escape(service_name))
return webpage_with_body(element)
@cherrypy.expose
def connect_service(self, service_path, service_name, service_passphrase):
agent_args = [AGENT_APP,
'--' + agent.ARG_AGENT_PATH, AGENT_PATH,
'--' + agent.ARG_SERV_PATH, service_path,
'--' + agent.ARG_SERV_NAME, service_name,
'--' + agent.ARG_PASSPHRASE, service_passphrase]
agent_app = Popen(agent_args, preexec_fn = setsid)
sleep(SLEEP_TIME)
try:
service = pyconnman.ConnService(service_path)
service.connect()
except DBusException:
print 'Error: ', exc_info()
killpg(getpgid(agent_app.pid), SIGTERM)
raise cherrypy.HTTPRedirect('list_connected_services')
@cherrypy.expose
def list_connected_services(self):
manager = pyconnman.ConnManager()
services = manager.get_services()
ol = []
for (path, params) in services:
serv = pyconnman.ConnService(path)
if serv.get_property(KEY_STATE) in CONNECTED_SERVICE_STATE:
text = (params[KEY_NAME]
if (KEY_NAME in params)
else path)
link = 'service_info?service_path=' + escape(path)
element = html_link_with_text(text, link = link)
ol.append(html_section_with_html_element(element))
return webpage_with_orderedlist(ol)
# Main
if __name__ == '__main__':
parser = ArgumentParser(description = 'cherrypy + pyconnman')
parser.add_argument('--' + ARGUMENT_HOST,
help = 'network interface (0.0.0.0 for all interfaces)')
parser.add_argument('--' + ARGUMENT_PORT,
type = int,
help = 'TCP port')
args = vars(parser.parse_args())
host = args[ARGUMENT_HOST]
if host:
cherrypy.config.update({'server.socket_host': host})
port = args[ARGUMENT_PORT]
if port:
cherrypy.config.update({'server.socket_port': port})
cherrypy.quickstart(Server())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment