path: root/weahome.py

#!/usr/bin/env python3

import sys, math, datetime, struct, atexit
from enum import Enum
from argparse import ArgumentParser
from functools import partial
import dbus
from gi.repository import GObject
from dbus.mainloop.glib import DBusGMainLoop

BLUEZ_SERVICE_NAME = 'org.bluez'
DBUS_OM_IFACE =      'org.freedesktop.DBus.ObjectManager'
DBUS_PROP_IFACE =    'org.freedesktop.DBus.Properties'

BLUEZ_DEVICE_IFACE = 'org.bluez.Device1'
GATT_SERVICE_IFACE = 'org.bluez.GattService1'
GATT_CHAR_IFACE =    'org.bluez.GattCharacteristic1'

WEAHOME_SERVICE =    '74e7fe00-c6a4-11e2-b7a9-0002a5d5c51b'
DEVINFO_CHAR =       '74e78e02-c6a4-11e2-b7a9-0002a5d5c51b'
COMMAND_CHAR =       '74e78e03-c6a4-11e2-b7a9-0002a5d5c51b'
INDOOR_CH_1_3_CHAR = '74e78e10-c6a4-11e2-b7a9-0002a5d5c51b'
CH_4_7_CHAR =        '74e78e14-c6a4-11e2-b7a9-0002a5d5c51b'
PRESSURE_CHAR =      '74e78e20-c6a4-11e2-b7a9-0002a5d5c51b'
SETTINGS_CHAR =      '74e78e2c-c6a4-11e2-b7a9-0002a5d5c51b'

BATTERY_SERVICE =    '0000180f-0000-1000-8000-00805f9b34fb'
BATTERY_CHAR =       '00002a19-0000-1000-8000-00805f9b34fb'

DEVINFO_SERVICE =    '0000180a-0000-1000-8000-00805f9b34fb'
MODEL_NAME_CHAR =    '00002a24-0000-1000-8000-00805f9b34fb'
MANUFACTURER_CHAR =  '00002a29-0000-1000-8000-00805f9b34fb'

MAX_SENSORS = 16

parser = ArgumentParser(description="Weather@home")
parser.add_argument("--sync", "-s", action='store_true', help="Sync time")
args = parser.parse_args()

DBusGMainLoop(set_as_default=True)
bus = dbus.SystemBus()
mainloop = GObject.MainLoop()

def _get_bluez_objects():
	obj_manager = bus.get_object(BLUEZ_SERVICE_NAME, '/')
	return obj_manager.GetManagedObjects(dbus_interface=DBUS_OM_IFACE)

class Event:
	__slots__ = ['_observers']

	def __init__(self):
		self._observers = []

	def observe(self, callback):
		self._observers.append(callback)

	def notify(self, *args, **kwargs):
		for cb in self._observers:
			cb(*args, **kwargs)

class Bitset:
	__slots__ = ['_val', '_len']

	def _compute_len(self):
		return (math.log(self._val, 2)) + 1

	def __init__(self, value=0, length=0):
		self._val = int(value)
		self._len = int(length)
		if self._len == 0 and self._val > 0:
			self._len = self._compute_len()

	def __len__(self):
		return self._len

	def __getitem__(self, i): # Note: LSB first
		if i < 0 or i >= self._len:
			raise IndexError
		return bool(self._val & (1 << i))

	def __str__(self):
		return ''.join(('1' if self[i] else '0') for i in reversed(range(self._len)))

class WeahomeWeather(Enum):
	PARTLY_CLOUDY = 0
	CLOUDY = 1
	RAINY = 2
	SUNNY = 3
	SNOWY = 4
	NO_DATA = 5

