feat(gateway): add deduplication and continuous BLE scan
- Add 10-second deduplication window per MAC address to avoid duplicate entries in the database (each Thingy advertises 3-4 times per frame on BLE channels 37, 38 and 39) - Configure BleakScanner with continuous scan (interval = window = 10ms) to ensure no advertising packets are missed during the 500ms Thingy advertising window Assisted-by: Claude:claude-sonnet-4-6 — BleakScanner scan interval and window configuration
This commit is contained in:
@@ -32,6 +32,11 @@ class Gateway:
|
|||||||
# 5 keys (1B each) + window(1B) + humidity(1B) + temp(2B) + co2(4B) + battery(1B) = 14 bytes
|
# 5 keys (1B each) + window(1B) + humidity(1B) + temp(2B) + co2(4B) + battery(1B) = 14 bytes
|
||||||
EXPECTED_PAYLOAD_SIZE = 14
|
EXPECTED_PAYLOAD_SIZE = 14
|
||||||
|
|
||||||
|
# Deduplication window in seconds — ignore packets from the same node
|
||||||
|
# within this period to avoid duplicate entries in the database.
|
||||||
|
# Each node advertises 3-4 times per frame on BLE channels 37, 38 and 39.
|
||||||
|
DEDUP_WINDOW_SECONDS = 10
|
||||||
|
|
||||||
def __init__(self, config: dict):
|
def __init__(self, config: dict):
|
||||||
self.gateway_id = config["gateway_id"]
|
self.gateway_id = config["gateway_id"]
|
||||||
self.mqtt_broker = config["mqtt"]["broker"]
|
self.mqtt_broker = config["mqtt"]["broker"]
|
||||||
@@ -40,6 +45,11 @@ class Gateway:
|
|||||||
# BLE service UUID used to identify Thingy:52 advertising packets
|
# BLE service UUID used to identify Thingy:52 advertising packets
|
||||||
self.service_uuid = config["ble"]["service_uuid"]
|
self.service_uuid = config["ble"]["service_uuid"]
|
||||||
|
|
||||||
|
# Deduplication cache — stores the last time a packet was published
|
||||||
|
# for each MAC address. Packets received within DEDUP_WINDOW_SECONDS
|
||||||
|
# from the same node are ignored.
|
||||||
|
self._last_published = {}
|
||||||
|
|
||||||
log.info(f"Gateway ID : {self.gateway_id}")
|
log.info(f"Gateway ID : {self.gateway_id}")
|
||||||
log.info(f"MQTT broker: {self.mqtt_broker}:{self.mqtt_port}")
|
log.info(f"MQTT broker: {self.mqtt_broker}:{self.mqtt_port}")
|
||||||
|
|
||||||
@@ -128,6 +138,23 @@ class Gateway:
|
|||||||
break
|
break
|
||||||
return result
|
return result
|
||||||
|
|
||||||
|
def is_duplicate(self, mac: str) -> bool:
|
||||||
|
"""Check if a packet from this MAC was already published recently.
|
||||||
|
|
||||||
|
Each Thingy:52 advertises 3-4 times per broadcast frame on BLE
|
||||||
|
advertising channels 37, 38 and 39. Without deduplication, the same
|
||||||
|
measurement would be published multiple times to the broker.
|
||||||
|
This method ignores packets from the same node received within
|
||||||
|
DEDUP_WINDOW_SECONDS of the last published packet.
|
||||||
|
"""
|
||||||
|
now = datetime.now(timezone.utc)
|
||||||
|
last = self._last_published.get(mac)
|
||||||
|
if last and (now - last).total_seconds() < self.DEDUP_WINDOW_SECONDS:
|
||||||
|
log.debug(f"{mac} | duplicate ignored — last published {(now - last).total_seconds():.1f}s ago")
|
||||||
|
return True
|
||||||
|
self._last_published[mac] = now
|
||||||
|
return False
|
||||||
|
|
||||||
def publish(self, mac: str, data: dict):
|
def publish(self, mac: str, data: dict):
|
||||||
"""Publish decoded sensor data to MQTT broker.
|
"""Publish decoded sensor data to MQTT broker.
|
||||||
Topic: {gateway_id}/{thingy_mac}/update
|
Topic: {gateway_id}/{thingy_mac}/update
|
||||||
@@ -163,6 +190,10 @@ class Gateway:
|
|||||||
log.debug(f"{device.address} | ignored — unexpected payload size: {len(raw)}")
|
log.debug(f"{device.address} | ignored — unexpected payload size: {len(raw)}")
|
||||||
return
|
return
|
||||||
|
|
||||||
|
# Deduplication — ignore packets from the same node within 10 seconds
|
||||||
|
if self.is_duplicate(device.address):
|
||||||
|
return
|
||||||
|
|
||||||
log.debug(f"{device.address} | Thingy detected, raw: {list(raw)}")
|
log.debug(f"{device.address} | Thingy detected, raw: {list(raw)}")
|
||||||
|
|
||||||
data = self.decode_payload(raw)
|
data = self.decode_payload(raw)
|
||||||
@@ -174,9 +205,20 @@ class Gateway:
|
|||||||
self.publish(device.address, data)
|
self.publish(device.address, data)
|
||||||
|
|
||||||
async def run(self):
|
async def run(self):
|
||||||
"""Start passive BLE scan and run indefinitely."""
|
"""Start passive BLE scan and run indefinitely.
|
||||||
|
|
||||||
|
Scanner is configured with a continuous scan (interval = window = 10ms)
|
||||||
|
to ensure no advertising packets are missed. By default, BlueZ uses
|
||||||
|
an interval of 1280ms and a window of 11.25ms — less than 1% duty cycle.
|
||||||
|
With the Thingy:52 advertising window of only 500ms, packets could be
|
||||||
|
missed. Setting interval = window guarantees continuous listening on
|
||||||
|
all three BLE advertising channels (37, 38 and 39).
|
||||||
|
"""
|
||||||
log.info("BLE scan started, listening for Thingy:52 advertising packets...")
|
log.info("BLE scan started, listening for Thingy:52 advertising packets...")
|
||||||
scanner = BleakScanner(detection_callback=self.on_device_found)
|
scanner = BleakScanner(
|
||||||
|
detection_callback=self.on_device_found,
|
||||||
|
bluez={"interval": 0x0010, "window": 0x0010}
|
||||||
|
)
|
||||||
await scanner.start()
|
await scanner.start()
|
||||||
while True:
|
while True:
|
||||||
await asyncio.sleep(1)
|
await asyncio.sleep(1)
|
||||||
|
|||||||
Reference in New Issue
Block a user