fleet-platform/app/models/jimi.py

from datetime import UTC, datetime
from typing import Any

from pydantic import 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)


class JimiPushGps(_JimiBase):
    imei: str = Field(validation_alias="deviceImei")
    gps_time: datetime = Field(validation_alias="gpsTime")
    lat: float
    lng: float
    speed_kmh: float | None = Field(default=None, validation_alias="gpsSpeed")
    direction_deg: float | None = Field(default=None, validation_alias="direction")
    acc: int | str | None = None
    satellites: int | None = Field(default=None, validation_alias="satelliteNum")
    altitude_m: float | None = Field(default=None, validation_alias="altitude")
    post_type: int | 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

    @field_validator("imei", mode="before")
    @classmethod
    def _resolve_imei(cls, v: Any, info: Any) -> Any:
        return v


class JimiPushHeartbeat(_JimiBase):
    imei: str = Field(validation_alias="deviceImei")
    gate_time: datetime = Field(validation_alias="gateTime")
    power_level: int | None = Field(default=None, validation_alias="powerLevel")
    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="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


class JimiPollFix(_JimiBase):
    imei: str = Field(validation_alias="deviceImei")
    gps_time: datetime = Field(validation_alias="gpsTime")
    lat: float
    lng: float
    speed_kmh: float | None = Field(default=None, validation_alias="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="satelliteNum")
    acc: int | str | None = None

    @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