class WeahomeSettings:
	__slots__ = [ 'status', 'datetime', 'moon_phase', 'longitude', 'latitude', 'sun_rise', 'sun_set', 'moon_rise', 'moon_set' ]

	def __init__(self):
		self.status = None
		self.datetime = datetime.datetime.now()
		self.moon_phase = 0
		self.longitude = 0
		self.latitude = 0
		self.sun_rise = datetime.time()
		self.sun_set = datetime.time()
		self.moon_rise = datetime.time()
		self.moon_set = datetime.time()

	@staticmethod
	def unpack(value):
		status, \
		year, month, day, hour, minute, second, \
		moon_phase, \
		longitude, latitude, \
		sun_rise_hour, sun_rise_minute, sun_set_hour, sun_set_minute, \
		moon_rise_hour, moon_rise_minute, moon_set_hour, moon_set_minute \
		= struct.unpack('<B BBBBBB B HH BBBB BBBB', value)

		s = WeahomeSettings()
		s.status = status
		s.datetime = datetime.datetime(2000 + year, month, day, hour, minute, second)
		s.moon_phase = moon_phase
		s.longitude = longitude
		s.latitude = latitude
		s.sun_rise = datetime.time(sun_rise_hour, sun_rise_minute)
		s.sun_set = datetime.time(sun_set_hour, sun_set_minute)
		s.moon_rise = datetime.time(moon_rise_hour, moon_rise_minute)
		s.moon_set = datetime.time(moon_set_hour, moon_set_minute)

		return s

	def pack(self):
		return struct.pack('<B BBBBBB B HH BBBB BBBB', self.status, \
		self.datetime.year - 2000, self.datetime.month, self.datetime.day, self.datetime.hour, self.datetime.minute, self.datetime.second,
		self.moon_phase, \
		self.longitude, self.latitude, \
		self.sun_rise.hour, self.sun_rise.minute, self.sun_set.hour, self.sun_set.minute, \
		self.moon_rise.hour, self.moon_rise.minute, self.moon_set.hour, self.moon_set.minute \
		)
		

class WeahomeSensor:
	__slots__ = [ 'name', 'registered', 'online', 'low_battery', 
				  'temperature', 'temperature_max', 'temperature_min', 'temperature_trend',
				  'humidity', 'humidity_min', 'humidity_max', 'humidity_trend',
				  'pressure', 'altitude', 'altitude_pressure', 'weather' ]

	@property
	def has_temperature(self):
		return hasattr(self, 'temperature')

	@property
	def has_temperature_mem(self):
		return hasattr(self, 'temperature_min') and hasattr(self, 'temperature_max')

	@property
	def has_temperature_trend(self):
		return hasattr(self, 'temperature_trend')

	@property
	def has_humidity(self):
		return hasattr(self, 'humidity')

	@property
	def has_humidity_mem(self):
		return hasattr(self, 'humidity_min') and hasattr(self, 'humidity_max')

	@property
	def has_humidity_trend(self):
		return hasattr(self, 'humidity_trend')

	@property
	def has_pressure(self):
		return hasattr(self, 'pressure')

	@property
	def has_altitude(self):
		return hasattr(self, 'altitude') and hasattr(self, 'altitude_pressure')

	@property
	def has_weather(self):
		return hasattr(self, 'weather')

