vent_ai_platform/app/db/repositories/monitoring.py

from __future__ import annotations

from sqlalchemy import desc, func, select
from sqlalchemy.orm import Session

from app.db.models.monitoring import (
    DataSource,
    Sensor,
    SensorReading,
    VentilationFacility,
)
from app.db.models.work_order import Alarm


def monitoring_overview(db: Session, mine_id: str) -> dict[str, int]:
    return {
        "data_source_count": int(
            db.scalar(
                select(func.count(DataSource.id)).where(DataSource.mine_id == mine_id)
            )
            or 0
        ),
        "online_source_count": int(
            db.scalar(
                select(func.count(DataSource.id)).where(
                    DataSource.mine_id == mine_id, DataSource.status == "online"
                )
            )
            or 0
        ),
        "sensor_count": int(
            db.scalar(select(func.count(Sensor.id)).where(Sensor.mine_id == mine_id)) or 0
        ),
        "online_sensor_count": int(
            db.scalar(
                select(func.count(Sensor.id)).where(
                    Sensor.mine_id == mine_id, Sensor.status == "online"
                )
            )
            or 0
        ),
        "facility_count": int(
            db.scalar(
                select(func.count(VentilationFacility.id)).where(
                    VentilationFacility.mine_id == mine_id
                )
            )
            or 0
        ),
        "abnormal_facility_count": int(
            db.scalar(
                select(func.count(VentilationFacility.id)).where(
                    VentilationFacility.mine_id == mine_id,
                    VentilationFacility.status != "normal",
                )
            )
            or 0
        ),
        "open_alarm_count": int(
            db.scalar(
                select(func.count(Alarm.id)).where(
                    Alarm.mine_id == mine_id, Alarm.status == "open"
                )
            )
            or 0
        ),
        "high_alarm_count": int(
            db.scalar(
                select(func.count(Alarm.id)).where(
                    Alarm.mine_id == mine_id,
                    Alarm.status == "open",
                    Alarm.level == "high",
                )
            )
            or 0
        ),
    }


def list_data_sources(db: Session, mine_id: str) -> list[DataSource]:
    return list(
        db.scalars(
            select(DataSource)
            .where(DataSource.mine_id == mine_id)
            .order_by(DataSource.name)
        ).all()
    )


def list_sensors(
    db: Session,
    mine_id: str,
    sensor_type: str | None = None,
) -> list[Sensor]:
    statement = select(Sensor).where(Sensor.mine_id == mine_id)
    if sensor_type:
        statement = statement.where(Sensor.sensor_type == sensor_type)
    return list(db.scalars(statement.order_by(Sensor.code)).all())


def get_sensor(db: Session, sensor_id: str) -> Sensor | None:
    return db.get(Sensor, sensor_id)


def list_readings(
    db: Session,
    sensor_id: str,
    *,
    limit: int = 200,
) -> list[SensorReading]:
    return list(
        db.scalars(
            select(SensorReading)
            .where(SensorReading.sensor_id == sensor_id)
            .order_by(desc(SensorReading.measured_at))
            .limit(limit)
        ).all()
    )


def list_facilities(db: Session, mine_id: str) -> list[VentilationFacility]:
    return list(
        db.scalars(
            select(VentilationFacility)
            .where(VentilationFacility.mine_id == mine_id)
            .order_by(VentilationFacility.code)
        ).all()
    )


def list_alarms(
    db: Session,
    mine_id: str,
    *,
    status: str | None = None,
    limit: int = 100,
) -> list[Alarm]:
    statement = select(Alarm).where(Alarm.mine_id == mine_id)
    if status:
        statement = statement.where(Alarm.status == status)
    return list(
        db.scalars(statement.order_by(desc(Alarm.occurred_at)).limit(limit)).all()
    )