n0mad1k/bigbrother
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix P1 blockers: configurable port, connection leak, race condition, DB path + WAL

Browse Source 
- P1 #551: Add STATUS_PORT env var (default 9191) for configurable HTTP status server port
- P1 #552: Fix SQLite connection leak — wrap all sqlite3.connect() with try/finally in lookup_person(), seed_persons_from_db(), log_signal(), log_presence_change()
- P1 #553: Fix race condition in update_person_state() — compute certainty snapshot under devices_lock, determine state transition under persons_lock, send alert outside locks to prevent stale snapshot
- P2 #559 (included): Add WAL mode + pragmas to init_db() for concurrent write safety
- Database path: Align to /opt/sensor/presence.db in both daemon and install.sh
- Increase connection timeout from 1s to 10s across all functions
- install.sh: Document PRESENCE_STATUS_PORT and PRESENCE_MATRIX_WEBHOOK env vars
This commit is contained in:
Cobra
2026-04-10 15:02:26 -04:00
parent 09b81deb45
commit fd1eeeda47
2 changed files with 51 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
presence/install.sh
+5 -1
View File
@@ -28,7 +28,11 @@ Type=simple
ExecStart=/usr/bin/python3 /opt/sensor/sensor.py
Restart=on-failure
RestartSec=10
Environment=PRESENCE_DB_PATH=/opt/sensor/sensor.db
Environment=PRESENCE_DB_PATH=/opt/sensor/presence.db
# Optional: override default status port (9191)
# Environment=PRESENCE_STATUS_PORT=9191
# Optional: configure Matrix webhook for alerts
# Environment=PRESENCE_MATRIX_WEBHOOK=https://matrix.example.com/hook
[Install]
WantedBy=multi-user.target
presence/presence_daemon.py
+46 -27
View File
@@ -26,8 +26,9 @@ from typing import Optional
# --- Config (env-driven) ---
MONITOR_IFACE = os.getenv("PRESENCE_MONITOR_IFACE", "wlan1")
DATA_IFACE = os.getenv("PRESENCE_DATA_IFACE", "wlan0")
DB_PATH = os.getenv("PRESENCE_DB_PATH", "/opt/presence/presence.db")
DB_PATH = os.getenv("PRESENCE_DB_PATH", "/opt/sensor/presence.db")
MATRIX_WEBHOOK = os.getenv("PRESENCE_MATRIX_WEBHOOK", "")
STATUS_PORT = int(os.getenv("PRESENCE_STATUS_PORT", "9191"))
PRESENCE_THRESHOLD = 0.40
ABSENCE_THRESHOLD = 0.10
ANCHOR_MINUTES = 45
@@ -138,42 +139,47 @@ def init_db() -> None:
db_path = Path(DB_PATH)
db_path.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(DB_PATH, timeout=10)
try:
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
schema = schema_path.read_text()
cursor.executescript(schema)
conn.commit()
conn.close()
cursor.execute("PRAGMA journal_mode=WAL")
cursor.execute("PRAGMA synchronous=NORMAL")
cursor.execute("PRAGMA foreign_keys=ON")
conn.commit()
logger.info(f"Database initialized at {DB_PATH}")
except Exception as e:
logger.error(f"Failed to init database: {e}")
finally:
conn.close()
def lookup_person(mac: str) -> Optional[int]:
"""Look up person_id for a MAC from database."""
conn = sqlite3.connect(DB_PATH, timeout=10)
try:
conn = sqlite3.connect(DB_PATH, timeout=1)
cursor = conn.cursor()
result = cursor.execute(
"SELECT person_id FROM devices WHERE mac = ?",
(mac.lower(),)
).fetchone()
conn.close()
return result[0] if result else None
except Exception as e:
logger.debug(f"lookup_person error for {mac}: {e}")
return None
finally:
conn.close()
def seed_persons_from_db() -> None:
"""Load all persons from database into memory."""
conn = sqlite3.connect(DB_PATH, timeout=10)
try:
conn = sqlite3.connect(DB_PATH, timeout=1)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
rows = cursor.execute("SELECT id, name FROM persons").fetchall()
conn.close()
with persons_lock:
for row in rows:
@@ -185,36 +191,40 @@ def seed_persons_from_db() -> None:
logger.info(f"Loaded {len(rows)} persons from database")
except Exception as e:
logger.error(f"Failed to seed persons: {e}")
finally:
conn.close()
def log_signal(mac: str, signal_type: str, certainty: float) -> None:
"""Insert signal into database (non-blocking via queue would be ideal, but inline for simplicity)."""
conn = sqlite3.connect(DB_PATH, timeout=10)
try:
conn = sqlite3.connect(DB_PATH, timeout=1)
cursor = conn.cursor()
cursor.execute(
"INSERT INTO signals (mac, signal_type, certainty, ts) VALUES (?, ?, ?, ?)",
(mac.lower(), signal_type, certainty, time.time())
)
conn.commit()
conn.close()
except Exception as e:
logger.debug(f"Failed to log signal: {e}")
finally:
conn.close()
def log_presence_change(person_id: int, old_status: str, new_status: str) -> None:
"""Insert occupancy change into database."""
conn = sqlite3.connect(DB_PATH, timeout=10)
try:
conn = sqlite3.connect(DB_PATH, timeout=1)
cursor = conn.cursor()
cursor.execute(
"INSERT INTO occupancy_log (person_id, old_status, new_status, ts) VALUES (?, ?, ?, ?)",
(person_id, old_status, new_status, time.time())
)
conn.commit()
conn.close()
except Exception as e:
logger.debug(f"Failed to log presence change: {e}")
finally:
conn.close()
def send_alert(person_name: str, status: str, signal_type: str, certainty: float) -> None:
@@ -277,6 +287,7 @@ def update_person_state(person_id: Optional[int]) -> None:
if person_id is None:
return
# Snapshot device state under lock
with devices_lock:
# Get all devices for this person
devices_for_person = [d for d in devices.values() if d.person_id == person_id]
@@ -293,34 +304,42 @@ def update_person_state(person_id: Optional[int]) -> None:
# Get max last_seen time
max_last_seen = max(d.last_seen for d in devices_for_person)
# Determine new status based on snapshot
current_time = time.time()
elapsed_since_signal = (current_time - max_last_seen) / 60.0
with persons_lock:
person = persons.get(person_id)
if person is None:
return
old_status = person.status
elapsed_since_signal = (time.time() - max_last_seen) / 60.0
new_status = old_status
# State machine logic
if person.status == "UNKNOWN":
if agg_certainty >= PRESENCE_THRESHOLD:
person.status = "PRESENT"
person.last_change = time.time()
log_presence_change(person_id, old_status, person.status)
send_alert(person.name, "PRESENT", "signal", agg_certainty)
new_status = "PRESENT"
elif person.status == "PRESENT":
# Check presence anchor
if elapsed_since_signal >= ANCHOR_MINUTES:
person.status = "ABSENT"
person.last_change = time.time()
log_presence_change(person_id, old_status, person.status)
send_alert(person.name, "ABSENT", "timeout", agg_certainty)
new_status = "ABSENT"
elif person.status == "ABSENT":
if agg_certainty >= PRESENCE_THRESHOLD:
person.status = "PRESENT"
person.last_change = time.time()
log_presence_change(person_id, old_status, person.status)
send_alert(person.name, "PRESENT", "signal", agg_certainty)
new_status = "PRESENT"
# Apply state transition if changed
if new_status != old_status:
person.status = new_status
person.last_change = current_time
log_presence_change(person_id, old_status, new_status)
# Send alert outside of lock
if new_status != old_status:
if new_status == "PRESENT":
send_alert(person.name, "PRESENT", "signal", agg_certainty)
else:
send_alert(person.name, "ABSENT", "timeout", agg_certainty)
def probe_sniffer(iface: str) -> None:
@@ -646,13 +665,13 @@ def status_handler(request, client_address, server):
def status_server() -> None:
"""Start HTTP status server on 127.0.0.1:9191."""
"""Start HTTP status server on 127.0.0.1:STATUS_PORT."""
try:
server = http.server.HTTPServer(
("127.0.0.1", 9191),
("127.0.0.1", STATUS_PORT),
status_handler
)
logger.info("Status server started on http://127.0.0.1:9191")
logger.info(f"Status server started on http://127.0.0.1:{STATUS_PORT}")
server.serve_forever()
except Exception as e:
logger.error(f"Failed to start status server: {e}")