Rebuilt the network view in Svelte
This commit is contained in:
David Carley
@@ -1,30 +1,3 @@
|
||||
################################################################################
|
||||
# #
|
||||
# This file is part of the Buildbotics firmware. #
|
||||
# #
|
||||
# Copyright (c) 2015 - 2018, Buildbotics LLC #
|
||||
# All rights reserved. #
|
||||
# #
|
||||
# This file ("the software") is free software: you can redistribute it #
|
||||
# and/or modify it under the terms of the GNU General Public License, #
|
||||
# version 2 as published by the Free Software Foundation. You should #
|
||||
# have received a copy of the GNU General Public License, version 2 #
|
||||
# along with the software. If not, see <http://www.gnu.org/licenses/>. #
|
||||
# #
|
||||
# The software is distributed in the hope that it will be useful, but #
|
||||
# WITHOUT ANY WARRANTY; without even the implied warranty of #
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU #
|
||||
# Lesser General Public License for more details. #
|
||||
# #
|
||||
# You should have received a copy of the GNU Lesser General Public #
|
||||
# License along with the software. If not, see #
|
||||
# <http://www.gnu.org/licenses/>. #
|
||||
# #
|
||||
# For information regarding this software email: #
|
||||
# "Joseph Coffland" <joseph@buildbotics.com> #
|
||||
# #
|
||||
################################################################################
|
||||
|
||||
import os
|
||||
import json
|
||||
import tornado
|
||||
@@ -36,50 +9,26 @@ from tornado.web import HTTPError
|
||||
from tornado import gen
|
||||
|
||||
import bbctrl
|
||||
|
||||
import iw_parse
|
||||
|
||||
|
||||
def call_get_output(cmd):
|
||||
p = subprocess.Popen(cmd, stdout = subprocess.PIPE)
|
||||
p = subprocess.Popen(cmd, stdout=subprocess.PIPE)
|
||||
s = p.communicate()[0].decode('utf-8')
|
||||
if p.returncode: raise HTTPError(400, 'Command failed')
|
||||
if p.returncode:
|
||||
raise HTTPError(400, 'Command failed')
|
||||
return s
|
||||
|
||||
|
||||
def get_username():
|
||||
return call_get_output(['getent', 'passwd', '1001']).split(':')[0]
|
||||
|
||||
|
||||
def set_username(username):
|
||||
if subprocess.call(['usermod', '-l', username, get_username()]):
|
||||
raise HTTPError(400, 'Failed to set username to "%s"' % username)
|
||||
|
||||
|
||||
def check_password(password):
|
||||
# Get current password
|
||||
s = call_get_output(['getent', 'shadow', get_username()])
|
||||
current = s.split(':')[1].split('$')
|
||||
|
||||
# Check password type
|
||||
if len(current) < 2 or current[1] != '1':
|
||||
raise HTTPError(401, "Password invalid")
|
||||
|
||||
# Check current password
|
||||
cmd = ['openssl', 'passwd', '-salt', current[2], '-1', password]
|
||||
s = call_get_output(cmd).strip()
|
||||
|
||||
if s.split('$') != current: raise HTTPError(401, 'Wrong password')
|
||||
|
||||
|
||||
|
||||
class RebootHandler(bbctrl.APIHandler):
|
||||
def put_ok(self):
|
||||
self.get_ctrl().lcd.goodbye('Rebooting...')
|
||||
subprocess.Popen('reboot')
|
||||
|
||||
|
||||
|
||||
class ShutdownHandler(bbctrl.APIHandler):
|
||||
def put_ok(self):
|
||||
subprocess.Popen(['shutdown','-h','now'])
|
||||
subprocess.Popen(['shutdown', '-h', 'now'])
|
||||
|
||||
|
||||
class LogHandler(bbctrl.RequestHandler):
|
||||
@@ -87,7 +36,6 @@ class LogHandler(bbctrl.RequestHandler):
|
||||
with open(self.get_ctrl().log.get_path(), 'r') as f:
|
||||
self.write(f.read())
|
||||
|
||||
|
||||
def set_default_headers(self):
|
||||
fmt = socket.gethostname() + '-%Y%m%d.log'
|
||||
filename = datetime.date.today().strftime(fmt)
|
||||
@@ -102,14 +50,16 @@ class MessageAckHandler(bbctrl.APIHandler):
|
||||
|
||||
class BugReportHandler(bbctrl.RequestHandler):
|
||||
def get(self):
|
||||
import tarfile, io
|
||||
import tarfile
|
||||
import io
|
||||
|
||||
buf = io.BytesIO()
|
||||
tar = tarfile.open(mode = 'w:bz2', fileobj = buf)
|
||||
tar = tarfile.open(mode='w:bz2', fileobj=buf)
|
||||
|
||||
def check_add(path, arcname = None):
|
||||
def check_add(path, arcname=None):
|
||||
if os.path.isfile(path):
|
||||
if arcname is None: arcname = path
|
||||
if arcname is None:
|
||||
arcname = path
|
||||
tar.add(path, self.basename + '/' + arcname)
|
||||
|
||||
def check_add_basename(path):
|
||||
@@ -128,7 +78,6 @@ class BugReportHandler(bbctrl.RequestHandler):
|
||||
|
||||
self.write(buf.getvalue())
|
||||
|
||||
|
||||
def set_default_headers(self):
|
||||
fmt = socket.gethostname() + '-%Y%m%d-%H%M%S'
|
||||
self.basename = datetime.datetime.now().strftime(fmt)
|
||||
@@ -138,8 +87,6 @@ class BugReportHandler(bbctrl.RequestHandler):
|
||||
|
||||
|
||||
class HostnameHandler(bbctrl.APIHandler):
|
||||
def get(self): self.write_json(socket.gethostname())
|
||||
|
||||
def put(self):
|
||||
if self.get_ctrl().args.demo:
|
||||
raise HTTPError(400, 'Cannot set hostname in demo mode')
|
||||
@@ -153,77 +100,59 @@ class HostnameHandler(bbctrl.APIHandler):
|
||||
raise HTTPError(400, 'Failed to set hostname')
|
||||
|
||||
|
||||
class WifiHandler(bbctrl.APIHandler):
|
||||
class NetworkHandler(bbctrl.APIHandler):
|
||||
def get(self):
|
||||
data = {'ssid': '', 'channel': 0}
|
||||
try:
|
||||
data = json.loads(call_get_output(['config-wifi', '-j']))
|
||||
except: pass
|
||||
self.write_json(data)
|
||||
ipAddresses = call_get_output(['hostname', '-I']).split()
|
||||
except:
|
||||
ipAddresses = ""
|
||||
|
||||
hostname = socket.gethostname()
|
||||
|
||||
try:
|
||||
wifi = json.loads(call_get_output(['config-wifi', '-j']))
|
||||
except:
|
||||
wifi = {'enabled': False}
|
||||
|
||||
try:
|
||||
lines = iw_parse.call_iwlist().decode("utf-8").split("\n")
|
||||
wifi['networks'] = iw_parse.get_parsed_cells(lines)
|
||||
except:
|
||||
wifi['networks'] = []
|
||||
|
||||
self.write_json({
|
||||
'ipAddresses': ipAddresses,
|
||||
'hostname': hostname,
|
||||
'wifi': wifi
|
||||
})
|
||||
|
||||
def put(self):
|
||||
if self.get_ctrl().args.demo:
|
||||
raise HTTPError(400, 'Cannot configure WiFi in demo mode')
|
||||
|
||||
if 'mode' in self.json:
|
||||
cmd = ['config-wifi', '-r']
|
||||
mode = self.json['mode']
|
||||
if not 'wifi' in self.json:
|
||||
raise HTTPError(400, 'Payload is missing wifi config information')
|
||||
|
||||
if mode == 'disabled': cmd += ['-d']
|
||||
elif 'ssid' in self.json:
|
||||
cmd += ['-s', self.json['ssid']]
|
||||
wifi = self.json['wifi']
|
||||
|
||||
if mode == 'ap':
|
||||
cmd += ['-a']
|
||||
if 'channel' in self.json:
|
||||
cmd += ['-c', self.json['channel']]
|
||||
cmd = ['config-wifi', '-r']
|
||||
|
||||
if 'pass' in self.json:
|
||||
cmd += ['-p', self.json['pass']]
|
||||
if not wifi['enabled']:
|
||||
cmd += ['-d']
|
||||
else:
|
||||
if 'ssid' in wifi:
|
||||
cmd += ['-s', wifi['ssid']]
|
||||
|
||||
if subprocess.call(cmd) == 0:
|
||||
self.write_json('ok')
|
||||
return
|
||||
if 'password' in wifi:
|
||||
cmd += ['-p', wifi['password']]
|
||||
|
||||
if subprocess.call(cmd) == 0:
|
||||
self.write_json('ok')
|
||||
return
|
||||
|
||||
raise HTTPError(400, 'Failed to configure wifi')
|
||||
|
||||
|
||||
class UsernameHandler(bbctrl.APIHandler):
|
||||
def get(self): self.write_json(get_username())
|
||||
|
||||
|
||||
def put_ok(self):
|
||||
if self.get_ctrl().args.demo:
|
||||
raise HTTPError(400, 'Cannot set username in demo mode')
|
||||
|
||||
if 'username' in self.json: set_username(self.json['username'])
|
||||
else: raise HTTPError(400, 'Missing "username"')
|
||||
|
||||
|
||||
class PasswordHandler(bbctrl.APIHandler):
|
||||
def put(self):
|
||||
if self.get_ctrl().args.demo:
|
||||
raise HTTPError(400, 'Cannot set password in demo mode')
|
||||
|
||||
if 'current' in self.json and 'password' in self.json:
|
||||
check_password(self.json['current'])
|
||||
|
||||
# Set password
|
||||
s = '%s:%s' % (get_username(), self.json['password'])
|
||||
s = s.encode('utf-8')
|
||||
|
||||
p = subprocess.Popen(['chpasswd', '-c', 'MD5'],
|
||||
stdin = subprocess.PIPE)
|
||||
p.communicate(input = s)
|
||||
|
||||
if p.returncode == 0:
|
||||
self.write_json('ok')
|
||||
return
|
||||
|
||||
raise HTTPError(401, 'Failed to set password')
|
||||
|
||||
|
||||
class ConfigLoadHandler(bbctrl.APIHandler):
|
||||
def get(self):
|
||||
self.write_json(self.get_ctrl().config.load())
|
||||
@@ -238,7 +167,7 @@ class ConfigDownloadHandler(bbctrl.APIHandler):
|
||||
'attachment; filename="%s"' % filename)
|
||||
|
||||
def get(self):
|
||||
self.write_json(self.get_ctrl().config.load(), pretty = True)
|
||||
self.write_json(self.get_ctrl().config.load(), pretty=True)
|
||||
|
||||
|
||||
class ConfigSaveHandler(bbctrl.APIHandler):
|
||||
@@ -252,14 +181,14 @@ class ConfigResetHandler(bbctrl.APIHandler):
|
||||
class FirmwareUpdateHandler(bbctrl.APIHandler):
|
||||
def prepare(self): pass
|
||||
|
||||
|
||||
def put_ok(self):
|
||||
if not 'firmware' in self.request.files:
|
||||
raise HTTPError(401, 'Missing "firmware"')
|
||||
|
||||
firmware = self.request.files['firmware'][0]
|
||||
|
||||
if not os.path.exists('firmware'): os.mkdir('firmware')
|
||||
if not os.path.exists('firmware'):
|
||||
os.mkdir('firmware')
|
||||
|
||||
with open('firmware/update.tar.bz2', 'wb') as f:
|
||||
f.write(firmware['body'])
|
||||
@@ -284,20 +213,23 @@ class PathHandler(bbctrl.APIHandler):
|
||||
future = preplanner.get_plan(filename)
|
||||
|
||||
try:
|
||||
delta = datetime.timedelta(seconds = 1)
|
||||
delta = datetime.timedelta(seconds=1)
|
||||
data = yield gen.with_timeout(delta, future)
|
||||
|
||||
except gen.TimeoutError:
|
||||
progress = preplanner.get_plan_progress(filename)
|
||||
self.write_json(dict(progress = progress))
|
||||
self.write_json(dict(progress=progress))
|
||||
return
|
||||
|
||||
try:
|
||||
if data is None: return
|
||||
if data is None:
|
||||
return
|
||||
meta, positions, speeds = data
|
||||
|
||||
if dataType == '/positions': data = positions
|
||||
elif dataType == '/speeds': data = speeds
|
||||
if dataType == '/positions':
|
||||
data = positions
|
||||
elif dataType == '/speeds':
|
||||
data = speeds
|
||||
else:
|
||||
self.get_ctrl().state.set_bounds(meta['bounds'])
|
||||
self.write_json(meta)
|
||||
@@ -316,12 +248,14 @@ class PathHandler(bbctrl.APIHandler):
|
||||
self.write(chunk)
|
||||
yield self.flush()
|
||||
|
||||
except tornado.iostream.StreamClosedError as e: pass
|
||||
except tornado.iostream.StreamClosedError as e:
|
||||
pass
|
||||
|
||||
|
||||
class HomeHandler(bbctrl.APIHandler):
|
||||
def put_ok(self, axis, action, *args):
|
||||
if axis is not None: axis = ord(axis[1:2].lower())
|
||||
if axis is not None:
|
||||
axis = ord(axis[1:2].lower())
|
||||
|
||||
if action == '/set':
|
||||
if not 'position' in self.json:
|
||||
@@ -329,8 +263,10 @@ class HomeHandler(bbctrl.APIHandler):
|
||||
|
||||
self.get_ctrl().mach.home(axis, self.json['position'])
|
||||
|
||||
elif action == '/clear': self.get_ctrl().mach.unhome(axis)
|
||||
else: self.get_ctrl().mach.home(axis)
|
||||
elif action == '/clear':
|
||||
self.get_ctrl().mach.unhome(axis)
|
||||
else:
|
||||
self.get_ctrl().mach.home(axis)
|
||||
|
||||
|
||||
class StartHandler(bbctrl.APIHandler):
|
||||
@@ -386,7 +322,7 @@ class ModbusReadHandler(bbctrl.APIHandler):
|
||||
class ModbusWriteHandler(bbctrl.APIHandler):
|
||||
def put_ok(self):
|
||||
self.get_ctrl().mach.modbus_write(int(self.json['address']),
|
||||
int(self.json['value']))
|
||||
int(self.json['value']))
|
||||
|
||||
|
||||
class JogHandler(bbctrl.APIHandler):
|
||||
@@ -402,7 +338,8 @@ class JogHandler(bbctrl.APIHandler):
|
||||
last = self.app.last_jog.get(id, 0)
|
||||
self.app.last_jog[id] = ts
|
||||
|
||||
if ts < last: return # Out of order
|
||||
if ts < last:
|
||||
return # Out of order
|
||||
|
||||
self.get_ctrl().mach.jog(self.json)
|
||||
|
||||
@@ -413,17 +350,14 @@ class ClientConnection(object):
|
||||
self.app = app
|
||||
self.count = 0
|
||||
|
||||
|
||||
def heartbeat(self):
|
||||
self.timer = self.app.ioloop.call_later(3, self.heartbeat)
|
||||
self.send({'heartbeat': self.count})
|
||||
self.count += 1
|
||||
|
||||
|
||||
def send(self, msg): raise HTTPError(400, 'Not implemented')
|
||||
|
||||
|
||||
def on_open(self, id = None):
|
||||
def on_open(self, id=None):
|
||||
self.ctrl = self.app.get_ctrl(id)
|
||||
|
||||
self.ctrl.state.add_listener(self.send)
|
||||
@@ -432,7 +366,6 @@ class ClientConnection(object):
|
||||
self.heartbeat()
|
||||
self.app.opened(self.ctrl)
|
||||
|
||||
|
||||
def on_close(self):
|
||||
self.app.ioloop.remove_timeout(self.timer)
|
||||
self.ctrl.state.remove_listener(self.send)
|
||||
@@ -440,7 +373,6 @@ class ClientConnection(object):
|
||||
self.is_open = False
|
||||
self.app.closed(self.ctrl)
|
||||
|
||||
|
||||
def on_message(self, data):
|
||||
self.ctrl.mach.mdi(data)
|
||||
|
||||
@@ -465,23 +397,24 @@ class SockJSConnection(ClientConnection, sockjs.tornado.SockJSConnection):
|
||||
ClientConnection.__init__(self, session.server.app)
|
||||
sockjs.tornado.SockJSConnection.__init__(self, session)
|
||||
|
||||
|
||||
def send(self, msg):
|
||||
try:
|
||||
sockjs.tornado.SockJSConnection.send(self, msg)
|
||||
except:
|
||||
self.close()
|
||||
|
||||
|
||||
def on_open(self, info):
|
||||
cookie = info.get_cookie('client-id')
|
||||
if cookie is None: self.send(dict(sid = '')) # Trigger client reset
|
||||
if cookie is None:
|
||||
self.send(dict(sid='')) # Trigger client reset
|
||||
else:
|
||||
id = cookie.value
|
||||
|
||||
ip = info.ip
|
||||
if 'X-Real-IP' in info.headers: ip = info.headers['X-Real-IP']
|
||||
self.app.get_ctrl(id).log.get('Web').info('Connection from %s' % ip)
|
||||
if 'X-Real-IP' in info.headers:
|
||||
ip = info.headers['X-Real-IP']
|
||||
self.app.get_ctrl(id).log.get(
|
||||
'Web').info('Connection from %s' % ip)
|
||||
super().on_open(id)
|
||||
|
||||
|
||||
@@ -499,10 +432,13 @@ class Web(tornado.web.Application):
|
||||
|
||||
# Init camera
|
||||
if not args.disable_camera:
|
||||
if self.args.demo: log = bbctrl.log.Log(args, ioloop, 'camera.log')
|
||||
else: log = self.get_ctrl().log
|
||||
if self.args.demo:
|
||||
log = bbctrl.log.Log(args, ioloop, 'camera.log')
|
||||
else:
|
||||
log = self.get_ctrl().log
|
||||
self.camera = bbctrl.Camera(ioloop, args, log)
|
||||
else: self.camera = None
|
||||
else:
|
||||
self.camera = None
|
||||
|
||||
# Init controller
|
||||
if not self.args.demo:
|
||||
@@ -517,9 +453,7 @@ class Web(tornado.web.Application):
|
||||
(r'/api/reboot', RebootHandler),
|
||||
(r'/api/shutdown', ShutdownHandler),
|
||||
(r'/api/hostname', HostnameHandler),
|
||||
(r'/api/wifi', WifiHandler),
|
||||
(r'/api/remote/username', UsernameHandler),
|
||||
(r'/api/remote/password', PasswordHandler),
|
||||
(r'/api/network', NetworkHandler),
|
||||
(r'/api/config/load', ConfigLoadHandler),
|
||||
(r'/api/config/download', ConfigDownloadHandler),
|
||||
(r'/api/config/save', ConfigSaveHandler),
|
||||
@@ -547,7 +481,7 @@ class Web(tornado.web.Application):
|
||||
(r'/(.*)', StaticFileHandler,
|
||||
{'path': bbctrl.get_resource('http/'),
|
||||
'default_filename': 'index.html'}),
|
||||
]
|
||||
]
|
||||
|
||||
router = sockjs.tornado.SockJSRouter(SockJSConnection, '/sockjs')
|
||||
router.app = self
|
||||
@@ -555,7 +489,7 @@ class Web(tornado.web.Application):
|
||||
tornado.web.Application.__init__(self, router.urls + handlers)
|
||||
|
||||
try:
|
||||
self.listen(args.port, address = args.addr)
|
||||
self.listen(args.port, address=args.addr)
|
||||
|
||||
except Exception as e:
|
||||
raise Exception('Failed to bind %s:%d: %s' % (
|
||||
@@ -563,33 +497,32 @@ class Web(tornado.web.Application):
|
||||
|
||||
print('Listening on http://%s:%d/' % (args.addr, args.port))
|
||||
|
||||
|
||||
def opened(self, ctrl): ctrl.clear_timeout()
|
||||
|
||||
|
||||
def closed(self, ctrl):
|
||||
# Time out clients in demo mode
|
||||
if self.args.demo: ctrl.set_timeout(self._reap_ctrl, ctrl)
|
||||
|
||||
if self.args.demo:
|
||||
ctrl.set_timeout(self._reap_ctrl, ctrl)
|
||||
|
||||
def _reap_ctrl(self, ctrl):
|
||||
ctrl.close()
|
||||
del self.ctrls[ctrl.id]
|
||||
|
||||
|
||||
def get_ctrl(self, id = None):
|
||||
if not id or not self.args.demo: id = ''
|
||||
def get_ctrl(self, id=None):
|
||||
if not id or not self.args.demo:
|
||||
id = ''
|
||||
|
||||
if not id in self.ctrls:
|
||||
ctrl = bbctrl.Ctrl(self.args, self.ioloop, id)
|
||||
self.ctrls[id] = ctrl
|
||||
|
||||
else: ctrl = self.ctrls[id]
|
||||
else:
|
||||
ctrl = self.ctrls[id]
|
||||
|
||||
return ctrl
|
||||
|
||||
|
||||
# Override default logger
|
||||
|
||||
def log_request(self, handler):
|
||||
log = self.get_ctrl(handler.get_cookie('client-id')).log.get('Web')
|
||||
log.info("%d %s", handler.get_status(), handler._request_summary())
|
||||
|
||||
80
src/py/iw_parse/README.md
Normal file
80
src/py/iw_parse/README.md
Normal file
@@ -0,0 +1,80 @@
|
||||
iw_parse
|
||||
========
|
||||
|
||||
Parse the output of iwlist scan to get the name, address, quality, channel, and encryption type of all networks broadcasting within your Wireless NIC's reach.
|
||||
|
||||
Dependencies
|
||||
------------
|
||||
|
||||
* [pip](http://www.pip-installer.org/en/latest/installing.html "pip installation guide") - If you don't have pip installed, followed the link.
|
||||
|
||||
|
||||
Installation
|
||||
------------
|
||||
|
||||
```bash
|
||||
pip install iw_parse
|
||||
```
|
||||
|
||||
Usage
|
||||
-----
|
||||
|
||||
```bash
|
||||
iwlist <INTERFACE_NAME> scan | iw_parse
|
||||
```
|
||||
|
||||
Replace `<INTERFACE_NAME>` with the system name for your wireless NIC. It is usually something like `wlan0`. The command `iwconfig` will list all of your network interfaces.
|
||||
|
||||
Example:
|
||||
|
||||
```bash
|
||||
iwlist wlan0 scan | iw_parse
|
||||
```
|
||||
|
||||
The result should look something like:
|
||||
|
||||
```
|
||||
Name Address Quality Channel Encryption
|
||||
wireless1 20:AA:4B:34:2C:F5 100 % 11 WEP
|
||||
wireless2 00:26:F2:1E:FC:03 84 % 1 WPA v.1
|
||||
wireless3 00:1D:D3:6A:3C:60 66 % 6 WEP
|
||||
wireless4 20:10:7A:E5:02:98 64 % 1 WEP
|
||||
wireless5 CC:A4:62:B7:D2:B0 54 % 8 WPA v.1
|
||||
wireless6 30:46:9A:53:3C:76 47 % 11 WPA v.1
|
||||
wireless7 A0:21:B7:5F:84:B0 44 % 11 WEP
|
||||
wireless8 04:A1:51:18:E8:E0 41 % 6 WPA v.1
|
||||
```
|
||||
|
||||
Example from Python shell:
|
||||
|
||||
```python
|
||||
>>> import iw_parse
|
||||
>>> networks = iw_parse.get_interfaces(interface='wlan0')
|
||||
>>> print networks
|
||||
[{'Address': 'F8:1E:DF:F9:B0:0B',
|
||||
'Channel': '3',
|
||||
'Encryption': 'WEP',
|
||||
'Name': 'Francis',
|
||||
'Bit Rates': '144 Mb/s',
|
||||
'Signal Level': '42',
|
||||
'Name': 'Francis',
|
||||
'Quality': '100'},
|
||||
{'Address': '86:1B:5E:33:17:D4',
|
||||
'Channel': '6',
|
||||
'Encryption': 'Open',
|
||||
'Bit Rates': '54 Mb/s',
|
||||
'Signal Level': '72',
|
||||
'Name': 'optimumwifi',
|
||||
'Quality': '100'},
|
||||
...
|
||||
```
|
||||
|
||||
Acknowledgements
|
||||
----------------
|
||||
|
||||
* The vast majority of iw_parse was written by Hugo Chargois.
|
||||
|
||||
License
|
||||
-------
|
||||
|
||||
iw_parse is free--as in BSD. Hack your heart out, hackers.
|
||||
1
src/py/iw_parse/__init__.py
Normal file
1
src/py/iw_parse/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
from .iw_parse import *
|
||||
25
src/py/iw_parse/iw_parse
Normal file
25
src/py/iw_parse/iw_parse
Normal file
@@ -0,0 +1,25 @@
|
||||
#! /usr/bin/env python
|
||||
|
||||
import sys
|
||||
|
||||
from iw_parse import get_parsed_cells, print_cells
|
||||
|
||||
def main():
|
||||
""" Pretty prints the output of iwlist scan into a table. """
|
||||
parsed_cells = get_parsed_cells(sys.stdin)
|
||||
|
||||
# You can choose which columns to display here, and most importantly
|
||||
# in what order. Of course, they must exist as keys in the dict rules.
|
||||
columns = [
|
||||
"Name",
|
||||
"Address",
|
||||
"Quality",
|
||||
"Channel",
|
||||
"Signal Level",
|
||||
"Encryption"
|
||||
]
|
||||
|
||||
print_cells(parsed_cells, columns)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
337
src/py/iw_parse/iw_parse.py
Normal file
337
src/py/iw_parse/iw_parse.py
Normal file
@@ -0,0 +1,337 @@
|
||||
#! /usr/bin/env python
|
||||
|
||||
# Hugo Chargois - 17 jan. 2010 - v.0.1
|
||||
# Parses the output of iwlist scan into a table
|
||||
|
||||
# You can add or change the functions to parse the properties
|
||||
# of each AP (cell) below. They take one argument, the bunch of text
|
||||
# describing one cell in iwlist scan and return a property of that cell.
|
||||
|
||||
import re
|
||||
import subprocess
|
||||
|
||||
VERSION_RGX = re.compile("version\s+\d+", re.IGNORECASE)
|
||||
|
||||
def get_name(cell):
|
||||
""" Gets the name / essid of a network / cell.
|
||||
@param string cell
|
||||
A network / cell from iwlist scan.
|
||||
|
||||
@return string
|
||||
The name / essid of the network.
|
||||
"""
|
||||
|
||||
essid = matching_line(cell, "ESSID:")
|
||||
if not essid:
|
||||
return ""
|
||||
return essid[1:-1]
|
||||
|
||||
def get_quality(cell):
|
||||
""" Gets the quality of a network / cell.
|
||||
@param string cell
|
||||
A network / cell from iwlist scan.
|
||||
|
||||
@return string
|
||||
The quality of the network.
|
||||
"""
|
||||
|
||||
quality = matching_line(cell, "Quality=")
|
||||
if quality is None:
|
||||
return ""
|
||||
quality = quality.split()[0].split("/")
|
||||
quality = matching_line(cell, "Quality=").split()[0].split("/")
|
||||
return str(int(round(float(quality[0]) / float(quality[1]) * 100)))
|
||||
|
||||
def get_signal_level(cell):
|
||||
""" Gets the signal level of a network / cell.
|
||||
@param string cell
|
||||
A network / cell from iwlist scan.
|
||||
|
||||
@return string
|
||||
The signal level of the network.
|
||||
"""
|
||||
|
||||
signal = matching_line(cell, "Signal level=")
|
||||
if signal is None:
|
||||
return ""
|
||||
signal = signal.split("=")[1].split("/")
|
||||
if len(signal) == 2:
|
||||
return str(int(round(float(signal[0]) / float(signal[1]) * 100)))
|
||||
elif len(signal) == 1:
|
||||
return signal[0].split(' ')[0]
|
||||
else:
|
||||
return ""
|
||||
|
||||
def get_noise_level(cell):
|
||||
""" Gets the noise level of a network / cell.
|
||||
@param string cell
|
||||
A network / cell from iwlist scan.
|
||||
|
||||
@return string
|
||||
The noise level of the network.
|
||||
"""
|
||||
|
||||
noise = matching_line(cell, "Noise level=")
|
||||
if noise is None:
|
||||
return ""
|
||||
noise = noise.split("=")[1]
|
||||
return noise.split(' ')[0]
|
||||
|
||||
def get_channel(cell):
|
||||
""" Gets the channel of a network / cell.
|
||||
@param string cell
|
||||
A network / cell from iwlist scan.
|
||||
|
||||
@return string
|
||||
The channel of the network.
|
||||
"""
|
||||
|
||||
channel = matching_line(cell, "Channel:")
|
||||
if channel:
|
||||
return channel
|
||||
frequency = matching_line(cell, "Frequency:")
|
||||
channel = re.sub(r".*\(Channel\s(\d{1,3})\).*", r"\1", frequency)
|
||||
return channel
|
||||
|
||||
def get_frequency(cell):
|
||||
""" Gets the frequency of a network / cell.
|
||||
@param string cell
|
||||
A network / cell from iwlist scan.
|
||||
|
||||
@return string
|
||||
The frequency of the network.
|
||||
"""
|
||||
|
||||
frequency = matching_line(cell, "Frequency:")
|
||||
if frequency is None:
|
||||
return ""
|
||||
return frequency.split()[0]
|
||||
|
||||
def get_encryption(cell, emit_version=False):
|
||||
""" Gets the encryption type of a network / cell.
|
||||
@param string cell
|
||||
A network / cell from iwlist scan.
|
||||
|
||||
@return string
|
||||
The encryption type of the network.
|
||||
"""
|
||||
|
||||
enc = ""
|
||||
if matching_line(cell, "Encryption key:") == "off":
|
||||
enc = "Open"
|
||||
else:
|
||||
for line in cell:
|
||||
matching = match(line,"IE:")
|
||||
if matching == None:
|
||||
continue
|
||||
|
||||
wpa = match(matching,"WPA")
|
||||
if wpa == None:
|
||||
continue
|
||||
|
||||
version_matches = VERSION_RGX.search(wpa)
|
||||
if len(version_matches.regs) == 1:
|
||||
version = version_matches \
|
||||
.group(0) \
|
||||
.lower() \
|
||||
.replace("version", "") \
|
||||
.strip()
|
||||
wpa = wpa.replace(version_matches.group(0), "").strip()
|
||||
if wpa == "":
|
||||
wpa = "WPA"
|
||||
if emit_version:
|
||||
enc = "{0} v.{1}".format(wpa, version)
|
||||
else:
|
||||
enc = wpa
|
||||
if wpa == "WPA2":
|
||||
return enc
|
||||
else:
|
||||
enc = wpa
|
||||
if enc == "":
|
||||
enc = "WEP"
|
||||
return enc
|
||||
|
||||
def get_mode(cell):
|
||||
""" Gets the mode of a network / cell.
|
||||
@param string cell
|
||||
A network / cell from iwlist scan.
|
||||
|
||||
@return string
|
||||
The IEEE 802.11 mode of the network.
|
||||
"""
|
||||
|
||||
mode = matching_line(cell, "Extra:ieee_mode=")
|
||||
if mode is None:
|
||||
return ""
|
||||
return mode
|
||||
|
||||
def get_address(cell):
|
||||
""" Gets the address of a network / cell.
|
||||
@param string cell
|
||||
A network / cell from iwlist scan.
|
||||
|
||||
@return string
|
||||
The address of the network.
|
||||
"""
|
||||
|
||||
return matching_line(cell, "Address: ")
|
||||
|
||||
def get_bit_rates(cell):
|
||||
""" Gets the bit rate of a network / cell.
|
||||
@param string cell
|
||||
A network / cell from iwlist scan.
|
||||
|
||||
@return string
|
||||
The bit rate of the network.
|
||||
"""
|
||||
|
||||
return matching_line(cell, "Bit Rates:")
|
||||
|
||||
# Here you can choose the way of sorting the table. sortby should be a key of
|
||||
# the dictionary rules.
|
||||
|
||||
def sort_cells(cells):
|
||||
sortby = "Quality"
|
||||
reverse = True
|
||||
cells.sort(key=lambda el:el[sortby], reverse=reverse)
|
||||
|
||||
|
||||
# Below here goes the boring stuff. You shouldn't have to edit anything below
|
||||
# this point
|
||||
|
||||
def matching_line(lines, keyword):
|
||||
""" Returns the first matching line in a list of lines.
|
||||
@see match()
|
||||
"""
|
||||
for line in lines:
|
||||
matching = match(line,keyword)
|
||||
if matching != None:
|
||||
return matching
|
||||
return None
|
||||
|
||||
def match(line, keyword):
|
||||
""" If the first part of line (modulo blanks) matches keyword,
|
||||
returns the end of that line. Otherwise checks if keyword is
|
||||
anywhere in the line and returns that section, else returns None"""
|
||||
|
||||
line = line.lstrip()
|
||||
length = len(keyword)
|
||||
if line[:length] == keyword:
|
||||
return line[length:]
|
||||
else:
|
||||
if keyword in line:
|
||||
return line[line.index(keyword):]
|
||||
else:
|
||||
return None
|
||||
|
||||
def parse_cell(cell, rules):
|
||||
""" Applies the rules to the bunch of text describing a cell.
|
||||
@param string cell
|
||||
A network / cell from iwlist scan.
|
||||
@param dictionary rules
|
||||
A dictionary of parse rules.
|
||||
|
||||
@return dictionary
|
||||
parsed networks. """
|
||||
|
||||
parsed_cell = {}
|
||||
for key in rules:
|
||||
rule = rules[key]
|
||||
parsed_cell.update({key: rule(cell)})
|
||||
return parsed_cell
|
||||
|
||||
def print_table(table):
|
||||
# Functional black magic.
|
||||
widths = list(map(max, map(lambda l: map(len, l), zip(*table))))
|
||||
|
||||
justified_table = []
|
||||
for line in table:
|
||||
justified_line = []
|
||||
for i, el in enumerate(line):
|
||||
justified_line.append(el.ljust(widths[i] + 2))
|
||||
justified_table.append(justified_line)
|
||||
|
||||
for line in justified_table:
|
||||
print("\t".join(line))
|
||||
|
||||
def print_cells(cells, columns):
|
||||
table = [columns]
|
||||
for cell in cells:
|
||||
cell_properties = []
|
||||
for column in columns:
|
||||
if column == 'Quality':
|
||||
# make print nicer
|
||||
cell[column] = cell[column].rjust(3) + " %"
|
||||
cell_properties.append(cell[column])
|
||||
table.append(cell_properties)
|
||||
print_table(table)
|
||||
|
||||
def get_parsed_cells(iw_data, rules=None):
|
||||
""" Parses iwlist output into a list of networks.
|
||||
@param list iw_data
|
||||
Output from iwlist scan.
|
||||
A list of strings.
|
||||
|
||||
@return list
|
||||
properties: Name, Address, Quality, Channel, Frequency, Encryption, Signal Level, Noise Level, Bit Rates, Mode.
|
||||
"""
|
||||
|
||||
# Here's a dictionary of rules that will be applied to the description
|
||||
# of each cell. The key will be the name of the column in the table.
|
||||
# The value is a function defined above.
|
||||
rules = rules or {
|
||||
"Name": get_name,
|
||||
"Quality": get_quality,
|
||||
"Channel": get_channel,
|
||||
"Frequency": get_frequency,
|
||||
"Encryption": get_encryption,
|
||||
"Address": get_address,
|
||||
"Signal Level": get_signal_level,
|
||||
"Noise Level": get_noise_level,
|
||||
"Bit Rates": get_bit_rates,
|
||||
"Mode": get_mode,
|
||||
}
|
||||
|
||||
cells = [[]]
|
||||
parsed_cells = []
|
||||
|
||||
for line in iw_data:
|
||||
cell_line = match(line, "Cell ")
|
||||
if cell_line != None:
|
||||
cells.append([])
|
||||
line = cell_line[-27:]
|
||||
cells[-1].append(line.rstrip())
|
||||
|
||||
cells = cells[1:]
|
||||
|
||||
for cell in cells:
|
||||
parsed_cells.append(parse_cell(cell, rules))
|
||||
|
||||
sort_cells(parsed_cells)
|
||||
return parsed_cells
|
||||
|
||||
def call_iwlist(interface='wlan0'):
|
||||
""" Get iwlist output via subprocess
|
||||
@param string interface
|
||||
interface to scan
|
||||
default is wlan0
|
||||
|
||||
@return string
|
||||
properties: iwlist output
|
||||
"""
|
||||
return subprocess.check_output(['iwlist', interface, 'scanning'])
|
||||
|
||||
def get_interfaces(interface="wlan0"):
|
||||
""" Get parsed iwlist output
|
||||
@param string interface
|
||||
interface to scan
|
||||
default is wlan0
|
||||
|
||||
@param list columns
|
||||
default data attributes to return
|
||||
|
||||
@return dict
|
||||
properties: dictionary of iwlist attributes
|
||||
"""
|
||||
return get_parsed_cells(call_iwlist(interface).split('\n'))
|
||||
|
||||
10
src/py/iw_parse/license.txt
Normal file
10
src/py/iw_parse/license.txt
Normal file
@@ -0,0 +1,10 @@
|
||||
Copyright (c) 2013, Cuzzo Yahn
|
||||
All rights reserved.
|
||||
|
||||
Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met:
|
||||
|
||||
Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer.
|
||||
Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution.
|
||||
|
||||
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
|
||||
|
||||
Reference in New Issue
Block a user