c05b47abe2
BUG-01 [FIX-E06]: jimi.device.alarm.list poll response uses alertTypeId/ alarmTypeName/alertTime, not the webhook field names. All 1,054 stored alarm records had null alarm_type/alarm_name as a result. Corrected field mapping in ingest_events_rev.py; also added alarm_name and source columns to INSERT. BUG-02 [FIX-M11/M12]: trips.distance_m was storing millimetres due to an erroneous * 1000 on an already-km API value. Removed the multiplication in poll_trips() and push_trip_report(). Column renamed to distance_km in migration 04 (historical rows divided by 1,000,000 to correct to km). All SQL in both ingestion files updated to reference distance_km. POLL-02 [FIX-M13]: parking poll returned 0 rows because the required account and acc_type=0 parameters were missing. Also fixed response field mapping: durSecond was incorrectly read as 'seconds'. Migration 04: corrects and renames distance_m → distance_km. Migration 05: adds normalized OBD columns, alarm/device enrichment columns, new tables (device_events, fuel_readings, temperature_readings, lbs_readings, geofences), expands dwh_gold fact table, and adds refresh_daily_metrics() ETL. tracksolid_DB_manual.md updated to reflect column rename and mark fixed issues. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
117 lines
No EOL
4.9 KiB
Python
117 lines
No EOL
4.9 KiB
Python
"""
|
|
ingest_events_rev.py — Fireside Communications · Tracksolid Events Pipeline
|
|
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
|
|
RESPONSIBILITY: Alarm event polling (catch-up/fallback for webhook push data).
|
|
|
|
OBD diagnostics are received via the webhook_receiver_rev.py push service —
|
|
jimi.device.obd.list does not exist in the Tracksolid Pro API.
|
|
|
|
REVISIONS (QA-Verified):
|
|
[FIX-E01] Batching: Polls 50 IMEIs per call to stay within API limits.
|
|
[FIX-E03] Atomic Logging: One log row per batch per endpoint.
|
|
[FIX-E04] Signal Handling: Clean pool closure on SIGTERM/SIGINT.
|
|
[FIX-E05] Removed poll_obd: OBD data is push-only via /pushobd webhook.
|
|
[FIX-11] Uses shared safe_task/setup_shutdown from ts_shared_rev (DRY).
|
|
[FIX-E06] BUG-01: jimi.device.alarm.list returns alertTypeId/alarmTypeName/
|
|
alertTime — not alarmType/alarmName/alarmTime (those are webhook
|
|
field names). Corrected field mapping so alarm_type and alarm_name
|
|
are no longer silently stored as NULL.
|
|
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
|
|
"""
|
|
|
|
import time
|
|
import schedule
|
|
from datetime import datetime, timezone, timedelta
|
|
|
|
from ts_shared_rev import (
|
|
api_post,
|
|
get_active_imeis,
|
|
get_conn,
|
|
get_token,
|
|
log_ingestion,
|
|
clean,
|
|
clean_num,
|
|
clean_int,
|
|
clean_ts,
|
|
get_logger,
|
|
safe_task,
|
|
setup_shutdown,
|
|
)
|
|
|
|
log = get_logger("events")
|
|
setup_shutdown(log)
|
|
|
|
# ── 1. Alarms & Geofence Events (Every 5m) ────────────────────────────────────
|
|
|
|
def poll_alarms():
|
|
log.info("Polling device alarms...")
|
|
t0, token, imeis = time.time(), get_token(), get_active_imeis()
|
|
if not token or not imeis: return
|
|
|
|
end_ts = datetime.now(timezone.utc)
|
|
start_ts = end_ts - timedelta(minutes=30) # Look back 30m to ensure coverage
|
|
inserted = 0
|
|
|
|
for i in range(0, len(imeis), 50):
|
|
batch = imeis[i:i+50]
|
|
resp = api_post("jimi.device.alarm.list", {
|
|
"imeis": ",".join(batch),
|
|
"begin_time": start_ts.strftime("%Y-%m-%d %H:%M:%S"),
|
|
"end_time": end_ts.strftime("%Y-%m-%d %H:%M:%S"),
|
|
"page_size": 100
|
|
}, token)
|
|
|
|
alarms = resp.get("result") or []
|
|
if not alarms: continue
|
|
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
for a in alarms:
|
|
lat, lng = clean_num(a.get("lat")), clean_num(a.get("lng"))
|
|
# [FIX-E06] Poll response uses alertTypeId/alarmTypeName/alertTime,
|
|
# not alarmType/alarmName/alarmTime (those are webhook push field names).
|
|
alarm_type = clean(a.get("alertTypeId"))
|
|
alarm_name = clean(a.get("alarmTypeName"))
|
|
alarm_time = clean_ts(a.get("alertTime"))
|
|
|
|
cur.execute("""
|
|
INSERT INTO tracksolid.alarms (
|
|
imei, alarm_type, alarm_name, alarm_time, geom, lat, lng,
|
|
speed, acc_status, source, updated_at
|
|
) VALUES (
|
|
%s, %s, %s, %s,
|
|
CASE WHEN %s IS NOT NULL AND %s IS NOT NULL
|
|
THEN ST_SetSRID(ST_MakePoint(%s, %s), 4326)
|
|
ELSE NULL END,
|
|
%s, %s, %s, %s, 'poll', NOW()
|
|
) ON CONFLICT (imei, alarm_type, alarm_time) DO NOTHING
|
|
""", (
|
|
a.get("imei"), alarm_type, alarm_name, alarm_time,
|
|
lng, lat, lng, lat, lat, lng,
|
|
clean_num(a.get("speed")), clean(a.get("accStatus"))
|
|
))
|
|
inserted += 1
|
|
|
|
log_ingestion(cur, "jimi.device.alarm.list", len(batch), 0, inserted, int((time.time()-t0)*1000), True)
|
|
conn.commit()
|
|
|
|
log.info("Alarms: %d new events inserted.", inserted)
|
|
|
|
# ── Main Loop ─────────────────────────────────────────────────────────────────
|
|
|
|
def main():
|
|
log.info("Starting EVENTS PIPELINE (v2.1)...")
|
|
# OBD removed: Data arrives via webhook push (/pushobd), not polling.
|
|
|
|
# Startup catch-up
|
|
safe_task(poll_alarms, log)()
|
|
|
|
# Schedule
|
|
schedule.every(5).minutes.do(safe_task(poll_alarms, log))
|
|
|
|
while True:
|
|
schedule.run_pending()
|
|
time.sleep(1)
|
|
|
|
if __name__ == "__main__":
|
|
main() |