""" infrastructure/database/models/ppg_model.py ───────────────────────────────────────────── PPGModel — SQLAlchemy ORM model for the ``raw_ppg`` table. Supabase / PostgreSQL optimisations: • Native UUID type for ``id`` (inherited) and FK ``ppg_signal_id``. • Composite index on ``(user_id, timestamp DESC)`` — accelerates the most common query pattern: "latest signals for a user". • JSONB instead of JSON for ``ppg_values`` — Supabase PostgreSQL stores JSONB in binary format with native indexing support. Mapping: PPGModel (ORM) ↔ PPGSignal (domain entity) Conversion is handled by the repository (_to_entity / _to_model). """ from __future__ import annotations from datetime import datetime from sqlalchemy import DateTime, Float, Index, String, text from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.orm import Mapped, mapped_column, relationship from src.infrastructure.database.models.base import Base from src.shared.constants import RAW_PPG_TABLE_NAME class PPGModel(Base): """ ORM representation of a raw PPG signal recording. Table: ``raw_ppg`` """ __tablename__ = RAW_PPG_TABLE_NAME # Composite + single-column indexes declared here for Supabase/PostgreSQL. # SQLAlchemy will emit the correct CREATE INDEX statements via Alembic. __table_args__ = ( # Composite index: most queries filter by user_id then sort by timestamp Index( "ix_raw_ppg_user_timestamp", "user_id", text("timestamp DESC"), ), # Composite index: device queries (e.g. listing all signals from a device) Index( "ix_raw_ppg_device_timestamp", "device_id", text("timestamp DESC"), ), ) # ── Business columns ────────────────────────────────────────────────────── device_id: Mapped[str] = mapped_column( String(128), nullable=False, index=True, comment="Unique identifier of the IoT sensor/device", ) user_id: Mapped[str] = mapped_column( String(128), nullable=False, index=True, comment="Unique identifier of the patient/user", ) timestamp: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, comment="UTC datetime of signal capture (set by the IoT device)", ) sampling_rate: Mapped[float] = mapped_column( Float, nullable=False, comment="Sampling rate in Hz (e.g. 125.0)", ) ppg_values: Mapped[list] = mapped_column( JSONB, # JSONB: binary JSON with indexing on Supabase nullable=False, comment="JSONB array of raw PPG amplitude values", ) duration_seconds: Mapped[float] = mapped_column( Float, nullable=False, comment="Duration of the recording in seconds", ) # ── Relationships ───────────────────────────────────────────────────────── predictions: Mapped[list["PredictionModel"]] = relationship( # type: ignore[name-defined] "PredictionModel", back_populates="ppg_signal", cascade="all, delete-orphan", lazy="select", ) def __repr__(self) -> str: return ( f"PPGModel(id={self.id!r}, device={self.device_id!r}, " f"user={self.user_id!r}, samples={len(self.ppg_values or [])})" )