ble: barometer on its own connection - simultaneous HR+baro, auto-connect, UI
Separate BleBaroService so the HR strap and the barometer each hold their own GATT link and work at the same time (one adapter, several LE connections). Verified on device against a PC BlueZ emulator of the sensor: BleHR and BleBaro deliver in parallel, no cross-talk. - BleBaroService: dedicated baro connection (scan-before-connect + direct-connect fallback, ESS subscribe, writes ble_baro_data.json). MainActivity routes baro connects to it and auto-connects both HR (ble_device.json) and baro (baro_device.json) at start. - Rust: ble_baro_disconnect, ble_save_baro_device. - Settings: a 'Датчики (BLE)' section - heart-rate and barometer as separate rows with their own slots, bind/connect/unbind, and a live pressure/temperature/altitude readout. - Tracking: altitude_baro_m stored per point, preferred in elevation stats (null without baro). - Record screen: 'Высота' from the barometer (1 Hz, dot marker), temperature from the sensor. - Fixes: direct-connect on scan timeout; Locale.US JSON (RU decimal comma broke the parse); parsing routed by characteristic (baro no longer written into the pulse); VR/HR stop returns to the tracks list instead of into details. - tools/baro-sensor/emulate_baro.py: PC BlueZ peripheral that emulates the sensor for testing. Emulator-verified only; real board still pending.
This commit is contained in:
parent
82f3dc9d4a
commit
9251c1dd90
12 changed files with 653 additions and 34 deletions
168
tools/baro-sensor/emulate_baro.py
Normal file
168
tools/baro-sensor/emulate_baro.py
Normal file
|
|
@ -0,0 +1,168 @@
|
|||
#!/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()
|
||||
Loading…
Add table
Add a link
Reference in a new issue