Last active
May 10, 2017 15:14
-
-
Save isometry/a0313976e0cc642dc1611414f24f462b to your computer and use it in GitHub Desktop.
Log macOS Wi-Fi connection status along with RTT to external TCP socket
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
#!/usr/bin/python | |
""" | |
Robin Breathe, 2017 | |
Documentation: https://developer.apple.com/reference/corewlan | |
""" | |
from __future__ import print_function, division | |
import argparse | |
import socket | |
import logging | |
from datetime import datetime | |
from time import sleep | |
from timeit import default_timer as timer | |
try: | |
# Upstream PyObjC | |
from CoreWLAN import CWWiFiClient | |
except ImportError: | |
# Bundled PyObjC | |
import objc | |
objc.loadBundle('CoreWLAN', | |
bundle_path='/System/Library/Frameworks/CoreWLAN.framework', | |
module_globals=globals()) | |
logging.basicConfig(filename='wifi_status.log', level=logging.DEBUG) | |
def timed_socket(host, port=80, timeout=50): | |
""" | |
Test a socket connection to host with timeout specified in ms. | |
Return status and RTT in ms. | |
""" | |
try: | |
t0 = timer() | |
s = socket.create_connection((host, port), timeout/1000) | |
s.close() | |
rcode = "ONLINE" | |
except: | |
rcode = "FAILED" | |
finally: | |
t1 = timer() | |
return (rcode, (t1-t0)*1000) | |
def wifi_status(properties=('bssid', 'channel', 'txRate', 'mcsIndex', 'rssi', 'noise')): | |
xface = CWWiFiClient.sharedWiFiClient().interface() | |
while True: | |
yield({name: getattr(xface, name)() for name in properties}) | |
def main(): | |
parser = argparse.ArgumentParser(description="Monitor and log Wi-Fi Status") | |
parser.add_argument('-q', '--quiet', action='store_true', help="disable output to console") | |
parser.add_argument('-t', '--timeout', action='store', type=int, default=50, help="socket timeout (default=%(default)s ms)") | |
parser.add_argument('-i', '--interval', action='store', type=float, default=0.99, help="test intervar (default=%(default)s s)") | |
parser.add_argument('-p', '--port', action='store', type=int, default=80, help="target port (default=%(default)s s)") | |
parser.add_argument('host', nargs='?', default='www.google.com', help="target host (default=%(default)s)") | |
args = parser.parse_args() | |
if not args.quiet: | |
logger = logging.getLogger() | |
console = logging.StreamHandler() | |
logger.addHandler(console) | |
for status in wifi_status(): | |
logging.debug('{timestamp} network={network[0]} rtt={network[1]:.1f} bssid={bssid} channel={channel} txrate={txRate} mcs={mcsIndex} rssi={rssi} noise={noise}'.format( | |
timestamp=datetime.now().isoformat(), | |
network=timed_socket(args.host, port=args.port, timeout=args.timeout), | |
**status)) | |
sleep(args.interval) | |
if __name__ == "__main__": | |
main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment