feat: add multi-scanner trilateration and signal heat map
- Add database methods for multi-scanner RSSI queries - Add weighted trilateration function supporting 2+ scanners - Add /api/positions/trilaterated endpoint - Add /api/heatmap/signal endpoint for heat map data - Update frontend to show trilaterated positions with gold markers - Add heat map toggle button for signal coverage visualization Trilateration uses RSSI from multiple scanners to calculate device positions with confidence scores. Devices seen by 2+ scanners within 60 seconds get trilaterated positions shown with gold border markers. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
@@ -496,6 +496,56 @@ class DeviceDatabase:
|
||||
max_distance_m=round(stats['max_distance_m'], 2) if stats['max_distance_m'] else 0
|
||||
)
|
||||
|
||||
def get_device_multi_scanner_rssi(self, device_id: str, seconds: int = 60) -> list[dict]:
|
||||
"""Get recent RSSI readings per scanner for a device.
|
||||
|
||||
Args:
|
||||
device_id: The device MAC address
|
||||
seconds: Time window in seconds (default 60)
|
||||
|
||||
Returns:
|
||||
List of dicts with scanner_id, avg_rssi, sample_count, last_seen
|
||||
"""
|
||||
conn = self._get_connection()
|
||||
cursor = conn.cursor()
|
||||
|
||||
query = """
|
||||
SELECT scanner_id,
|
||||
AVG(rssi) as avg_rssi,
|
||||
COUNT(*) as sample_count,
|
||||
MAX(timestamp) as last_seen
|
||||
FROM rssi_history
|
||||
WHERE device_id = ?
|
||||
AND scanner_id IS NOT NULL
|
||||
AND timestamp >= datetime('now', '-' || ? || ' seconds')
|
||||
GROUP BY scanner_id
|
||||
"""
|
||||
cursor.execute(query, (device_id, seconds))
|
||||
return [dict(r) for r in cursor.fetchall()]
|
||||
|
||||
def get_devices_seen_by_multiple_scanners(self, seconds: int = 60) -> list[str]:
|
||||
"""Get device IDs seen by 2+ scanners recently.
|
||||
|
||||
Args:
|
||||
seconds: Time window in seconds (default 60)
|
||||
|
||||
Returns:
|
||||
List of device IDs seen by multiple scanners
|
||||
"""
|
||||
conn = self._get_connection()
|
||||
cursor = conn.cursor()
|
||||
|
||||
query = """
|
||||
SELECT device_id
|
||||
FROM rssi_history
|
||||
WHERE scanner_id IS NOT NULL
|
||||
AND timestamp >= datetime('now', '-' || ? || ' seconds')
|
||||
GROUP BY device_id
|
||||
HAVING COUNT(DISTINCT scanner_id) >= 2
|
||||
"""
|
||||
cursor.execute(query, (seconds,))
|
||||
return [r['device_id'] for r in cursor.fetchall()]
|
||||
|
||||
def get_movement_events(self, device_id: Optional[str] = None,
|
||||
since: Optional[str] = None,
|
||||
limit: int = 100) -> list[dict]:
|
||||
|
||||
Reference in New Issue
Block a user