289 lines
9.9 KiB
Python
289 lines
9.9 KiB
Python
import math
|
|
from datetime import datetime
|
|
|
|
|
|
INACTIVITY_SECONDS = 5 * 60
|
|
GPS_MOTION_METERS = 8
|
|
MIN_GPS_MOTION_SAMPLES = 3
|
|
MAX_GPS_SAMPLE_GAP_SECONDS = 30
|
|
MAX_PLAUSIBLE_SPEED_KMH = 220
|
|
EARTH_RADIUS_KM = 6371.0088
|
|
|
|
|
|
def _number(value):
|
|
if value is None:
|
|
return None
|
|
try:
|
|
return float(str(value).replace("\x00", "").strip())
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _timestamp(value):
|
|
try:
|
|
return datetime.fromisoformat(str(value))
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def haversine_km(left, right):
|
|
lat1, lon1 = map(math.radians, left)
|
|
lat2, lon2 = map(math.radians, right)
|
|
latitude_delta = lat2 - lat1
|
|
longitude_delta = lon2 - lon1
|
|
value = (
|
|
math.sin(latitude_delta / 2) ** 2
|
|
+ math.cos(lat1) * math.cos(lat2)
|
|
* math.sin(longitude_delta / 2) ** 2
|
|
)
|
|
return EARTH_RADIUS_KM * 2 * math.atan2(math.sqrt(value), math.sqrt(1 - value))
|
|
|
|
|
|
def normalize_tracking_rows(rows):
|
|
normalized = []
|
|
for row in rows:
|
|
timestamp = _timestamp(row[1])
|
|
latitude = _number(row[2])
|
|
longitude = _number(row[3])
|
|
speed = _number(row[4]) or 0
|
|
odometer = _number(row[5]) if len(row) > 5 else None
|
|
if timestamp is None:
|
|
continue
|
|
if latitude is not None and not (-90 <= latitude <= 90):
|
|
continue
|
|
if longitude is not None and not (-180 <= longitude <= 180):
|
|
continue
|
|
normalized.append({
|
|
"id": int(row[0]),
|
|
"datetime": timestamp,
|
|
"latitude": latitude,
|
|
"longitude": longitude,
|
|
"speed": max(0, speed),
|
|
"odometer": odometer,
|
|
})
|
|
return normalized
|
|
|
|
|
|
def _segment(row, previous):
|
|
seconds = (row["datetime"] - previous["datetime"]).total_seconds()
|
|
if seconds <= 0:
|
|
return 0, 0
|
|
if (
|
|
previous["latitude"] is None or previous["longitude"] is None
|
|
or row["latitude"] is None or row["longitude"] is None
|
|
):
|
|
return 0, 0
|
|
distance_km = haversine_km(
|
|
(previous["latitude"], previous["longitude"]),
|
|
(row["latitude"], row["longitude"]),
|
|
)
|
|
speed_kmh = distance_km / seconds * 3600
|
|
if speed_kmh > MAX_PLAUSIBLE_SPEED_KMH:
|
|
return 0, speed_kmh
|
|
return distance_km, speed_kmh
|
|
|
|
|
|
def build_trip_summaries(raw_rows):
|
|
rows = normalize_tracking_rows(raw_rows)
|
|
if len(rows) < 2:
|
|
return []
|
|
|
|
# Don't generate trips from GPS drift when the car was never connected.
|
|
# Without OBD, GPS jitter can produce false motion events on the bench.
|
|
from modules import app_obd
|
|
if not app_obd.was_car_ever_connected():
|
|
return []
|
|
|
|
obd_motion = []
|
|
gps_motion = []
|
|
for index in range(1, len(rows)):
|
|
seconds = (
|
|
rows[index]["datetime"] - rows[index - 1]["datetime"]
|
|
).total_seconds()
|
|
distance_km, _ = _segment(rows[index], rows[index - 1])
|
|
has_obd_motion = rows[index]["speed"] > 1
|
|
has_gps_motion = (
|
|
0 < seconds <= MAX_GPS_SAMPLE_GAP_SECONDS
|
|
and distance_km * 1000 >= GPS_MOTION_METERS
|
|
)
|
|
if has_obd_motion:
|
|
obd_motion.append({
|
|
"index": index,
|
|
"obd": True,
|
|
"distance_km": distance_km,
|
|
})
|
|
if has_gps_motion:
|
|
gps_motion.append({
|
|
"index": index,
|
|
"obd": False,
|
|
"distance_km": distance_km,
|
|
})
|
|
|
|
qualified_gps_motion = []
|
|
gps_sequences = []
|
|
for event in gps_motion:
|
|
if (
|
|
not gps_sequences
|
|
or (
|
|
rows[event["index"]]["datetime"]
|
|
- rows[gps_sequences[-1][-1]["index"]]["datetime"]
|
|
).total_seconds() > MAX_GPS_SAMPLE_GAP_SECONDS
|
|
):
|
|
gps_sequences.append([event])
|
|
else:
|
|
gps_sequences[-1].append(event)
|
|
for sequence in gps_sequences:
|
|
if len(sequence) >= MIN_GPS_MOTION_SAMPLES:
|
|
qualified_gps_motion.extend(sequence)
|
|
|
|
motion_by_index = {}
|
|
for event in obd_motion + qualified_gps_motion:
|
|
existing = motion_by_index.get(event["index"])
|
|
if existing is None:
|
|
motion_by_index[event["index"]] = event.copy()
|
|
else:
|
|
existing["obd"] = existing["obd"] or event["obd"]
|
|
existing["distance_km"] = max(
|
|
existing["distance_km"], event["distance_km"]
|
|
)
|
|
motion = [
|
|
motion_by_index[index] for index in sorted(motion_by_index)
|
|
]
|
|
|
|
groups = []
|
|
for event in motion:
|
|
event_gap = (
|
|
(
|
|
rows[event["index"]]["datetime"]
|
|
- rows[groups[-1][-1]["index"]]["datetime"]
|
|
).total_seconds()
|
|
if groups
|
|
else None
|
|
)
|
|
if (
|
|
not groups
|
|
or event_gap <= 0
|
|
or event_gap > INACTIVITY_SECONDS
|
|
):
|
|
groups.append([event])
|
|
else:
|
|
groups[-1].append(event)
|
|
|
|
trips = []
|
|
for events in groups:
|
|
if not any(event["obd"] for event in events) and len(events) < MIN_GPS_MOTION_SAMPLES:
|
|
continue
|
|
|
|
start_index = max(0, events[0]["index"] - 1)
|
|
end_index = events[-1]["index"]
|
|
trip_rows = rows[start_index:end_index + 1]
|
|
distance_km = 0
|
|
segment_speeds = []
|
|
obd_speeds = []
|
|
has_obd_motion = any(event["obd"] for event in events)
|
|
for previous, row in zip(trip_rows, trip_rows[1:]):
|
|
segment_km, segment_speed = _segment(row, previous)
|
|
minimum_distance_meters = 1 if has_obd_motion else GPS_MOTION_METERS
|
|
if segment_km * 1000 >= minimum_distance_meters:
|
|
distance_km += segment_km
|
|
segment_speeds.append(segment_speed)
|
|
if row["speed"] > 0:
|
|
obd_speeds.append(row["speed"])
|
|
|
|
start = trip_rows[0]
|
|
end = trip_rows[-1]
|
|
duration_seconds = max(
|
|
0, int((end["datetime"] - start["datetime"]).total_seconds())
|
|
)
|
|
if not any(event["obd"] for event in events) and duration_seconds < 10:
|
|
continue
|
|
|
|
# Speed-only rows (logged before the first GPS fix) have NULL coords.
|
|
# Use the first/last row that actually has a position for start/end GPS,
|
|
# and skip a trip that never got any fix at all (no location to record).
|
|
gps_rows = [
|
|
r for r in trip_rows
|
|
if r["latitude"] is not None and r["longitude"] is not None
|
|
]
|
|
if not gps_rows:
|
|
continue
|
|
start_gps = gps_rows[0]
|
|
end_gps = gps_rows[-1]
|
|
|
|
trips.append({
|
|
"id": start["id"],
|
|
"start_tracking_id": start["id"],
|
|
"end_tracking_id": end["id"],
|
|
"start_datetime": start["datetime"].isoformat(sep=" "),
|
|
"end_datetime": end["datetime"].isoformat(sep=" "),
|
|
"start_gpsx": start_gps["latitude"],
|
|
"start_gpsy": start_gps["longitude"],
|
|
"end_gpsx": end_gps["latitude"],
|
|
"end_gpsy": end_gps["longitude"],
|
|
"distance_km": round(distance_km, 3),
|
|
"duration_seconds": duration_seconds,
|
|
"point_count": len(trip_rows),
|
|
"average_speed_kmh": round(
|
|
distance_km / duration_seconds * 3600, 1
|
|
) if duration_seconds else 0,
|
|
"max_speed_kmh": round(
|
|
max((obd_speeds + segment_speeds) or [0]), 1
|
|
),
|
|
"start_odometer": next((r["odometer"] for r in trip_rows if r.get("odometer")), None),
|
|
"end_odometer": next((r["odometer"] for r in reversed(trip_rows) if r.get("odometer")), None),
|
|
})
|
|
return trips
|
|
|
|
|
|
def route_points(raw_rows, max_points=2000, odometer_baseline=0.0, prior_km=0.0):
|
|
rows = normalize_tracking_rows(raw_rows)
|
|
# The route polyline needs coordinates — drop speed-only rows logged before
|
|
# the first GPS fix (no position to draw; they add 0 distance anyway).
|
|
rows = [r for r in rows if r["latitude"] is not None and r["longitude"] is not None]
|
|
filtered_rows = []
|
|
for row in rows:
|
|
if filtered_rows:
|
|
_, segment_speed = _segment(row, filtered_rows[-1])
|
|
if segment_speed > MAX_PLAUSIBLE_SPEED_KMH:
|
|
continue
|
|
filtered_rows.append(row)
|
|
rows = filtered_rows
|
|
cumulative_km = 0.0
|
|
for index, row in enumerate(rows):
|
|
row["speed_source"] = "obd" if row["speed"] > 0 else "gps"
|
|
if row["speed"] <= 0 and index:
|
|
distance_km, segment_speed = _segment(row, rows[index - 1])
|
|
row["speed"] = (
|
|
segment_speed
|
|
if distance_km * 1000 >= GPS_MOTION_METERS
|
|
else 0
|
|
)
|
|
if index:
|
|
segment_km, _ = _segment(row, rows[index - 1])
|
|
cumulative_km += segment_km
|
|
# Odometer per point: use the raw OBD value when the vehicle reports
|
|
# one; otherwise simulate it as baseline + prior trips + within-trip
|
|
# distance (continuous across trips, matching get_overview_stats).
|
|
raw_odometer = row.get("odometer")
|
|
if raw_odometer is not None and raw_odometer > 0:
|
|
row["odometer"] = raw_odometer
|
|
else:
|
|
row["odometer"] = odometer_baseline + prior_km + cumulative_km
|
|
if len(rows) <= max_points:
|
|
selected = rows
|
|
else:
|
|
indexes = {
|
|
round(index * (len(rows) - 1) / (max_points - 1))
|
|
for index in range(max_points)
|
|
}
|
|
selected = [row for index, row in enumerate(rows) if index in indexes]
|
|
return [{
|
|
"id": row["id"],
|
|
"datetime": row["datetime"].isoformat(sep=" "),
|
|
"latitude": row["latitude"],
|
|
"longitude": row["longitude"],
|
|
"speed": round(row["speed"], 1),
|
|
"speed_source": row["speed_source"],
|
|
"odometer": round(row["odometer"], 1) if row["odometer"] is not None else None,
|
|
} for row in selected]
|