class WeahomeDevice:
	_interesting_servs = frozenset([DEVINFO_SERVICE, BATTERY_SERVICE, WEAHOME_SERVICE])
	_interesting_chars = frozenset([MODEL_NAME_CHAR, COMMAND_CHAR])
	_subscribe_chars = frozenset([DEVINFO_CHAR, INDOOR_CH_1_3_CHAR, CH_4_7_CHAR, PRESSURE_CHAR, SETTINGS_CHAR])

	def _read_char(self, uuid, refresh = False):
		char = self._chars[uuid]
		value = char.Get(GATT_CHAR_IFACE, 'Value', dbus_interface=DBUS_PROP_IFACE)
		if value and not refresh:
			return value
		return char.ReadValue({}, dbus_interface=GATT_CHAR_IFACE)

	def _write_char(self, uuid, value):
		char = self._chars[uuid]
		char.WriteValue(value, {}, dbus_interface=GATT_CHAR_IFACE)

	def _process_devinfo(self, data):
		information_type, indoor_information, registered_remote_sensor, sensor_link_status, sensor_battery_status = struct.unpack('<BBHHH', data)
		indoor_information = Bitset(indoor_information, length=8)
		registered_remote_sensor = Bitset(registered_remote_sensor, length=16)
		sensor_link_status = Bitset(sensor_link_status, length=16)
		sensor_battery_status = Bitset(sensor_battery_status, length=16)

		self.sensors[0].registered = True
		self.sensors[0].online = True
		self.sensors[0].low_battery = indoor_information[7]

		for i in range(MAX_SENSORS):
			sensor = self.sensors[i + 1]
			sensor.registered = registered_remote_sensor[i]
			sensor.online = sensor.registered and not sensor_link_status[i]

	def _parse_temperature(self, value):
		return value / 10.0

	def _parse_humidity(self, value):
		return value

	def _parse_trend(self, value, offset):
		if value[offset * 2]:
			return "up"
		elif value[offset * 2 + 1]:
			return "down"
		else:
			return "stable"

	def _parse_pressure(self, value):
		return value / 10.0

	def _parse_weather(self, value):
		return WeahomeWeather(value)

	def _process_sensor_data_1(self, offset, data):
		s0_temperature, s1_temperature, s2_temperature, s3_temperature, \
		s0_humidity, s1_humidity, s2_humidity, s3_humidity, \
		temperature_trend, humidity_trend, \
		s0_humidity_max, s0_humidity_min, s1_humidity_max, s1_humidity_min, \
		s2_humidity_max = struct.unpack('<HHHH BBBB BB BBBB B', data)

		temperature_trend = Bitset(temperature_trend, length=8)
		humidity_trend = Bitset(humidity_trend, length=8)

		self.sensors[offset+0].temperature_trend = self._parse_trend(temperature_trend, 0)
		self.sensors[offset+0].humidity_trend = self._parse_trend(humidity_trend, 0)
		self.sensors[offset+0].humidity = self._parse_humidity(s0_humidity)
		self.sensors[offset+0].humidity_min = self._parse_humidity(s0_humidity_min)
		self.sensors[offset+0].humidity_max = self._parse_humidity(s0_humidity_max)
		self.sensors[offset+0].temperature = self._parse_temperature(s0_temperature)

		self.sensors[offset+1].temperature_trend = self._parse_trend(temperature_trend, 1)
		self.sensors[offset+1].humidity_trend = self._parse_trend(humidity_trend, 1)
		self.sensors[offset+1].humidity = self._parse_humidity(s1_humidity)
		self.sensors[offset+1].humidity_min = self._parse_humidity(s1_humidity_min)
		self.sensors[offset+1].humidity_max = self._parse_humidity(s1_humidity_max)
		self.sensors[offset+1].temperature = self._parse_temperature(s1_temperature)

		self.sensors[offset+2].temperature_trend = self._parse_trend(temperature_trend, 2)
		self.sensors[offset+2].humidity_trend = self._parse_trend(humidity_trend, 2)
		self.sensors[offset+2].humidity = self._parse_humidity(s2_humidity)
		self.sensors[offset+2].humidity_max = self._parse_humidity(s2_humidity_max)
		self.sensors[offset+2].temperature = self._parse_temperature(s2_temperature)

		self.sensors[offset+3].temperature_trend = self._parse_trend(temperature_trend, 3)
		self.sensors[offset+3].humidity_trend = self._parse_trend(humidity_trend, 3)
		self.sensors[offset+3].humidity = self._parse_humidity(s3_humidity)
		self.sensors[offset+3].temperature = self._parse_temperature(s3_temperature)

	def _process_sensor_data_2(self, offset, data):
		s2_humidity_min, s3_humidity_max, s3_humidity_min, \
		s0_temperature_max, s0_temperature_min, s1_temperature_max, s1_temperature_min, \
		s2_temperature_max, s2_temperature_min, s3_temperature_max, s3_temperature_min \
		= struct.unpack('<BBB HHHH HHHH', data)

		self.sensors[offset+0].temperature_min = self._parse_temperature(s0_temperature_min)
		self.sensors[offset+0].temperature_max = self._parse_temperature(s0_temperature_max)

		self.sensors[offset+1].temperature_min = self._parse_temperature(s1_temperature_min)
		self.sensors[offset+1].temperature_max = self._parse_temperature(s1_temperature_max)

		self.sensors[offset+2].humidity_min = self._parse_humidity(s2_humidity_min)
		self.sensors[offset+2].temperature_min = self._parse_temperature(s2_temperature_min)
		self.sensors[offset+2].temperature_max = self._parse_temperature(s2_temperature_max)

		self.sensors[offset+3].humidity_min = self._parse_humidity(s3_humidity_min)
		self.sensors[offset+3].humidity_max = self._parse_humidity(s3_humidity_max)
		self.sensors[offset+3].temperature_min = self._parse_temperature(s3_temperature_min)
		self.sensors[offset+3].temperature_max = self._parse_temperature(s3_temperature_max)

	def _process_sensor_data(self, offset, data):
		last_packet = data[0] & 0x80
		data_type = data[0] & 0x7F	
		if data_type == 1:
			self._process_sensor_data_1(offset, data[1:])
		elif data_type == 2:
			self._process_sensor_data_2(offset, data[1:])
		else:
			print(" unknown sensor data type: ", data_type)
		if last_packet:
			changed_sensors = [s for s in range(offset, offset+4) if self.sensors[s].online]
			self.new_reading.notify(sender=self, sensors=changed_sensors)

	def _process_pressure_data(self, data):
		pressure, altitude_pressure, altitude, weather = struct.unpack('<HHHB', data)

		sensor_num = 15 # Barometer
		sensor = self.sensors[sensor_num]

		sensor.altitude = altitude
		sensor.altitude_pressure = self._parse_pressure(altitude_pressure)
		sensor.pressure = self._parse_pressure(pressure)
		sensor.weather = self._parse_weather(weather)

		self.new_reading.notify(sender=self, sensors=[sensor_num])

	def _process_settings_data(self, data):
		self.settings = WeahomeSettings.unpack(data)
		self.settings_changed.notify(sender=self)

	def _write_settings(self, settings):
		data = settings.pack()
		print(" new settings=", data)
		self._write_char(SETTINGS_CHAR, data)

	def _char_changed_cb(self, uuid, iface, changed_props, invalidated_props):
		if iface != GATT_CHAR_IFACE:
			return
		if len(changed_props) == 0:
			return

		value = changed_props.get('Value', None)
		if not value:
			return

		if uuid == DEVINFO_CHAR:
			self._process_devinfo(bytes(value))
		elif uuid == INDOOR_CH_1_3_CHAR:
			self._process_sensor_data(0, bytes(value))
		elif uuid == CH_4_7_CHAR:
			self._process_sensor_data(4, bytes(value))
		elif uuid == PRESSURE_CHAR:
			self._process_pressure_data(bytes(value))
		elif uuid == SETTINGS_CHAR:
			self._process_settings_data(bytes(value))
		else:
			print(" unknown notif:", uuid)

	def _char_notify_start_cb(self, uuid):
		#print("Notification set for {0}".format(uuid))
		pass

	def _char_notify_error_cb(self, uuid, error):
		print("Error while setting notification for {0}: {1}".format(uuid, error))

	def _device_changed_cb(self, iface, changed_props, invalidated_props):
		if iface != BLUEZ_DEVICE_IFACE:
			return
		if len(changed_props) == 0:
			return
		for key, value in changed_props.items():
			if key == 'Connected':
				self._test_connected_changed()
			elif key == 'ServicesResolved':
				if value:
					self._process_services()
				self._test_connected_changed()

	def _test_connected_changed(self):
		if self._old_connected != self.connected:
			self.connected_changed.notify(sender=self)
			self._old_connected = self.connected

	def _get_device_prop(self, prop):
		return self._device.Get(BLUEZ_DEVICE_IFACE, prop, dbus_interface=DBUS_PROP_IFACE)

	def _process_char(self, path):
		char = bus.get_object(BLUEZ_SERVICE_NAME, path)
		uuid = char.Get(GATT_CHAR_IFACE, 'UUID', dbus_interface=DBUS_PROP_IFACE)
		if uuid in self._interesting_chars or uuid in self._subscribe_chars:
			self._chars[uuid] = char
		if uuid in self._subscribe_chars:
			char.connect_to_signal("PropertiesChanged",
			                       partial(self._char_changed_cb, uuid),
			                       dbus_interface=DBUS_PROP_IFACE)
			char.StartNotify(reply_handler=partial(self._char_notify_start_cb, uuid),
			                 error_handler=partial(self._char_notify_error_cb, uuid), 
			                 dbus_interface=GATT_CHAR_IFACE)

	def _process_service(self, path, chars):
		service = bus.get_object(BLUEZ_SERVICE_NAME, path)
		uuid = service.Get(GATT_SERVICE_IFACE, 'UUID', dbus_interface=DBUS_PROP_IFACE)
		if uuid in self._interesting_servs:
			for char in chars:
				self._process_char(char)

	def _process_services(self):
		objects = _get_bluez_objects()
		prefix = self._device.object_path + "/"
		services = [path for path, interfaces in objects.items()
		            if path.startswith(prefix) and GATT_SERVICE_IFACE in interfaces]
		chars    = [path for path, interfaces in objects.items()
		            if path.startswith(prefix) and GATT_CHAR_IFACE in interfaces]

		self._chars.clear()
		for serv in services:
			self._process_service(serv, [char for char in chars if char.startswith(serv + '/')])

	def __init__(self, path):
		self._device = bus.get_object(BLUEZ_SERVICE_NAME, path)
		self._chars = {}

		self.sensors = [ WeahomeSensor() for i in range(MAX_SENSORS + 2) ]
		self.sensors[0].name = "Internal"
		for i in range(1, 11):
			self.sensors[i].name = "TH{0}".format(i)
		self.sensors[11].name = "Wind"
		self.sensors[12].name = "Rain"
		self.sensors[13].name = "UV"
		self.sensors[14].name = "Solar"
		self.sensors[15].name = "Barometer"
		self.sensors[16].name = "RFClock"

		self.settings = WeahomeSettings()

		self.connected_changed = Event()
		self._old_connected = self.connected

		self.new_reading = Event()

		self.settings_changed = Event()

		self._device.connect_to_signal("PropertiesChanged",
			                       self._device_changed_cb,
			                       dbus_interface=DBUS_PROP_IFACE)

		if self._get_device_prop('ServicesResolved'):
			self._process_services()

	@property
	def connected(self):
		return bool(self._get_device_prop('Connected') and len(self._chars) > 0)

	@property
	def name(self):
		return str(self._get_device_prop('Name'))

	@property
	def model(self):
		if MODEL_NAME_CHAR in self._chars:
			data = self._read_char(MODEL_NAME_CHAR)
			return ''.join(map(chr, data))
		else:
			return "Unknown"

	def connect(self):
		self._device.Connect(dbus_interface=BLUEZ_DEVICE_IFACE)

	def disconnect(self):
		self._device.Disconnect(dbus_interface=BLUEZ_DEVICE_IFACE)

	def sync_time(self):
		settings = WeahomeSettings()
		settings.status = 0
		settings.datetime = datetime.datetime.now()
		self._write_settings(settings)

