""" Modelo de Cámara para video vigilancia en vehículos. """ from datetime import datetime from sqlalchemy import ( Boolean, DateTime, ForeignKey, Index, Integer, String, Text, ) from sqlalchemy.orm import Mapped, mapped_column, relationship from app.core.database import Base from app.models.base import TimestampMixin class Camara(Base, TimestampMixin): """Modelo de cámara instalada en un vehículo.""" __tablename__ = "camaras" id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True) # Relación con vehículo vehiculo_id: Mapped[int] = mapped_column( ForeignKey("vehiculos.id", ondelete="CASCADE"), nullable=False, index=True, ) # Identificación nombre: Mapped[str] = mapped_column(String(100), nullable=False) posicion: Mapped[str] = mapped_column( String(50), default="frontal", nullable=False, ) # frontal, trasera, interior, lateral_izq, lateral_der # Tipo de cámara tipo: Mapped[str] = mapped_column( String(50), default="ip", nullable=False, ) # ip, dashcam, mdvr, usb # Información del hardware marca: Mapped[str | None] = mapped_column(String(50), nullable=True) modelo: Mapped[str | None] = mapped_column(String(50), nullable=True) numero_serie: Mapped[str | None] = mapped_column(String(100), nullable=True) resolucion: Mapped[str | None] = mapped_column(String(20), nullable=True) # 1080p, 720p, 4K # Conexión de streaming url_stream: Mapped[str | None] = mapped_column(String(500), nullable=True) # URL RTSP/RTMP puerto: Mapped[int | None] = mapped_column(Integer, nullable=True) protocolo: Mapped[str] = mapped_column( String(20), default="rtsp", nullable=False, ) # rtsp, rtmp, hls, webrtc # Autenticación usuario: Mapped[str | None] = mapped_column(String(100), nullable=True) password_encrypted: Mapped[str | None] = mapped_column(String(255), nullable=True) # Configuración de MediaMTX mediamtx_path: Mapped[str | None] = mapped_column(String(100), nullable=True) # Path en MediaMTX # Estado estado: Mapped[str] = mapped_column( String(20), default="desconectada", nullable=False, ) # conectada, desconectada, grabando, error activa: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False) # Última conexión ultima_conexion: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) # Configuración de grabación grabacion_continua: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False) grabacion_evento: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False) # Grabar en eventos duracion_pre_evento: Mapped[int] = mapped_column(Integer, default=10, nullable=False) # Segundos antes duracion_post_evento: Mapped[int] = mapped_column(Integer, default=20, nullable=False) # Segundos después # Detección de eventos (AI/ADAS) deteccion_colision: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False) deteccion_distraccion: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False) deteccion_fatiga: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False) deteccion_cambio_carril: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False) # Notas notas: Mapped[str | None] = mapped_column(Text, nullable=True) # Relaciones ORM vehiculo: Mapped["Vehiculo"] = relationship( "Vehiculo", back_populates="camaras", lazy="selectin", ) grabaciones: Mapped[list["Grabacion"]] = relationship( "Grabacion", back_populates="camara", lazy="dynamic", cascade="all, delete-orphan", ) eventos_video: Mapped[list["EventoVideo"]] = relationship( "EventoVideo", back_populates="camara", lazy="dynamic", cascade="all, delete-orphan", ) # Índices __table_args__ = ( Index("idx_camaras_vehiculo", "vehiculo_id"), Index("idx_camaras_estado", "estado"), ) @property def url_stream_completa(self) -> str | None: """Construye la URL completa de streaming.""" if not self.url_stream: return None if self.usuario and self.password_encrypted: # Desencriptar password y construir URL con autenticación from app.core.security import decrypt_sensitive_data try: password = decrypt_sensitive_data(self.password_encrypted) # Insertar credenciales en URL RTSP if self.url_stream.startswith("rtsp://"): return self.url_stream.replace("rtsp://", f"rtsp://{self.usuario}:{password}@") except Exception: pass return self.url_stream def __repr__(self) -> str: return f""