baldav/tools/baro-sensor/emulate_baro.py

#!/usr/bin/env python3
# Emulate the "BaroBalDav" sensor on this PC's Bluetooth so the phone's baldav can be tested
# against the REAL Kotlin BLE read path before the ESP32 arrives. Advertises the standard
# Environmental Sensing service (0x181A) with Pressure (0x2A6D, uint32 0.1 Pa) and Temperature
# (0x2A6E, int16 0.01 C), and pushes a simulated pressure that rises and falls so altitude moves.
#
# Pure BlueZ D-Bus GATT server (python3-dbus + python3-gi) - no bluetoothctl, no pip installs.
# Run:  python3 tools/baro-sensor/emulate_baro.py      (Ctrl+C to stop)
import struct, math, time, sys
import dbus, dbus.mainloop.glib, dbus.service
from gi.repository import GLib

BLUEZ = "org.bluez"
ADAPTER_IFACE = "org.bluez.Adapter1"
GATT_MANAGER = "org.bluez.GattManager1"
LE_ADV_MANAGER = "org.bluez.LEAdvertisingManager1"
DBUS_OM = "org.freedesktop.DBus.ObjectManager"
DBUS_PROP = "org.freedesktop.DBus.Properties"

ESS_UUID = "0000181a-0000-1000-8000-00805f9b34fb"
PRESSURE_UUID = "00002a6d-0000-1000-8000-00805f9b34fb"
TEMP_UUID = "00002a6e-0000-1000-8000-00805f9b34fb"

def find_adapter(bus):
    om = dbus.Interface(bus.get_object(BLUEZ, "/"), DBUS_OM)
    for path, ifaces in om.GetManagedObjects().items():
        if ADAPTER_IFACE in ifaces and LE_ADV_MANAGER in ifaces:
            return path
    return None

# ---- GATT application / service / characteristic boilerplate (BlueZ example pattern) ----
class Application(dbus.service.Object):
    def __init__(self, bus):
        self.path = "/dev/baldav/baro"
        self.services = []
        dbus.service.Object.__init__(self, bus, self.path)
    def get_path(self): return dbus.ObjectPath(self.path)
    def add_service(self, s): self.services.append(s)
    @dbus.service.method(DBUS_OM, out_signature="a{oa{sa{sv}}}")
    def GetManagedObjects(self):
        out = {}
        for s in self.services:
            out[s.get_path()] = s.get_properties()
            for c in s.chars:
                out[c.get_path()] = c.get_properties()
        return out

class Service(dbus.service.Object):
    def __init__(self, bus, index, uuid, primary=True):
        self.path = "/dev/baldav/baro/service%d" % index
        self.uuid = uuid; self.primary = primary; self.chars = []
        dbus.service.Object.__init__(self, bus, self.path)
    def get_path(self): return dbus.ObjectPath(self.path)
    def add_char(self, c): self.chars.append(c)
    def get_properties(self):
        return {"org.bluez.GattService1": {
            "UUID": self.uuid, "Primary": self.primary,
            "Characteristics": dbus.Array([c.get_path() for c in self.chars], signature="o")}}

class Characteristic(dbus.service.Object):
    def __init__(self, bus, index, uuid, flags, service):
        self.path = service.path + "/char%d" % index
        self.uuid = uuid; self.flags = flags; self.service = service
        self.notifying = False; self.value = []
        dbus.service.Object.__init__(self, bus, self.path)
    def get_path(self): return dbus.ObjectPath(self.path)
    def get_properties(self):
        return {"org.bluez.GattCharacteristic1": {
            "Service": self.service.get_path(), "UUID": self.uuid, "Flags": self.flags}}
    @dbus.service.method(DBUS_PROP, in_signature="s", out_signature="a{sv}")
    def GetAll(self, iface):
        if iface != "org.bluez.GattCharacteristic1":
            raise dbus.exceptions.DBusException("org.bluez.Error.InvalidArguments")
        return self.get_properties()["org.bluez.GattCharacteristic1"]
    @dbus.service.method("org.bluez.GattCharacteristic1", in_signature="a{sv}", out_signature="ay")
    def ReadValue(self, options): return dbus.Array(self.value, signature="y")
    @dbus.service.method("org.bluez.GattCharacteristic1")
    def StartNotify(self):
        self.notifying = True
        print("  [%s] StartNotify" % self.uuid[4:8])
    @dbus.service.method("org.bluez.GattCharacteristic1")
    def StopNotify(self):
        self.notifying = False
        print("  [%s] StopNotify" % self.uuid[4:8])
    @dbus.service.signal(DBUS_PROP, signature="sa{sv}as")
    def PropertiesChanged(self, iface, changed, invalidated): pass
    def set_value(self, raw):
        self.value = [dbus.Byte(b) for b in raw]
        if self.notifying:
            self.PropertiesChanged("org.bluez.GattCharacteristic1",
                                   {"Value": dbus.Array(self.value, signature="y")}, [])