def _is_weahome_device(path):
	device = bus.get_object(BLUEZ_SERVICE_NAME, path)
	uuids = device.Get(BLUEZ_DEVICE_IFACE, 'UUIDs', dbus_interface=DBUS_PROP_IFACE)
	return WEAHOME_SERVICE in uuids

def scan_for_weahome_devices():
	objects = _get_bluez_objects()
	devices = []

	for path, interfaces in objects.items():
		if BLUEZ_DEVICE_IFACE not in interfaces.keys():
			continue
		if not _is_weahome_device(path):
			continue

		device = WeahomeDevice(path)
		devices.append(device)

	return devices

def print_sensor(device, sensor):
	print("{0} Sensor {1}:".format(device.name, sensor.name))
	if sensor.has_temperature:
		print(" temperature = {0} C".format(sensor.temperature), end='')
		if sensor.has_temperature_mem:
			print(" ({0}C/{1}C)".format(sensor.temperature_min, sensor.temperature_max), end='')
		if sensor.has_temperature_trend:
			print(" ", sensor.temperature_trend, end='')
		print("")

	if sensor.has_humidity:
		print(" humidity = {0} %".format(sensor.humidity), end='')
		if sensor.has_humidity_mem:
			print(" ({0}%/{1}%)".format(sensor.humidity_min, sensor.humidity_max), end='')	
		if sensor.has_humidity_trend:
			print(" ", sensor.humidity_trend, end='')
		print("")

	if sensor.has_pressure:
		print(" pressure = {0} mbar".format(sensor.pressure))
	if sensor.has_weather:
		print(" weather = {0}".format(sensor.weather))

