#!/usr/bin/env python3
"""
=head1 NAME
switchbotmeterbt - Munin plugin to monitor temperature/humidity with SwitchBot Meter BLE
=head1 CONFIGURATION
Python 3 and bluepy are necessary.
Example for Ubuntu:
$ sudo apt install python3-pip
$ sudo pip3 install bluepy
=head1 ENVIRONMENT VARIABLES
user : root privilege is necessary for BLE scan
env.macaddr : Mac Address(es) of SwitchBot Meter(s) separated by white-space (Required)
env.hcidev : HCI device index. (Optional, default is 0)
env.tempunit : Temperature unit. (Optional, default is C)
env.scantimeout : Timeout for BLE scan. (Optional, default is 5.0 seconds)
Example:
[switchbotmeterbt]
user root
env.macaddr aa:aa:aa:aa:aa:aa bb:bb:bb:bb:bb:bb cc:cc:cc:cc:cc:cc
=head1 NOTES
For more details about SwitchBot Meter, see https://www.switch-bot.com/products/switchbot-meter
=head1 AUTHOR
K.Cima https://github.com/shakemid
=head1 LICENSE
GPLv2
SPDX-License-Identifier: GPL-2.0-only
=head1 Magic markers
#%# family=contrib
#%# capabilities=
=cut
"""
import sys
import os
import subprocess
from bluepy.btle import Scanner, DefaultDelegate
class SwitchbotScanDelegate(DefaultDelegate):
def __init__(self, macaddrs):
super().__init__()
self.sensorValues = {}
self.macaddrs = macaddrs
# Called when advertising data is received from an LE device
def handleDiscovery(self, dev, isNewDev, isNewData):
if dev.addr in self.macaddrs:
for (adtype, desc, value) in dev.getScanData():
if desc == '16b Service Data':
munin_debug(value)
self._decodeSensorData(dev, value)
def _decodeSensorData(self, dev, value):
# Extract lower 6 octets
valueBinary = bytes.fromhex(value[4:16])
# Refer to Meter BLE open API
# https://github.com/OpenWonderLabs/python-host/wiki/Meter-BLE-open-API
deviceType = chr(valueBinary[0] & 0b01111111)
battery = valueBinary[2] & 0b01111111
tint = valueBinary[4] & 0b01111111
tdec = valueBinary[3] & 0b00001111
temperature = tint + tdec / 10
isTemperatureAboveFreezing = valueBinary[4] & 0b10000000
if not isTemperatureAboveFreezing:
temperature = -temperature
humidity = valueBinary[5] & 0b01111111
self.sensorValues[dev.addr] = {
'DeviceType': deviceType,
'Temperature': temperature,
'Humidity': humidity,
'Battery': battery,
'RSSI': dev.rssi
}
munin_debug(str(self.sensorValues))
class SwitchbotMeterPlugin():
def __init__(self):
self.params = {}
self.params['pluginname'] = os.path.basename(__file__)
self.params['macaddrs'] = [str.lower(macaddr)
for macaddr in os.getenv('macaddr').split()]
self.params['hcidev'] = int(os.getenv('hcidev', 0))
self.params['tempunit'] = os.getenv('tempunit', 'C')
self.params['scantimeout'] = float(os.getenv('scantimeout', 5.0))
self.graphs = {
'temp': {
'name': 'Temperature',
'attrs': [
'graph_title SwitchBot Meter Temperature',
'graph_category sensors',
'graph_vlabel Temp ' + self.params['tempunit'],
'graph_scale no',
'graph_args --base 1000'
]
},
'humid': {
'name': 'Humidity',
'attrs': [
'graph_title SwitchBot Meter Humidity',
'graph_category sensors',
'graph_vlabel Humid %',
'graph_scale no',
'graph_args --base 1000'
]
},
'batt': {
'name': 'Battery',
'attrs': [
'graph_title SwitchBot Meter Battery',
'graph_category sensors',
'graph_vlabel %',
'graph_scale no',
'graph_args --base 1000 --lower-limit 0 --upper-limit 100'
]
},
'rssi': {
'name': 'RSSI',
'attrs': [
'graph_title SwitchBot Meter RSSI',
'graph_category sensors',
'graph_vlabel dB',
'graph_scale no',
'graph_args --base 1000'
]
}
}
def config(self):
# print config
for k in self.graphs.keys():
print('multigraph ' + self.params['pluginname'] + '_' + k)
for line in self.graphs[k]['attrs']:
print(line)
for macaddr in self.params['macaddrs']:
field = macaddr.replace(':', '')
print(field + '.label ' + macaddr)
def fetch(self):
# scan to fetch data
scanner = Scanner(self.params['hcidev']).withDelegate(SwitchbotScanDelegate(self.params['macaddrs']))
try:
# sometimes it might fail
scanner.scan(self.params['scantimeout'])
for macaddr in self.params['macaddrs']:
check = scanner.delegate.sensorValues[macaddr]
except KeyError as e:
munin_error('retry scan for exception: ' + str(type(e)))
scanner.scan(self.params['scantimeout'])
except Exception as e:
munin_error('reset hci and retry scan for exception: ' + str(type(e)))
subprocess.call(f'hciconfig hci{self.params["hcidev"]} down && hciconfig hci{self.params["hcidev"]} up', shell=True)
scanner.scan(self.params['scantimeout'])
# print value
for k in self.graphs.keys():
print('multigraph ' + self.params['pluginname'] + '_' + k)
for macaddr in self.params['macaddrs']:
field = macaddr.replace(':', '')
try:
print(field + '.value ' +
str(scanner.delegate.sensorValues[macaddr][self.graphs[k]['name']]))
except KeyError:
pass
def munin_error(message):
print(message, file=sys.stderr)
def munin_debug(message):
if os.getenv('MUNIN_DEBUG') == '1':
print('# ' + message)
def main():
plugin = SwitchbotMeterPlugin()
if len(sys.argv) > 1 and sys.argv[1] == 'config':
plugin.config()
if os.getenv('MUNIN_CAP_DIRTYCONFIG') == '1':
plugin.fetch()
else:
plugin.fetch()
if __name__ == '__main__':
main()