# ---- LE advertisement ----
class Advertisement(dbus.service.Object):
    def __init__(self, bus, index):
        self.path = "/dev/baldav/baro/adv%d" % index
        dbus.service.Object.__init__(self, bus, self.path)
    def get_path(self): return dbus.ObjectPath(self.path)
    @dbus.service.method(DBUS_PROP, in_signature="s", out_signature="a{sv}")
    def GetAll(self, iface):
        if iface != "org.bluez.LEAdvertisement1":
            raise dbus.exceptions.DBusException("org.bluez.Error.InvalidArguments")
        return {
            "Type": "peripheral",
            "ServiceUUIDs": dbus.Array([ESS_UUID], signature="s"),
            "LocalName": dbus.String("BaroBalDav"),
            "Includes": dbus.Array(["tx-power"], signature="s"),
        }
    @dbus.service.method("org.bluez.LEAdvertisement1")
    def Release(self): print("Advertisement released")

# ---- pressure/temperature simulation ----
def enc_pressure(pa): return struct.pack("<I", round(pa * 10))   # uint32, 0.1 Pa
def enc_temp(tc):     return struct.pack("<h", round(tc * 100))  # int16, 0.01 C

def main():
    dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
    bus = dbus.SystemBus()
    adapter = find_adapter(bus)
    if not adapter:
        print("No BLE adapter with GATT+Advertising found"); sys.exit(1)
    print("adapter:", adapter)

    app = Application(bus)
    svc = Service(bus, 0, ESS_UUID, True)
    press = Characteristic(bus, 0, PRESSURE_UUID, ["read", "notify"], svc)
    temp = Characteristic(bus, 1, TEMP_UUID, ["read", "notify"], svc)
    svc.add_char(press); svc.add_char(temp); app.add_service(svc)

    gatt = dbus.Interface(bus.get_object(BLUEZ, adapter), GATT_MANAGER)
    adv_mgr = dbus.Interface(bus.get_object(BLUEZ, adapter), LE_ADV_MANAGER)
    adv = Advertisement(bus, 0)

    t0 = time.time()
    def tick():
        t = time.time() - t0
        # sea-level-ish pressure that swings ~600 Pa (~ +-50 m) with a ~60 s period, so altitude moves
        pa = 101325.0 - 600.0 * math.sin(t / 10.0)
        tc = 24.0 + 0.5 * math.sin(t / 25.0)
        press.set_value(enc_pressure(pa))
        temp.set_value(enc_temp(tc))
        if int(t) % 5 == 0:
            print("  sim: P=%.1f Pa (%.2f hPa)  T=%.2f C" % (pa, pa/100.0, tc))
        return True
    tick()

    def reg_ok(): print("GATT application registered")
    def reg_err(e): print("GATT register error:", e); mainloop.quit()
    def adv_ok(): print("Advertising as 'BaroBalDav' (service 0x181A)")
    def adv_err(e): print("Advertising error:", e); mainloop.quit()

    gatt.RegisterApplication(app.get_path(), {}, reply_handler=reg_ok, error_handler=reg_err)
    adv_mgr.RegisterAdvertisement(adv.get_path(), {}, reply_handler=adv_ok, error_handler=adv_err)

    GLib.timeout_add(1000, tick)
    mainloop = GLib.MainLoop()
    print("BaroBalDav emulator running - connect from the phone. Ctrl+C to stop.")
    try:
        mainloop.run()
    except KeyboardInterrupt:
        print("\nstopping")
        try: adv_mgr.UnregisterAdvertisement(adv.get_path())
        except Exception: pass
        try: gatt.UnregisterApplication(app.get_path())
        except Exception: pass

if __name__ == "__main__":
    main()