def _connected_changed_cb(sender):
	name = sender.name
	connected = sender.connected
	print("{0} Connected: {1}".format(name, connected))
	if connected and args.sync:
		print("{0} Syncing time".format(name))
		sender.sync_time()
	if all(not device.connected for device in devices):
		print("All disconnected")
		mainloop.quit()

def _new_reading_cb(sender, sensors):
	for s in sensors:
		sensor = sender.sensors[s]
		print_sensor(sender, sensor)

def _settings_changed_cb(sender):
	settings = sender.settings
	print("{0} Settings: status={1} datetime={2} moon_phase={3} position={4},{5} sun={6},{7} moon={8},{9}".format(sender.name, settings.status, settings.datetime, settings.moon_phase, settings.latitude, settings.longitude, settings.sun_rise, settings.sun_set, settings.moon_rise, settings.moon_set))

@atexit.register
def disconnect_all_devices():
	print("Disconnecting devices")
	for device in devices:
		if device.connected:
			device.disconnect()

devices = scan_for_weahome_devices()
print("Found {0} weather@home devices: {1}".format(len(devices), [dev.name for dev in devices]))

for device in devices:
	device.connected_changed.observe(_connected_changed_cb)
	device.new_reading.observe(_new_reading_cb)
	device.settings_changed.observe(_settings_changed_cb)
	if device.connected:
		_connected_changed_cb(device)
	else:
		device.connect()

if len(devices) > 0:
	mainloop.run()