Implement POLL-01 high-res GPS trails and POLL-03 on-demand location refresh
POLL-01 (FIX-M14): Add poll_track_list() calling jimi.device.track.list - Runs every 30 min with 35-min lookback window (5-min overlap prevents gaps) - Inserts all device waypoints into position_history with source='track_list' - Increases position density from ~1/min to 2-6 fixes/min per active vehicle - Single shared DB connection for all devices per cycle (efficient) POLL-03 (FIX-M15): Add get_device_locations() utility function - Calls jimi.device.location.get for up to 50 specific IMEIs on demand - Used for alarm enrichment, stale device recovery, dashboard precision refresh Manual updates: - position_history section rewritten to document dual ingestion sources - Three new queries: data density check, harsh driving detection, route trace - Known Data Issues: issues 10 and 11 added and marked Fixed - API coverage table updated to reflect all three endpoints now in use Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
David Kiania
parent
d534aceadc
commit
3797a4e2ca
3 changed files with 254 additions and 7 deletions
|
|
@ -21,6 +21,15 @@ REVISIONS (QA-Verified):
|
|||
[FIX-M13] POLL-02: Parking poll was returning 0 rows — added missing
|
||||
acc_type=0 and account params; fixed response field durSecond
|
||||
(was mapped as 'seconds').
|
||||
[FIX-M14] POLL-01: New poll_track_list() — calls jimi.device.track.list
|
||||
per device every 30 minutes to capture high-resolution GPS
|
||||
waypoints between the 60-second fleet sweep snapshots. Writes to
|
||||
position_history with source='track_list'. Fills gaps in route
|
||||
reconstruction and enables accurate path drawing on maps.
|
||||
[FIX-M15] POLL-03: New get_device_locations() utility — calls
|
||||
jimi.device.location.get for up to 50 specific IMEIs on demand.
|
||||
Used for precision refreshes (alarm enrichment, stale device
|
||||
recovery) without waiting for the next full fleet sweep.
|
||||
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
|
||||
"""
|
||||
|
||||
|
|
@ -258,21 +267,173 @@ def poll_parking():
|
|||
int((time.time() - t0) * 1000), True)
|
||||
log.info("Parking: %d events processed.", inserted)
|
||||
|
||||
# ── 5. High-Resolution GPS Trail (Every 30m) — POLL-01 ───────────────────────
|
||||
|
||||
def poll_track_list():
|
||||
"""[FIX-M14] Fetch per-device GPS waypoint trails via jimi.device.track.list.
|
||||
|
||||
The 60-second fleet sweep (poll_live_positions) captures only the most
|
||||
recent fix per vehicle — all motion between sweeps is invisible. This
|
||||
function retrieves every waypoint the device logged in the last 35 minutes
|
||||
(5-min overlap ensures no gaps at scheduling boundaries) and inserts them
|
||||
into position_history with source='track_list'.
|
||||
|
||||
Impact on reporting:
|
||||
- position_history row density increases from ~1/min to ~1–4/min per device
|
||||
- Route traces in Grafana become accurate continuous paths
|
||||
- Speed profile queries gain meaningful resolution (avg over 10s intervals
|
||||
vs 60s intervals) — enables hard-braking / harsh-acceleration detection
|
||||
- v_mileage_daily_cagg continuous aggregate gains finer odometer deltas
|
||||
"""
|
||||
t0 = time.time()
|
||||
token, imeis = get_token(), get_active_imeis()
|
||||
if not token or not imeis:
|
||||
return
|
||||
|
||||
end_ts = datetime.now(timezone.utc)
|
||||
start_ts = end_ts - timedelta(minutes=35) # 5-min overlap avoids boundary gaps
|
||||
total_inserted = 0
|
||||
devices_with_data = 0
|
||||
|
||||
with get_conn() as conn:
|
||||
with conn.cursor() as cur:
|
||||
for imei in imeis:
|
||||
resp = api_post("jimi.device.track.list", {
|
||||
"imei": imei,
|
||||
"begin_time": start_ts.strftime("%Y-%m-%d %H:%M:%S"),
|
||||
"end_time": end_ts.strftime("%Y-%m-%d %H:%M:%S"),
|
||||
"map_type": "GOOGLE",
|
||||
}, token)
|
||||
|
||||
waypoints = resp.get("result") or []
|
||||
if not waypoints:
|
||||
continue
|
||||
|
||||
inserted = 0
|
||||
for wp in waypoints:
|
||||
lat = clean_num(wp.get("lat"))
|
||||
lng = clean_num(wp.get("lng"))
|
||||
gps_time = clean_ts(wp.get("gpsTime"))
|
||||
if not is_valid_fix(lat, lng) or not gps_time:
|
||||
continue
|
||||
|
||||
cur.execute("""
|
||||
INSERT INTO tracksolid.position_history (
|
||||
imei, gps_time, geom, lat, lng,
|
||||
speed, direction, acc_status, source
|
||||
) VALUES (
|
||||
%s, %s,
|
||||
ST_SetSRID(ST_MakePoint(%s, %s), 4326),
|
||||
%s, %s, %s, %s, %s, 'track_list'
|
||||
)
|
||||
ON CONFLICT (imei, gps_time) DO NOTHING
|
||||
""", (
|
||||
imei, gps_time,
|
||||
lng, lat, # ST_MakePoint(lng, lat)
|
||||
lat, lng, # lat, lng columns
|
||||
clean_num(wp.get("gpsSpeed")),
|
||||
clean_num(wp.get("direction")),
|
||||
clean(wp.get("accStatus")),
|
||||
))
|
||||
inserted += 1
|
||||
|
||||
if inserted:
|
||||
total_inserted += inserted
|
||||
devices_with_data += 1
|
||||
|
||||
log_ingestion(cur, "jimi.device.track.list", len(imeis),
|
||||
0, total_inserted, int((time.time() - t0) * 1000), True)
|
||||
conn.commit()
|
||||
|
||||
log.info("Track list: %d waypoints inserted across %d/%d devices.",
|
||||
total_inserted, devices_with_data, len(imeis))
|
||||
|
||||
|
||||
# ── 6. On-Demand Device Location Refresh — POLL-03 ───────────────────────────
|
||||
|
||||
def get_device_locations(imeis: list) -> int:
|
||||
"""[FIX-M15] Precision position refresh for a specific list of IMEIs.
|
||||
|
||||
Calls jimi.device.location.get (up to 50 IMEIs per call) and upserts
|
||||
results into live_positions. Use this for:
|
||||
- Alarm enrichment: get exact position immediately after an alarm fires
|
||||
- Stale device recovery: force-refresh a vehicle that has been offline
|
||||
- Dashboard on-demand refresh without waiting for the 60s fleet sweep
|
||||
|
||||
Returns the number of positions successfully upserted.
|
||||
"""
|
||||
token = get_token()
|
||||
if not token or not imeis:
|
||||
return 0
|
||||
|
||||
upserted = 0
|
||||
with get_conn() as conn:
|
||||
with conn.cursor() as cur:
|
||||
for i in range(0, len(imeis), 50):
|
||||
batch = imeis[i:i + 50]
|
||||
resp = api_post("jimi.device.location.get", {
|
||||
"imeis": ",".join(batch),
|
||||
"map_type": "GOOGLE",
|
||||
}, token)
|
||||
|
||||
positions = resp.get("result") or []
|
||||
for p in positions:
|
||||
imei = p.get("imei")
|
||||
lat = clean_num(p.get("lat"))
|
||||
lng = clean_num(p.get("lng"))
|
||||
if not imei or not is_valid_fix(lat, lng):
|
||||
continue
|
||||
|
||||
cur.execute("""
|
||||
INSERT INTO tracksolid.live_positions (
|
||||
imei, geom, lat, lng, speed, direction,
|
||||
gps_time, acc_status, current_mileage, recorded_at
|
||||
) VALUES (
|
||||
%s, ST_SetSRID(ST_MakePoint(%s, %s), 4326),
|
||||
%s, %s, %s, %s, %s, %s, %s, NOW()
|
||||
)
|
||||
ON CONFLICT (imei) DO UPDATE SET
|
||||
geom = EXCLUDED.geom,
|
||||
lat = EXCLUDED.lat,
|
||||
lng = EXCLUDED.lng,
|
||||
speed = EXCLUDED.speed,
|
||||
direction = EXCLUDED.direction,
|
||||
gps_time = EXCLUDED.gps_time,
|
||||
acc_status = EXCLUDED.acc_status,
|
||||
current_mileage = EXCLUDED.current_mileage,
|
||||
updated_at = NOW()
|
||||
""", (
|
||||
imei, lng, lat, lat, lng,
|
||||
clean_num(p.get("speed")),
|
||||
clean_num(p.get("direction")),
|
||||
clean_ts(p.get("gpsTime")),
|
||||
clean(p.get("accStatus")),
|
||||
clean_num(p.get("currentMileage")),
|
||||
))
|
||||
upserted += 1
|
||||
|
||||
conn.commit()
|
||||
log.info("get_device_locations: %d positions refreshed.", upserted)
|
||||
return upserted
|
||||
|
||||
|
||||
# ── Main Loop ─────────────────────────────────────────────────────────────────
|
||||
|
||||
def main():
|
||||
log.info("Starting MOVEMENT PIPELINE (v2.1)...")
|
||||
log.info("Starting MOVEMENT PIPELINE (v2.2)...")
|
||||
|
||||
# Startup catch-up
|
||||
safe_task(sync_devices, log)()
|
||||
safe_task(poll_live_positions, log)()
|
||||
safe_task(poll_trips, log)()
|
||||
safe_task(poll_parking, log)()
|
||||
safe_task(poll_track_list, log)()
|
||||
|
||||
# Schedule
|
||||
schedule.every(60).seconds.do(safe_task(poll_live_positions, log))
|
||||
schedule.every(15).minutes.do(safe_task(poll_trips, log))
|
||||
schedule.every(15).minutes.do(safe_task(poll_parking, log))
|
||||
schedule.every(30).minutes.do(safe_task(poll_track_list, log)) # [FIX-M14]
|
||||
schedule.every().day.at("02:00").do(safe_task(sync_devices, log))
|
||||
|
||||
while True:
|
||||
|
|
|
|||
|
|
@ -1655,10 +1655,11 @@ Common parameters: `deviceImei`, `proNo` (protocol number), `cmdContent` (JSON),
|
|||
| `jimi.track.device.detail` | `ingest_movement_rev.py` | In use |
|
||||
| `jimi.user.device.location.list` | `ingest_movement_rev.py` | In use |
|
||||
| `jimi.device.track.mileage` | `ingest_movement_rev.py` | In use |
|
||||
| `jimi.device.alarm.list` | `ingest_events_rev.py` | In use (verify field names) |
|
||||
| `jimi.device.alarm.list` | `ingest_events_rev.py` | In use — field mapping corrected (FIX-E06) |
|
||||
| `jimi.device.obd.list` | `ingest_events_rev.py` | **Not a real endpoint** — OBD is push-only |
|
||||
| `jimi.device.track.list` | — | Not used (high-res GPS trails) |
|
||||
| `jimi.open.platform.report.parking` | — | Not used (parking_events table exists) |
|
||||
| `jimi.device.track.list` | `ingest_movement_rev.py` | **In use** — poll_track_list() every 30m (FIX-M14) |
|
||||
| `jimi.device.location.get` | `ingest_movement_rev.py` | **In use** — get_device_locations() on-demand (FIX-M15) |
|
||||
| `jimi.open.platform.report.parking` | `ingest_movement_rev.py` | **In use** — acc_type/durSecond fixed (FIX-M13) |
|
||||
| `jimi.device.jimi.media.URL` | — | Not used (media catalog) |
|
||||
| `jimi.device.media.event.URL` | — | Not used (alarm-triggered media) |
|
||||
| All Data Push endpoints | — | Not used (webhook receiver needed) |
|
||||
|
|
|
|||
|
|
@ -127,11 +127,18 @@ ORDER BY current_mileage_km DESC NULLS LAST;
|
|||
|
||||
## 3. tracksolid.position_history
|
||||
|
||||
This is the core time-series table and is implemented as a **TimescaleDB hypertable**, automatically partitioned by time into chunks for efficient storage and querying of large volumes of GPS data. Every time the ingestion service polls the Tracksolid API (approximately every 1 minute), it writes a GPS breadcrumb for each active device into this table. Each row captures the device's location (latitude, longitude, and PostGIS geometry), speed, heading, ignition status, satellite count, and running odometer.
|
||||
This is the core time-series table and is implemented as a **TimescaleDB hypertable**, automatically partitioned by time into chunks for efficient storage and querying of large volumes of GPS data. Two separate ingestion paths write to this table:
|
||||
|
||||
The `acc_status` field indicates whether the vehicle's ignition/accessory circuit is on (`1`) or off (`0`) at the time of the ping. The table uses a composite primary key of `(imei, gps_time)`, ensuring no duplicate pings are stored. Older chunks are transparently compressed by TimescaleDB to save disk space.
|
||||
| Source | `source` value | Frequency | What it captures |
|
||||
|---|---|---|---|
|
||||
| `poll_live_positions` | `poll` | Every 60 seconds | Latest position snapshot across entire fleet |
|
||||
| `poll_track_list` | `track_list` | Every 30 minutes (35-min window) | Every waypoint the device logged — typically 1 fix per 10–30 seconds while moving |
|
||||
|
||||
**Important:** `altitude` is present in the schema but not currently populated by the ingestion pipeline.
|
||||
The `track_list` source (added in FIX-M14) fills the critical gap where a vehicle can travel several kilometres between 60-second fleet sweeps with no trace. Combined, both sources yield approximately **2–6 fixes per minute per active vehicle**.
|
||||
|
||||
Each row captures the device's location (latitude, longitude, and PostGIS geometry), speed, heading, ignition status, satellite count, and running odometer. The `acc_status` field indicates whether the vehicle's ignition/accessory circuit is on (`1`) or off (`0`). The table uses a composite primary key of `(imei, gps_time)`, ensuring no duplicates regardless of which path writes the row. Older chunks are transparently compressed by TimescaleDB to save disk space.
|
||||
|
||||
**Important:** `altitude` is present in the schema but not currently populated by either ingestion path.
|
||||
|
||||
### Describe table structure
|
||||
|
||||
|
|
@ -216,6 +223,82 @@ ORDER BY gps_time ASC;
|
|||
|
||||
---
|
||||
|
||||
### Data density check — poll vs track_list coverage
|
||||
|
||||
Use this to verify that `poll_track_list` (POLL-01) is ingesting data and to compare row counts between the two sources. After the first 30-minute cycle you should see `track_list` rows outnumber `poll` rows by roughly 4–10×.
|
||||
|
||||
```sql
|
||||
SELECT
|
||||
source,
|
||||
COUNT(*) AS total_rows,
|
||||
COUNT(DISTINCT imei) AS devices_seen,
|
||||
MIN(gps_time AT TIME ZONE 'Africa/Nairobi') AS earliest,
|
||||
MAX(gps_time AT TIME ZONE 'Africa/Nairobi') AS latest
|
||||
FROM tracksolid.position_history
|
||||
WHERE gps_time > NOW() - INTERVAL '2 hours'
|
||||
GROUP BY source
|
||||
ORDER BY source;
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### High-resolution speed profile — harsh driving detection
|
||||
|
||||
Flags 30-second windows where a vehicle's speed changed by more than 30 km/h (sudden acceleration or hard braking). Requires `track_list` data to have meaningful resolution — with only 60-second fleet sweeps this query would miss most events.
|
||||
|
||||
```sql
|
||||
WITH ordered AS (
|
||||
SELECT
|
||||
imei,
|
||||
gps_time,
|
||||
speed,
|
||||
LAG(speed) OVER (PARTITION BY imei ORDER BY gps_time) AS prev_speed,
|
||||
LAG(gps_time) OVER (PARTITION BY imei ORDER BY gps_time) AS prev_time
|
||||
FROM tracksolid.position_history
|
||||
WHERE source = 'track_list'
|
||||
AND gps_time > NOW() - INTERVAL '24 hours'
|
||||
AND gps_time < NOW()
|
||||
)
|
||||
SELECT
|
||||
imei,
|
||||
gps_time AT TIME ZONE 'Africa/Nairobi' AS event_time,
|
||||
prev_speed AS speed_before,
|
||||
speed AS speed_after,
|
||||
ABS(speed - prev_speed) AS delta_kmh,
|
||||
EXTRACT(EPOCH FROM (gps_time - prev_time))::INT AS interval_s,
|
||||
CASE
|
||||
WHEN speed > prev_speed THEN 'hard_acceleration'
|
||||
ELSE 'hard_braking'
|
||||
END AS event_type
|
||||
FROM ordered
|
||||
WHERE ABS(speed - prev_speed) > 30
|
||||
AND EXTRACT(EPOCH FROM (gps_time - prev_time)) BETWEEN 5 AND 60
|
||||
ORDER BY delta_kmh DESC;
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Continuous route trace for Grafana map panel
|
||||
|
||||
Returns ordered waypoints for all vehicles in the last hour suitable for rendering as a continuous path in Grafana's Geomap plugin. The `source` filter includes both ingestion paths to maximise trace density.
|
||||
|
||||
```sql
|
||||
SELECT
|
||||
imei,
|
||||
gps_time AT TIME ZONE 'Africa/Nairobi' AS gps_nairobi,
|
||||
lat,
|
||||
lng,
|
||||
speed,
|
||||
acc_status,
|
||||
source
|
||||
FROM tracksolid.position_history
|
||||
WHERE gps_time > NOW() - INTERVAL '1 hour'
|
||||
AND gps_time < NOW()
|
||||
ORDER BY imei, gps_time ASC;
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 4. tracksolid.trips
|
||||
|
||||
The trips table stores auto-detected journey segments. A trip begins when the vehicle's ignition turns on and ends when it turns off again (or after a prolonged stationary period). Each row summarises one journey: start and end times, start and end coordinates, total distance, average and maximum speed, driving time, idle time, and estimated fuel consumption.
|
||||
|
|
@ -1452,3 +1535,5 @@ The following issues were identified during the April 2026 audit. Each represent
|
|||
| 7 | `obd_readings` | 0 rows | No engine health data | **Open** — requires OBD cable installation + `/pushobd` webhook registration in Tracksolid account |
|
||||
| 8 | `parking_events` | 0 rows despite 358 successful API calls | No parking dwell-time reporting | **Fixed** in `ingest_movement_rev.py` [FIX-M13] — added missing `account` and `acc_type=0` params; fixed `durSecond` field mapping |
|
||||
| 9 | `dwh_gold.*` | Both tables empty | Grafana dashboards have no data | **Fixed** — migration 05 adds `refresh_daily_metrics()` ETL function; run nightly via cron or n8n |
|
||||
| 10 | `position_history` | Only 1 fix/min per vehicle from fleet sweep — route traces incomplete | Grafana map paths had gaps; speed profiles too coarse for harsh-driving detection | **Fixed** — `poll_track_list()` added [FIX-M14]; captures every device waypoint every 30 min; density increases to 2–6 fixes/min per vehicle |
|
||||
| 11 | `live_positions` | No on-demand refresh mechanism for specific vehicles | Alarm enrichment and stale-device recovery required waiting up to 60s | **Fixed** — `get_device_locations()` utility added [FIX-M15]; call with specific IMEIs for instant precision refresh |
|
||||
|
|
|
|||
Loading…
Reference in a new issue