File size: 3,742 Bytes
e391a84 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 | """
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 [])})"
)
|