File size: 4,211 Bytes
e391a84 6e84e40 e391a84 6e84e40 9cabacf e391a84 6e84e40 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 104 105 106 107 108 109 110 111 | """
infrastructure/database/models/prediction_model.py
────────────────────────────────────────────────────
PredictionModel — SQLAlchemy ORM model for the ``predictions`` table.
Supabase / PostgreSQL optimisations:
• Native UUID type for the FK ``ppg_signal_id`` (matches ``raw_ppg.id``).
• Dedicated index on ``created_at DESC`` for efficient date-range queries.
• Composite index on ``(ppg_signal_id, created_at)`` for JOIN + sort queries.
Mapping: PredictionModel (ORM) ↔ BPPrediction (domain entity)
"""
from __future__ import annotations
from typing import Any
from sqlalchemy import Float, ForeignKey, Index, String, text
from sqlalchemy.dialects.postgresql import JSONB, UUID
from sqlalchemy.orm import Mapped, mapped_column, relationship
from src.infrastructure.database.models.base import Base
from src.shared.constants import PREDICTIONS_TABLE_NAME, RAW_PPG_TABLE_NAME
class PredictionModel(Base):
"""
ORM representation of an AI blood pressure prediction result.
Table: ``predictions``
"""
__tablename__ = PREDICTIONS_TABLE_NAME
__table_args__ = (
# Optimise date-range queries: "predictions for user X between start and end"
Index(
"ix_predictions_created_at",
text("created_at DESC"),
),
# Optimise JOIN queries: "prediction(s) for a given PPG signal"
Index(
"ix_predictions_signal_created",
"ppg_signal_id",
text("created_at DESC"),
),
)
# ── Foreign key → raw_ppg.id ──────────────────────────────────────────────
ppg_signal_id: Mapped[str] = mapped_column(
UUID(as_uuid=False),
ForeignKey(f"{RAW_PPG_TABLE_NAME}.id", ondelete="CASCADE"),
nullable=False,
index=True,
comment="FK to raw_ppg.id — the source signal for this prediction",
)
# ── Predicted values ──────────────────────────────────────────────────────
predicted_sbp: Mapped[float] = mapped_column(
Float,
nullable=False,
comment="Predicted Systolic Blood Pressure (mmHg)",
)
predicted_dbp: Mapped[float] = mapped_column(
Float,
nullable=False,
comment="Predicted Diastolic Blood Pressure (mmHg)",
)
predicted_ecg: Mapped[list[Any] | None] = mapped_column(
JSONB,
nullable=True,
comment="Synthetic ECG signal windows produced by CardioGAN (list[list[float]])",
)
# ── Model metadata ────────────────────────────────────────────────────────
model_version: Mapped[str] = mapped_column(
String(64),
nullable=False,
comment="Version string of the model that produced this prediction",
)
inference_time_ms: Mapped[float] = mapped_column(
Float,
nullable=False,
default=0.0,
comment="Wall-clock inference duration in milliseconds",
)
sa_log: Mapped[dict | None] = mapped_column(
JSONB,
nullable=True,
comment="Logs of the Simulated Annealing optimization process",
)
# ── Relationship back to PPGModel ─────────────────────────────────────────
ppg_signal: Mapped["PPGModel"] = relationship( # type: ignore[name-defined]
"PPGModel",
back_populates="predictions",
lazy="select",
)
def __repr__(self) -> str:
return (
f"PredictionModel(id={self.id!r}, "
f"ppg_signal_id={self.ppg_signal_id!r}, "
f"SBP={self.predicted_sbp}, DBP={self.predicted_dbp}, "
f"ecg_segments={len(self.predicted_ecg) if isinstance(self.predicted_ecg, list) else 0}, "
f"model={self.model_version!r})"
)
|