""" ingest_movement_rev.py — Fireside Communications · Tracksolid Movement Pipeline ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ RESPONSIBILITY: High-frequency telemetry (GPS), trip summaries, and parking. REVISIONS (QA-Verified): [FIX-M01] sync_devices: Atomic transaction for registry sync. [FIX-M03] is_valid_fix: Filters 0.0 coordinates (Zero Island bug). [FIX-M05] Batching: Groups 50 IMEIs per API call (API Limit Compliance). [FIX-M07] Signal Handling: Clean DB pool closure on SIGTERM/SIGINT. [FIX-M08] Atomic Logging: log_ingestion happens within the data transaction. [FIX-QA-01] Distance: Explicit km to meters conversion (* 1000). ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ """ import signal import sys import time import schedule from datetime import datetime, timezone, timedelta from ts_shared_rev import ( TARGET_ACCOUNT, api_post, close_pool, get_active_imeis, get_conn, get_token, is_valid_fix, log_ingestion, clean, clean_num, clean_int, clean_ts, get_logger, ) log = get_logger("movement") # ── Graceful Shutdown ───────────────────────────────────────────────────────── def _shutdown(signum, frame): log.info("Signal %s received. Closing DB pool...", signum) close_pool() sys.exit(0) signal.signal(signal.SIGTERM, _shutdown) signal.signal(signal.SIGINT, _shutdown) def _safe(fn): """Decorator to prevent scheduler death on single function failure.""" def wrapper(): try: fn() except Exception: log.exception("Task %s failed. Scheduler continuing...", fn.__name__) wrapper.__name__ = fn.__name__ return wrapper # ── 1. Device Registry Sync (Daily) ────────────────────────────────────────── def sync_devices(): log.info("Syncing device registry...") t0, token = time.time(), get_token() if not token: return resp = api_post("jimi.user.device.list", {"target": TARGET_ACCOUNT}, token) if resp.get("code") != 0: return devices = resp.get("result", []) upserted = 0 with get_conn() as conn: with conn.cursor() as cur: for d in devices: imei = d.get("imei") if not imei: continue detail_resp = api_post("jimi.track.device.detail", {"imei": imei}, token) dtl = detail_resp.get("result", {}) if detail_resp.get("code") == 0 else {} cur.execute(""" INSERT INTO tracksolid.devices ( imei, device_name, mc_type, mc_type_use_scope, vehicle_name, vehicle_number, vehicle_models, vehicle_icon, vin, engine_number, vehicle_brand, fuel_100km, driver_name, driver_phone, sim, iccid, imsi, account, customer_name, device_group_id, device_group, activation_time, expiration, enabled_flag, status, current_mileage_km, last_synced_at ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW() ) ON CONFLICT (imei) DO UPDATE SET device_name = EXCLUDED.device_name, vehicle_number = EXCLUDED.vehicle_number, driver_name = EXCLUDED.driver_name, enabled_flag = EXCLUDED.enabled_flag, current_mileage_km = EXCLUDED.current_mileage_km, last_synced_at = NOW(), updated_at = NOW() """, ( imei, clean(d.get("deviceName")), clean(d.get("mcType")), clean(d.get("mcTypeUseScope")), clean(d.get("vehicleName")), clean(d.get("vehicleNumber")), clean(d.get("vehicleModels")), clean(d.get("vehicleIcon")), clean(dtl.get("vin")), clean(dtl.get("engineNumber")), clean(dtl.get("vehicleBrand")), clean_num(dtl.get("fuel_100km")), clean(d.get("driverName")), clean(d.get("driverPhone")), clean(d.get("sim")), clean(dtl.get("iccid")), clean(dtl.get("imsi")), clean(dtl.get("account")), clean(dtl.get("customerName")), clean(d.get("deviceGroupId")), clean(d.get("deviceGroup")), clean_ts(d.get("activationTime")), clean_ts(d.get("expiration")), clean_int(d.get("enabledFlag", 1)), clean(dtl.get("status", "active")), clean_num(dtl.get("currentMileage")) )) upserted += 1 log_ingestion(cur, "jimi.user.device.list+detail", len(devices), upserted, 0, int((time.time()-t0)*1000), True) conn.commit() log.info("Registry sync: %d devices updated.", upserted) # ── 2. Live Positions (Every 60s) ───────────────────────────────────────────── def poll_live_positions(): t0, token = time.time(), get_token() if not token: return resp = api_post("jimi.user.device.location.list", {"target": TARGET_ACCOUNT, "map_type": "GOOGLE"}, token) if resp.get("code") != 0: return positions = resp.get("result", []) upserted, inserted = 0, 0 with get_conn() as conn: with conn.cursor() as cur: for p in positions: imei, lat, lng = p.get("imei"), clean_num(p.get("lat")), 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, pos_type, confidence, gps_time, hb_time, speed, direction, acc_status, gps_signal, gps_num, elec_quantity, power_value, battery_power_val, tracker_oil, temperature, current_mileage, device_status, loc_desc, recorded_at ) VALUES ( %s, ST_SetSRID(ST_MakePoint(%s, %s), 4326), %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW() ) ON CONFLICT (imei) DO UPDATE SET geom=EXCLUDED.geom, lat=EXCLUDED.lat, lng=EXCLUDED.lng, gps_time=EXCLUDED.gps_time, speed=EXCLUDED.speed, direction=EXCLUDED.direction, acc_status=EXCLUDED.acc_status, current_mileage=EXCLUDED.current_mileage, updated_at=NOW() """, ( imei, lng, lat, lat, lng, clean(p.get("posType")), clean_int(p.get("confidence")), clean_ts(p.get("gpsTime")), clean_ts(p.get("hbTime")), clean_num(p.get("speed")), clean_num(p.get("direction")), clean(p.get("accStatus")), clean_int(p.get("gpsSignal")), clean_int(p.get("gpsNum")), clean_num(p.get("electQuantity")), clean_num(p.get("powerValue")), clean_num(p.get("batteryPowerVal")), clean(p.get("trackerOil")), clean_num(p.get("temperature")), clean_num(p.get("currentMileage")), clean(p.get("status")), clean(p.get("locDesc")) )) upserted += 1 # History (Hypertable Source) if clean_ts(p.get("gpsTime")): cur.execute(""" INSERT INTO tracksolid.position_history (imei, gps_time, geom, lat, lng, speed, direction, acc_status, satellite, current_mileage) VALUES (%s, %s, ST_SetSRID(ST_MakePoint(%s, %s), 4326), %s, %s, %s, %s, %s, %s, %s) ON CONFLICT (imei, gps_time) DO NOTHING """, (imei, clean_ts(p.get("gpsTime")), lng, lat, lat, lng, clean_num(p.get("speed")), clean_num(p.get("direction")), clean(p.get("accStatus")), clean_int(p.get("gpsNum")), clean_num(p.get("currentMileage")))) inserted += 1 log_ingestion(cur, "jimi.user.device.location.list", len(positions), upserted, inserted, int((time.time()-t0)*1000), True) conn.commit() # ── 3. Trip Reports (Every 15m) ─────────────────────────────────────────────── def poll_trips(): 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(hours=1) inserted = 0 for i in range(0, len(imeis), 50): batch = imeis[i:i+50] resp = api_post("jimi.device.track.mileage", { "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") }, token) trips = resp.get("result", []) with get_conn() as conn: with conn.cursor() as cur: for t in trips: dist_km = clean_num(t.get("distance")) dist_m = dist_km * 1000 if dist_km is not None else 0 # [QA-01] Conversion cur.execute(""" INSERT INTO tracksolid.trips (imei, start_time, end_time, distance_m, avg_speed_kmh) VALUES (%s, %s, %s, %s, %s) ON CONFLICT (imei, start_time) DO NOTHING """, (t.get("imei"), clean_ts(t.get("startTime")), clean_ts(t.get("endTime")), dist_m, clean_num(t.get("avgSpeed")))) inserted += 1 conn.commit() log.info("Trips: %d records processed.", inserted) # ── Main Loop ───────────────────────────────────────────────────────────────── def main(): log.info("Starting MOVEMENT PIPELINE (v2.0)...") # Startup catch-up _safe(sync_devices)() _safe(poll_live_positions)() _safe(poll_trips)() # Schedule schedule.every(60).seconds.do(_safe(poll_live_positions)) schedule.every(15).minutes.do(_safe(poll_trips)) schedule.every().day.at("02:00").do(_safe(sync_devices)) while True: schedule.run_pending() time.sleep(1) if __name__ == "__main__": main()