7c43e4580d
- FastAPI backend with async SQLAlchemy - Camera RTSP management (add, start, stop) - Vehicle detection (YOLO + fast-alpr) - Type: car, motorcycle, truck, jeepney - Color detection (HSV) - License plate OCR - Motorcycle person count - Face detection + InsightFace ArcFace embedding - pgvector identity grouping (auto-cluster same face) - Vehicle + person movement trail APIs - Docker Compose with pgvector/pg16 - Models: Camera, VehicleIdentity, VehicleEvent, PersonIdentity, FaceEvent
66 lines
2.8 KiB
Python
66 lines
2.8 KiB
Python
from sqlalchemy import Column, String, Integer, Float, DateTime, Text, ForeignKey, JSON
|
|
from sqlalchemy.dialects.postgresql import UUID, ARRAY
|
|
from sqlalchemy.sql import func
|
|
from pgvector.sqlalchemy import Vector
|
|
import uuid
|
|
|
|
from app.core.database import Base
|
|
|
|
|
|
class PersonIdentity(Base):
|
|
"""
|
|
Unique person identity — grouped by face embedding similarity.
|
|
All face events linked to this identity let us trace where this person has been.
|
|
"""
|
|
__tablename__ = "person_identities"
|
|
|
|
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
|
name = Column(String(255), nullable=True) # e.g. "Juan Dela Cruz" if registered
|
|
label = Column(String(100), nullable=True) # e.g. "Unit 3A Resident", "Delivery Rider"
|
|
notes = Column(Text, nullable=True)
|
|
is_registered = Column(Integer, default=0) # 1 = manually registered with name
|
|
is_watchlisted = Column(Integer, default=0) # 1 = flagged for alerts
|
|
thumbnail_path = Column(Text, nullable=True) # Best face shot
|
|
|
|
# Face embedding (512-dim for InsightFace buffalo_l)
|
|
embedding = Column(Vector(512), nullable=True)
|
|
|
|
# Stats
|
|
first_seen_at = Column(DateTime(timezone=True), nullable=True)
|
|
last_seen_at = Column(DateTime(timezone=True), nullable=True)
|
|
total_sightings = Column(Integer, default=0)
|
|
created_at = Column(DateTime(timezone=True), server_default=func.now())
|
|
|
|
def __repr__(self):
|
|
return f"<PersonIdentity {self.name or 'Unknown'} ({self.total_sightings} sightings)>"
|
|
|
|
|
|
class FaceEvent(Base):
|
|
"""
|
|
Every face detection event — one row per detected face per frame.
|
|
"""
|
|
__tablename__ = "face_events"
|
|
|
|
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
|
camera_id = Column(UUID(as_uuid=True), ForeignKey("cameras.id"), nullable=False, index=True)
|
|
identity_id = Column(UUID(as_uuid=True), ForeignKey("person_identities.id"), nullable=True, index=True)
|
|
|
|
# Detection
|
|
detection_confidence = Column(Float, nullable=True)
|
|
face_bbox = Column(JSON, nullable=True) # {"x","y","w","h"}
|
|
face_embedding = Column(Vector(512), nullable=True) # Per-event embedding for re-clustering
|
|
similarity_score = Column(Float, nullable=True) # Match score to identity
|
|
|
|
# Context — if detected alongside a vehicle event
|
|
vehicle_event_id = Column(UUID(as_uuid=True), ForeignKey("vehicle_events.id"), nullable=True)
|
|
|
|
# Storage
|
|
snapshot_path = Column(Text, nullable=True) # Full frame
|
|
face_snapshot_path = Column(Text, nullable=True) # Cropped face chip
|
|
|
|
# Meta
|
|
captured_at = Column(DateTime(timezone=True), server_default=func.now(), index=True)
|
|
|
|
def __repr__(self):
|
|
return f"<FaceEvent identity={self.identity_id} cam={self.camera_id} at={self.captured_at}>"
|