Documentation Index
Fetch the complete documentation index at: https://mintlify.com/AngelAmoSanchez/TFG-RaspberryPi-BLE/llms.txt
Use this file to discover all available pages before exploring further.
The agent uses Bleak to run passive BLE scans on the Raspberry Pi’s Bluetooth adapter. Each scan cycle collects advertisement packets from nearby devices, deduplicates them by MAC address, anonymises identities, classifies devices into proximity zones, and publishes the result to the configured backend. No active connections are made to any device at any point.
Data model
Two frozen dataclasses in src/scanner/detection.py describe the data flowing through the pipeline.
Detection — raw scan result
@dataclass(frozen=True)
class Detection:
"""Represents a raw BLE detection directly from the scan"""
mac_address: str
rssi: int
timestamp: datetime
device_name: Optional[str] = None # (if available)
def __post_init__(self):
if not self.mac_address:
raise ValueError("MAC address cannot be empty")
if self.rssi > 0:
raise ValueError(f"RSSI must be negative, got {self.rssi}")
if self.rssi < -100:
raise ValueError(f"RSSI out of range, got {self.rssi}")
Device — processed and anonymised
@dataclass(frozen=True)
class Device:
"""Device after domain processing and anonymisation"""
device_hash: str # SHA-256 of MAC address (64 hex chars)
rssi: int
zone: Zone
timestamp: datetime
Zone — proximity classification
class Zone(Enum):
"""Proximity zones based on RSSI signal strength"""
NEAR = "near" # RSSI >= -60 dBm → 0–2 m
MEDIUM = "medium" # -75 <= RSSI < -60 dBm → 2–5 m
FAR = "far" # RSSI < -75 dBm → > 5 m
def get_description(self) -> str:
descriptions = {
Zone.NEAR: "Near zone (0-2m)",
Zone.MEDIUM: "Medium zone (2-5m)",
Zone.FAR: "Far zone (>5m)",
}
return descriptions[self]
NEAR ≥ -60 dBm, MEDIUM ≥ -75 dBm) can be overridden through NEAR_THRESHOLD and MEDIUM_THRESHOLD in .env. See calibration for how to choose values appropriate for your environment.
BLEScanner — passive scanning with Bleak
BLEScanner in src/scanner/ble_scanner.py wraps BleakScanner with a callback-based approach and per-cycle deduplication:
class BLEScanner:
"""BLE scanner using Bleak (passive scanning)"""
def __init__(self, scan_duration: int = 10):
self.scan_duration = scan_duration
self._devices_cache = {}
async def scan_devices(self) -> List[Detection]:
self._devices_cache.clear()
def detection_callback(device: BLEDevice, advertisement_data: AdvertisementData):
rssi = advertisement_data.rssi
# Keep the best (highest) RSSI seen per MAC in this cycle
if device.address in self._devices_cache:
if rssi > self._devices_cache[device.address].rssi:
self._devices_cache[device.address] = Detection(
mac_address=device.address,
rssi=rssi,
timestamp=datetime.now(SPAIN_TZ),
device_name=device.name,
)
else:
self._devices_cache[device.address] = Detection(
mac_address=device.address,
rssi=rssi,
timestamp=datetime.now(SPAIN_TZ),
device_name=device.name,
)
scanner = BleakScanner(detection_callback=detection_callback)
await scanner.start()
await asyncio.sleep(self.scan_duration)
await scanner.stop()
return list(self._devices_cache.values())
- Passive scanning — the adapter listens for advertisement packets only. No connection requests are sent, so scanning is invisible to nearby devices.
- Callback-based collection —
BleakScanner calls detection_callback for every advertisement received during the scan window, which may include multiple packets from the same device.
- Best-RSSI deduplication —
_devices_cache is keyed by MAC address. When the same MAC is seen more than once per cycle, only the entry with the stronger (higher) RSSI is kept. This gives the most optimistic distance estimate for the cycle.
DetectionProcessor — anonymisation and zone classification
DetectionProcessor in src/main.py converts a list of raw Detection objects into Device objects with no personally identifiable information:
class DetectionProcessor:
def __init__(self, near_threshold: int = -60, medium_threshold: int = -75):
self.near_threshold = near_threshold
self.medium_threshold = medium_threshold
def anonymize_mac(self, mac_address: str) -> str:
"""Anonymise MAC address using SHA-256.
Returns 64-character hex digest."""
normalized_mac = mac_address.replace(":", "").replace("-", "").upper()
hash_object = hashlib.sha256(normalized_mac.encode())
return hash_object.hexdigest()
def classify_zone(self, rssi: int) -> Zone:
"""Classify device into proximity zone based on RSSI."""
if rssi >= self.near_threshold:
return Zone.NEAR
elif rssi >= self.medium_threshold:
return Zone.MEDIUM
else:
return Zone.FAR
def process_detections(self, detections: List[Detection]) -> List[Device]:
processed = []
for detection in detections:
device_hash = self.anonymize_mac(detection.mac_address)
zone = self.classify_zone(detection.rssi)
processed.append(
Device.from_detection(detection=detection, device_hash=device_hash, zone=zone)
)
return processed
anonymize_mac strips separators and uppercases the address before hashing, so AA:BB:CC:DD:EE:FF, AA-BB-CC-DD-EE-FF, and aabbccddeeff all produce the same 64-character SHA-256 hex digest. The original MAC address is discarded and never written to disk or sent over the network.
IoTAgent run loop
IoTAgent.run() in src/main.py ties everything together in an async loop:
async def run(self):
"""
Scan → process → send → sleep → repeat
"""
if not self.client.connect():
logger.error("Could not connect to backend, entering offline mode")
while self.running:
# 1. Scan
detections = await self.scanner.scan_devices()
# 2. Process and anonymise
if detections:
processed_detections = self.processor.process_detections(detections)
# 3. Publish
success = self.client.publish_detections(processed_detections)
buffer_size = self.client.get_buffer_size()
if buffer_size > 0:
logger.warning(f"Buffer has {buffer_size} pending messages")
# 4. Sleep until next cycle
await asyncio.sleep(self.config.scanner.scan_interval)
MockBLEScanner — testing without hardware
When USE_MOCK_SCANNER=true the agent instantiates MockBLEScanner instead of BLEScanner. The mock generates a randomised subset of eight built-in device profiles each cycle:
class MockBLEScanner:
def __init__(self, num_devices: int = 5):
self._mock_devices = [
("A1:B2:C3:D4:E5:F1", "iPhone 13", -55),
("A1:B2:C3:D4:E5:F2", "Samsung Galaxy S3", -62),
("A1:B2:C3:D4:E5:F3", "Apple Watch", -48),
("A1:B2:C3:D4:E5:F4", "Xiaomi Band", -75),
("A1:B2:C3:D4:E5:F5", "Airpods Pro", -58),
("A1:B2:C3:D4:E5:F6", "Unknown Device", -80),
("A1:B2:C3:D4:E5:F7", "Smartwatch", -65),
("A1:B2:C3:D4:E5:F8", "Fitness Smartwatch", -72),
]
async def scan_devices(self) -> List[Detection]:
await asyncio.sleep(1) # simulate scan latency
num_to_detect = random.randint(
max(1, self.num_devices - 2),
min(len(self._mock_devices), self.num_devices + 2),
)
selected_devices = random.sample(self._mock_devices, num_to_detect)
detections = []
for mac, name, base_rssi in selected_devices:
rssi = base_rssi + random.randint(-5, 5) # ±5 dBm jitter
detections.append(
Detection(mac_address=mac, rssi=rssi,
timestamp=datetime.now(SPAIN_TZ), device_name=name)
)
return detections
MockBLEScanner is drop-in compatible with BLEScanner — both expose an async scan_devices() method and a synchronous is_available() method, so the IoTAgent code path is identical for both.