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