138 lines
4.9 KiB
Python
138 lines
4.9 KiB
Python
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
|