diff options
| -rwxr-xr-x | weahome.py | 550 |
1 files changed, 550 insertions, 0 deletions
diff --git a/weahome.py b/weahome.py new file mode 100755 index 0000000..96e8b6b --- /dev/null +++ b/weahome.py @@ -0,0 +1,550 @@ +#!/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() + |