- 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
74 lines
3.0 KiB
Python
74 lines
3.0 KiB
Python
from sqlalchemy import Column, String, Integer, Float, DateTime, Text, ForeignKey, Enum, JSON
|
|
from sqlalchemy.dialects.postgresql import UUID
|
|
from sqlalchemy.sql import func
|
|
import uuid
|
|
import enum
|
|
|
|
from app.core.database import Base
|
|
|
|
|
|
class VehicleType(str, enum.Enum):
|
|
car = "car"
|
|
motorcycle = "motorcycle"
|
|
truck = "truck"
|
|
van = "van"
|
|
jeepney = "jeepney"
|
|
tricycle = "tricycle"
|
|
unknown = "unknown"
|
|
|
|
|
|
class VehicleIdentity(Base):
|
|
"""
|
|
Unique vehicle identity — grouped by license plate.
|
|
Tracks every time this vehicle was seen across any camera.
|
|
"""
|
|
__tablename__ = "vehicle_identities"
|
|
|
|
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
|
plate_number = Column(String(20), nullable=True, index=True, unique=True)
|
|
vehicle_type = Column(Enum(VehicleType), default=VehicleType.unknown)
|
|
color = Column(String(50), nullable=True)
|
|
make_model = Column(String(100), nullable=True) # e.g. "Toyota Vios"
|
|
notes = Column(Text, nullable=True)
|
|
label = Column(String(100), nullable=True) # e.g. "Unit 4B Owner", "Delivery Van"
|
|
is_whitelisted = Column(Integer, default=0) # 1 = resident/approved
|
|
is_blacklisted = Column(Integer, default=0) # 1 = flagged/banned
|
|
thumbnail_path = Column(Text, nullable=True) # Best shot saved
|
|
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())
|
|
|
|
|
|
class VehicleEvent(Base):
|
|
"""
|
|
Every detection event — one row per camera capture.
|
|
"""
|
|
__tablename__ = "vehicle_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("vehicle_identities.id"), nullable=True, index=True)
|
|
|
|
# Detection data
|
|
plate_number = Column(String(20), nullable=True)
|
|
plate_confidence = Column(Float, nullable=True)
|
|
vehicle_type = Column(Enum(VehicleType), default=VehicleType.unknown)
|
|
color = Column(String(50), nullable=True)
|
|
person_count = Column(Integer, nullable=True) # For motorcycle: how many riders
|
|
detection_confidence = Column(Float, nullable=True)
|
|
|
|
# Bounding boxes (stored as JSON: {"x":0,"y":0,"w":100,"h":100})
|
|
vehicle_bbox = Column(JSON, nullable=True)
|
|
plate_bbox = Column(JSON, nullable=True)
|
|
|
|
# Storage
|
|
snapshot_path = Column(Text, nullable=True) # Full frame snapshot
|
|
plate_snapshot_path = Column(Text, nullable=True) # Cropped plate image
|
|
|
|
# Meta
|
|
captured_at = Column(DateTime(timezone=True), server_default=func.now(), index=True)
|
|
|
|
def __repr__(self):
|
|
return f"<VehicleEvent plate={self.plate_number} type={self.vehicle_type} at={self.captured_at}>"
|