Fix floor persistence and improve movement detection

- Store floor assignments in SQLite database for persistence
- Floor now saves correctly for live-tracked BT devices
- Add statistical movement detection (5-sample average + stddev)
- Require 1.5m + 2σ deviation to mark device as moving
- Reduces false positives from RSSI noise/fluctuations
- Add /api/device/floors endpoint for bulk floor retrieval

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
User
2026-02-01 00:34:49 +01:00
parent 0e99232582
commit dda8455813
3 changed files with 154 additions and 48 deletions

View File

@@ -693,7 +693,7 @@ def create_app(config: Config | None = None) -> Flask:
@app.route("/api/device/<device_id>/floor", methods=["POST"])
def api_device_floor(device_id: str):
"""Assign floor to a device in the most recent scan"""
"""Assign floor to a device - stores in database for persistence"""
data = request.get_json() or {}
floor = data.get("floor")
height_m = data.get("height_m")
@@ -701,50 +701,46 @@ def create_app(config: Config | None = None) -> Flask:
if floor is None and height_m is None:
return jsonify({"error": "Must provide floor or height_m"}), 400
# Load the most recent scan
# Store floor in database for persistence
db = app.config.get("DATABASE")
if db and floor is not None:
db.set_device_floor(device_id, int(floor))
# Also update the most recent scan file if device exists there
data_dir = app.config["DATA_DIR"]
scan_files = sorted(data_dir.glob("scan_*.json"), reverse=True)
device_type = "unknown"
if not scan_files:
return jsonify({"error": "No scans found"}), 404
if scan_files:
scan_file = scan_files[0]
with open(scan_file) as f:
scan = json.load(f)
scan_file = scan_files[0]
with open(scan_file) as f:
scan = json.load(f)
# Find and update the device
updated = False
device_type = None
# Check WiFi networks (by BSSID)
for net in scan.get("wifi_networks", []):
if net.get("bssid") == device_id:
if floor is not None:
net["floor"] = int(floor)
if height_m is not None:
net["height_m"] = float(height_m)
updated = True
device_type = "wifi"
break
# Check Bluetooth devices (by address)
if not updated:
for dev in scan.get("bluetooth_devices", []):
if dev.get("address") == device_id:
updated = False
for net in scan.get("wifi_networks", []):
if net.get("bssid") == device_id:
if floor is not None:
dev["floor"] = int(floor)
net["floor"] = int(floor)
if height_m is not None:
dev["height_m"] = float(height_m)
net["height_m"] = float(height_m)
updated = True
device_type = "bluetooth"
device_type = "wifi"
break
if not updated:
return jsonify({"error": "Device not found"}), 404
if not updated:
for dev in scan.get("bluetooth_devices", []):
if dev.get("address") == device_id:
if floor is not None:
dev["floor"] = int(floor)
if height_m is not None:
dev["height_m"] = float(height_m)
updated = True
device_type = "bluetooth"
break
# Save the updated scan
with open(scan_file, "w") as f:
json.dump(scan, f, indent=2)
if updated:
with open(scan_file, "w") as f:
json.dump(scan, f, indent=2)
return jsonify({
"status": "updated",
@@ -812,6 +808,14 @@ def create_app(config: Config | None = None) -> Flask:
# Store previous distances for movement detection logging
_bt_previous_distances: dict = {}
@app.route("/api/device/floors", methods=["GET"])
def api_device_floors():
"""Get all saved floor assignments"""
db = app.config.get("DATABASE")
if not db:
return jsonify({})
return jsonify(db.get_all_device_floors())
@app.route("/api/scan/bt", methods=["POST"])
def api_scan_bt():
"""Quick Bluetooth-only scan for real-time tracking using bleak (BLE)"""
@@ -842,6 +846,10 @@ def create_app(config: Config | None = None) -> Flask:
print(f"[BT] Bleak scan error: {e}")
bt = []
# Get saved floor assignments from database
db = app.config.get("DATABASE")
saved_floors = db.get_all_device_floors() if db else {}
# The scanner already does auto-identification if enabled
# Just use the device_type already assigned by the scanner
@@ -885,6 +893,9 @@ def create_app(config: Config | None = None) -> Flask:
scan_id=None # Live tracking, no scan_id
)
# Get saved floor from database
saved_floor = saved_floors.get(addr)
response_data["bluetooth_devices"].append({
"address": addr,
"name": name,
@@ -894,7 +905,7 @@ def create_app(config: Config | None = None) -> Flask:
"manufacturer": "",
"estimated_distance_m": round(dist, 2),
"signal_quality": "Fair",
"floor": None,
"floor": saved_floor,
"height_m": None
})