Kunpeng
25fa0c5825
Backend:
- Persist alert events and robot delivery counters in SQLite (data/app.db,
YOLO_DB_PATH override); EventStore keeps its interface, DeliveryStatsStore
uses read/write-through UPSERT; in-memory fallback remains for standalone
AlertManager use
- Serve /api/detect from a sync endpoint and serialize YOLO inference with a
lock, so concurrent requests no longer block the event loop
- Unify single-image and session detection: requests without session_id
share the fixed single-image session and run full frame confirmation
- Store UTC Z-suffixed timestamps for sortable string comparison; close
executor and database on shutdown via FastAPI lifespan
Frontend:
- Split app.js into ES modules (main.js + modules/{dom,api,ui,charts,store}
+ views/{dashboard,inspection,events,robots,robotCards}), no build step
- Escape all server data interpolated into innerHTML; guard missing
event.classes; replace lazy element ID list with memoized qs()
- Remove hardcoded fake stats (trend badge, device donut segment)
Docs:
- Rewrite README: accurate weight policy (only trained best.pt committed,
no Git LFS), SQLite persistence, single-worker note, data asset inventory
- Align models/pretrained/README.md with actual files; add YOLO_DB_PATH to
.env.example; add persistence unit tests; ruff clean
85 lines
2.7 KiB
Python
85 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from backend.events import EventStore
|
|
from backend.storage import Database
|
|
|
|
|
|
class EventStoreTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.store = EventStore(max_events=2)
|
|
|
|
def create_event(self, session_id: str = "session") -> dict:
|
|
return self.store.create(
|
|
session_id=session_id,
|
|
classes=["fire"],
|
|
detections=[{"class": "fire", "confidence": 0.82}],
|
|
notification_channels={"wechat": True, "feishu": False},
|
|
)
|
|
|
|
def test_create_and_summarize_event(self) -> None:
|
|
event = self.create_event()
|
|
|
|
summary = self.store.summary()
|
|
|
|
self.assertEqual(event["status"], "pending")
|
|
self.assertEqual(event["max_confidence"], 0.82)
|
|
self.assertEqual(summary["total"], 1)
|
|
self.assertEqual(summary["pending"], 1)
|
|
self.assertEqual(summary["fire"], 1)
|
|
|
|
def test_update_event_status(self) -> None:
|
|
event = self.create_event()
|
|
|
|
updated = self.store.update(event["id"], "resolved")
|
|
|
|
self.assertIsNotNone(updated)
|
|
self.assertEqual(updated["status"], "resolved")
|
|
self.assertIsNotNone(updated["handled_at"])
|
|
|
|
def test_store_respects_maximum_size(self) -> None:
|
|
first = self.create_event("first")
|
|
self.create_event("second")
|
|
self.create_event("third")
|
|
|
|
event_ids = {event["id"] for event in self.store.list()}
|
|
|
|
self.assertEqual(len(event_ids), 2)
|
|
self.assertNotIn(first["id"], event_ids)
|
|
|
|
def test_invalid_status_is_rejected(self) -> None:
|
|
event = self.create_event()
|
|
|
|
with self.assertRaisesRegex(ValueError, "Unsupported"):
|
|
self.store.update(event["id"], "invalid")
|
|
|
|
def test_events_persist_across_store_instances(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
db_path = Path(tmp_dir) / "app.db"
|
|
first_db = Database(db_path)
|
|
first = EventStore(db=first_db)
|
|
first.create(
|
|
session_id="session",
|
|
classes=["smoke"],
|
|
detections=[{"class": "smoke", "confidence": 0.55}],
|
|
notification_channels={"wechat": False, "feishu": False},
|
|
)
|
|
first_db.close()
|
|
|
|
second_db = Database(db_path)
|
|
second = EventStore(db=second_db)
|
|
events = second.list()
|
|
summary_total = second.summary()["total"]
|
|
second_db.close()
|
|
|
|
self.assertEqual(len(events), 1)
|
|
self.assertEqual(events[0]["classes"], ["smoke"])
|
|
self.assertEqual(events[0]["session_id"], "session")
|
|
self.assertEqual(summary_total, 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |