fleet-platform/app/models/jimi.py

from datetime import UTC, datetime
from typing import Any

from pydantic import AliasChoices, BaseModel, ConfigDict, Field, field_validator


def _coerce_ts(value: Any) -> datetime | None:  # noqa: PLR0911 — many wire formats
    """Tracksolid uses both unix epoch (sec/ms) and ISO/BCD strings.

    Returns a UTC datetime, or None when the value cannot be interpreted.
    """
    if value is None or value == "":
        return None
    if isinstance(value, (int, float)):
        n = int(value)
        if n > 10**12:
            n = n // 1000
        try:
            return datetime.fromtimestamp(n, tz=UTC)
        except (OverflowError, OSError, ValueError):
            return None
    if isinstance(value, str):
        s = value.strip()
        if not s:
            return None
        for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M:%SZ"):
            try:
                return datetime.strptime(s, fmt).replace(tzinfo=UTC)
            except ValueError:
                continue
        if s.isdigit() and len(s) in (12, 14):
            fmt = "%y%m%d%H%M%S" if len(s) == 12 else "%Y%m%d%H%M%S"
            try:
                return datetime.strptime(s, fmt).replace(tzinfo=UTC)
            except ValueError:
                return None
    return None


class _JimiBase(BaseModel):
    model_config = ConfigDict(extra="allow", populate_by_name=True)


# ── Push payloads (P3 push cutover; idle in P1 — Tracksolid still posts to legacy) ──


class JimiPushGps(_JimiBase):
    imei: str = Field(validation_alias=AliasChoices("imei", "deviceImei"))
    gps_time: datetime = Field(validation_alias=AliasChoices("gpsTime", "gps_time"))
    lat: float
    lng: float
    speed_kmh: float | None = Field(
        default=None, validation_alias=AliasChoices("gpsSpeed", "speed")
    )
    direction_deg: float | None = Field(default=None, validation_alias="direction")
    acc: int | str | None = Field(default=None, validation_alias=AliasChoices("acc", "accStatus"))
    satellites: int | None = Field(
        default=None, validation_alias=AliasChoices("satelliteNum", "gpsNum")
    )
    altitude_m: float | None = Field(default=None, validation_alias="altitude")
    post_type: int | str | None = Field(default=None, validation_alias="postType")

    @field_validator("gps_time", mode="before")
    @classmethod
    def _parse_time(cls, v: Any) -> Any:
        parsed = _coerce_ts(v)
        if parsed is None:
            raise ValueError(f"unparseable gpsTime: {v!r}")
        return parsed


class JimiPushAlarm(_JimiBase):
    imei: str
    alarm_type: str = Field(validation_alias="alarmType")
    alarm_name: str | None = Field(default=None, validation_alias="alarmName")
    alarm_time: datetime = Field(validation_alias="alarmTime")
    lat: float | None = None
    lng: float | None = None
    speed_kmh: float | None = Field(default=None, validation_alias="speed")
    device_name: str | None = Field(default=None, validation_alias="deviceName")

    @field_validator("alarm_time", mode="before")
    @classmethod
    def _parse_time(cls, v: Any) -> Any:
        parsed = _coerce_ts(v)
        if parsed is None:
            raise ValueError(f"unparseable alarmTime: {v!r}")
        return parsed


class JimiPushHeartbeat(_JimiBase):
    imei: str = Field(validation_alias=AliasChoices("imei", "deviceImei"))
    gate_time: datetime = Field(validation_alias=AliasChoices("gateTime", "hbTime"))
    power_level: int | None = Field(
        default=None, validation_alias=AliasChoices("powerLevel", "electQuantity")
    )
    gsm_signal: int | None = Field(default=None, validation_alias="gsmSign")
    acc: int | None = None
    power_status: int | None = Field(default=None, validation_alias="powerStatus")

    @field_validator("gate_time", mode="before")
    @classmethod
    def _parse_time(cls, v: Any) -> Any:
        parsed = _coerce_ts(v)
        if parsed is None:
            raise ValueError(f"unparseable gateTime: {v!r}")
        return parsed


class JimiPushEvent(_JimiBase):
    imei: str = Field(validation_alias=AliasChoices("imei", "deviceImei"))
    event_type: str = Field(validation_alias="type")
    event_time: datetime = Field(validation_alias="gateTime")
    timezone_str: str | None = Field(default=None, validation_alias="timezone")

    @field_validator("event_time", mode="before")
    @classmethod
    def _parse_time(cls, v: Any) -> Any:
        parsed = _coerce_ts(v)
        if parsed is None:
            raise ValueError(f"unparseable gateTime: {v!r}")
        return parsed


# ── Polled responses (Tracksolid Pro API — P1 ingest path) ──


class JimiPollFix(_JimiBase):
    """One device's fix from jimi.user.device.location.list / jimi.device.location.get.

    Tracksolid Pro returns null fields for offline devices (no recent fix). We
    keep gps_time, lat, lng optional and let the parser skip items that don't
    have a usable fix.
    """

    imei: str = Field(validation_alias=AliasChoices("imei", "deviceImei"))
    gps_time: datetime | None = Field(
        default=None, validation_alias=AliasChoices("gpsTime", "gps_time")
    )
    lat: float | None = None
    lng: float | None = None
    speed_kmh: float | None = Field(
        default=None, validation_alias=AliasChoices("speed", "gpsSpeed")
    )
    direction_deg: float | None = Field(default=None, validation_alias="direction")
    altitude_m: float | None = Field(default=None, validation_alias="altitude")
    satellites: int | None = Field(
        default=None, validation_alias=AliasChoices("gpsNum", "satelliteNum")
    )
    acc: int | str | None = Field(default=None, validation_alias=AliasChoices("accStatus", "acc"))
    mc_type: str | None = Field(default=None, validation_alias="mcType")
    pos_type: str | int | None = Field(default=None, validation_alias="posType")
    device_name: str | None = Field(default=None, validation_alias="deviceName")

    @field_validator("gps_time", mode="before")
    @classmethod
    def _parse_time(cls, v: Any) -> Any:
        return _coerce_ts(v)  # returns None for unparseable; field is Optional