sigrok-native-macosx: Fix an issue with recent boost versions.

[sigrok-util.git] / util / eevblog-121gw / eev121gw-ble-uart-relay

#!/usr/bin/env python
##
## This file is part of the sigrok-util project.
##
## Copyright (C) 2018 Gerhard Sittig <gerhard.sittig@gmx.net>
##
## This program is free software; you can redistribute it and/or modify
## it under the terms of the GNU General Public License as published by
## the Free Software Foundation; either version 2 of the License, or
## (at your option) any later version.
##
## This program 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 General Public License for more details.
##
## You should have received a copy of the GNU General Public License
## along with this program; if not, see <http://www.gnu.org/licenses/>.
##

"""
Disguise EEVblog 121GW Bluetooth communication as a COM port (devel HACK).

This script runs as an external support process, or optionally can run on
a dedicated machine, to relay the meter's BT communication to a COM port,
where sigrok's serial-dmm driver can access the multimeter's acquisition
data. The sigrok application then gets invoked like so:

  $ sigrok-cli -d eevblog-121gw:conn=/dev/ttyUSB0 --continuous

This approach is a development HACK until the sigrok project may receive
better integration of BLE communication. But it also helps abstract away
platform dependent details or the choice for BLE communication libraries
or tools, and helps speed up the development of the 121GW meter device
support. Another benefit is the separation of the BLE communication from
the packet inspection and data processing (priviledges are required for
the former but not for the latter, COM ports typically are accessible to
local users already).

A future approach might perhaps support something like:

  -d eevblog-121gw:conn=ble/121gw/<bdaddr>
  -d eevblog-121gw:conn=tcp-ser/<ip>/<port>
"""

import argparse
from bluepy import btle
import serial
import sys
import time

_ble_scan_dev = "hci0"
_ble_scan_time = 1
_ble_scan_desc = "Complete Local Name"
_ble_scan_value = "121GW"

_ble_write_handle = 9
_ble_write_data = [ 0x03, 0x00, ]
_ble_read_handle = 8

# Verbosity levels:
# - 0, quiet operation
# - 1 (default), "summary" (got/lost connection, termination)
# - 2, plus serial data
# - 3, plus BT scan/comm details
_verbose = None

def _diag_print(level, text):
	if level > _verbose:
		return
	print("DIAG: {}".format(text))

def _parse_cmdline():
	parser = argparse.ArgumentParser(
		description = "EEVblog 121GW Bluetooth to serial converter"
	)
	parser.add_argument(
		"-i", "--hcidev",
		help = "BT HCI device name",
		default = _ble_scan_dev
	)
	parser.add_argument(
		"-b", "--bdaddr",
		help = "BT device address",
		default = ""
	)
	parser.add_argument(
		"-p", "--comport",
		help = "COM port to send bytes to",
		default = ""
	)
	parser.add_argument(
		"-B", "--bitrate",
		help = "bit rate for COM port data",
		default = 115200
	)
	parser.add_argument(
		"-v", "--verbose",
		help = "increase verbosity",
		action = "count", default = 1
	)
	parser.add_argument(
		"-q", "--quiet",
		help = "decrease verbosity",
		action = "count", default = 0
	)
	args = parser.parse_args()
	return args

def _open_comport(args):
	if not args:
		return None
	if not args.comport:
		return None
	com_port = serial.Serial(
		args.comport, args.bitrate,
		bytesize = serial.EIGHTBITS,
		parity = serial.PARITY_NONE,
		stopbits = serial.STOPBITS_ONE)
	return com_port

def _send_comport(port, data):
	port.write(data)

class CommDelegate(btle.DefaultDelegate):

	def __init__(self, port):
		btle.DefaultDelegate.__init__(self)
		self._port = port

	def handleNotification(self, handle, data):
		if handle != _ble_read_handle:
			return
		data = bytearray(data)
		self.parse_data(data)

	def parse_data(self, data):
		text = " ".join([ "{:02x}".format(d) for d in data ])
		_diag_print(2, "got BLE data: {}".format(text))
		_send_comport(self._port, data)

def _scan_bleaddr(args):
	# If not specified by the user, Scan for the device's BDADDR.

	if not args:
		return None
	if args.bdaddr:
		return args.bdaddr

	iface = args.hcidev
	if iface.startswith("hci"):
		iface = iface.replace("hci", "")
	iface = int(iface)

	_diag_print(3, "Scanning (iface {}) ...".format(iface))
	scanner = btle.Scanner(iface)
	try:
		devices = scanner.scan(_ble_scan_time)
	except btle.BTLEException as e:
		return None
	except KeyboardInterrupt as e:
		return None
	bdaddr = False
	for dev in devices:
		a, t, sd = dev.addr, dev.addrType, dev.getScanData()
		for adtype, desc, value in sd:
			if desc != _ble_scan_desc:
				continue
			if value != _ble_scan_value:
				continue
			bdaddr = a
			break

	_diag_print(3, "Done scanning, addr {}".format(bdaddr))
	return bdaddr

def _open_bleconn(args, port):
	bdaddr = _scan_bleaddr(args)
	if not bdaddr:
		return False

	_diag_print(3, "Opening bluepy(3) connection ...")
	try:
		conn = btle.Peripheral(bdaddr, btle.ADDR_TYPE_PUBLIC)
	except btle.BTLEException as e:
		return False
	except KeyboardInterrupt as e:
		return None
	conn.setDelegate(CommDelegate(port))
	data = bytearray(_ble_write_data)
	conn.writeCharacteristic(_ble_write_handle, data)
	_diag_print(3, "bluepy(3) connection {}".format(conn))
	return conn

def main():
	global _verbose

	args = _parse_cmdline()
	_verbose = args.verbose - args.quiet

	com_port = _open_comport(args)
	if not com_port:
		sys.exit(1)

	conn = None
	while True:
		# Automatically (re-)create connections, handle absence
		# of devices (retry later).
		if not conn:
			_diag_print(3, "connecting to BT ...")
			conn = _open_bleconn(args, com_port)
			if conn is None:
				break
			if conn:
				_diag_print(1, "got BLE connection.")
		if not conn:
			_diag_print(3, "could not connect to BT, waiting ...")
			time.sleep(.5)
			continue
		# Wait for incoming notifications. A registered handler
		# will process the data. Gracefully handle lost connections.
		try:
			conn.waitForNotifications(1.0)
		except KeyboardInterrupt as e:
			_diag_print(1, "got CTRL-C, terminating ...")
			break
		# except btle.BTLEException.DISCONNECTED as e:
		except btle.BTLEException as e:
			_diag_print(1, "lost BLE connection.")
			conn = None

	return

if __name__ == "__main__":
	main()