home-automation/app/services/location.py

from datetime import datetime, timezone

from sqlalchemy import delete, insert, select
from sqlalchemy.orm import Session

from app.models.location import Location
from app.schemas.location import LocationRecordRequest


def _parse_optional_float_compat(value: str | None) -> float:
    try:
        return float(value)
    except (TypeError, ValueError):
        return 0.0


def _parse_required_float(value: str, field_name: str) -> float:
    try:
        return float(value)
    except (TypeError, ValueError) as exc:
        raise ValueError(f"Invalid numeric value for {field_name}") from exc


def _utc_now_rfc3339() -> str:
    now = datetime.now(timezone.utc).replace(microsecond=0)
    return now.isoformat().replace("+00:00", "Z")


def record_location(session: Session, payload: LocationRecordRequest) -> None:
    stmt = (
        insert(Location)
        .prefix_with("OR IGNORE")
        .values(
            person=payload.person,
            datetime=_utc_now_rfc3339(),
            latitude=_parse_required_float(payload.latitude, "latitude"),
            longitude=_parse_required_float(payload.longitude, "longitude"),
            altitude=_parse_optional_float_compat(payload.altitude),
        )
    )
    session.execute(stmt)
    session.commit()


def update_location(
    session: Session,
    person: str,
    datetime_pk: str,
    *,
    latitude: float | None,
    longitude: float | None,
    altitude: float | None,
) -> Location | None:
    """Update non-PK fields of a single location row.

    Returns the updated ORM object, or ``None`` if the PK does not exist.
    The caller must not pass PK fields — they are immutable.
    Only fields with a non-``None`` value are written; ``altitude`` being
    ``None`` in the request means "leave unchanged", not "clear to NULL".
    """
    row = session.execute(
        select(Location).where(
            Location.person == person,
            Location.datetime == datetime_pk,
        )
    ).scalar_one_or_none()

    if row is None:
        return None

    if latitude is not None:
        row.latitude = latitude
    if longitude is not None:
        row.longitude = longitude
    if altitude is not None:
        row.altitude = altitude

    session.commit()
    session.refresh(row)
    return row


def delete_location(session: Session, person: str, datetime_pk: str) -> bool:
    """Delete the single location row identified by its full composite PK.

    Returns ``True`` if exactly one row was deleted, ``False`` if the PK did
    not exist (caller should raise 404).  The DELETE is scoped to the exact PK
    — no batch/truncate path exists.
    """
    result = session.execute(
        delete(Location).where(
            Location.person == person,
            Location.datetime == datetime_pk,
        )
    )
    session.commit()
    return result.rowcount == 1