FleetFuel

·Context

FleetOps (15_fleetops) is the fleet operations analytics SPA. It already has a trip-derived fuel panel (GET /analytics/fuel), but that data is effectively empty — estimated fuel needs devices.fuel_100km (NULL fleet-wide) and actual oils is sparse.

A real fuel-spend feed now lands in the RustFS fuel bucket: WhatsApp fuel-update messages, extracted by an n8n CDC job from logistics_department.fuel_records — 1,922 rows (Feb–Jun 2026) of actual fills (litres, KES amount, odometer, fuel type, driver, department), keyed by number plate.

Goal (full vertical): a new 17_fleetfuel module pulls the bucket into the shared tracksolid_db under its own fuel schema, exposes it via dashboard_api, and adds a new “Fuel Log” tab to FleetOps. The existing trip-derived panel stays as-is — the two coexist. This mirrors the proven 16_fleettickets module pattern exactly.

·The data (confirmed by reading the bucket)

Bucket layout (s3.rahamafresh.com, path-style, region us-east-1):

• fuel_records/latest.json — full snapshot, envelope { metadata, records[] }, ~1.7 MB / 1922 rows.
• fuel_records/changes/<ISO-ts>.json — hourly CDC deltas (same envelope, includes soft-deletes).
• .csv siblings exist, but we ingest the JSON (richer, typed).

Record shape (stable PK id):

id, record_datetime, department, driver, car, liters, amount, fuel_type, odometer,
sender_name, sender_phone, raw_message, source, source_instance, source_message_id,
source_event_timestamp, message_fingerprint, deleted_at, deleted_by, delete_reason,
delete_source, created_at

The data is messy (WhatsApp-sourced) → normalization is essential:

• car: KCA 542Q vs KCA542Q, plus junk (ANY VEH). 162 distinct.
• fuel_type: PETROL/DIESEL + typos (DISIEL, DISEL, PETRO, /PETROL, VPOWER, null).
• department: ~30 case/spelling variants of ~12 real departments (OSP/osp/Osp, ROLL-OUT/ROLLOUT).
• deleted_at set on 34 rows (soft-deleted — must be excluded from reporting).

·Pattern mirrored: 16_fleettickets

Self-contained Python module → reads a RustFS bucket → upserts raw-jsonb rows into a namespaced schema in the shared tracksolid_db → idempotent migrations with a schema_migrations ledger → a reporting.* view consumed by dashboard_api → surfaced as a FleetOps tab. Reuse: shared.py (DB ctx-mgr + clean), run_migrations.py (ledger runner), the dry-run/--apply CLI convention, and the .env/pyproject/README layout.

AIngestion repo — 17_fleetfuel new

Created in /Users/kianiadee/Downloads/projects/17_fleetfuel, files mirroring fleettickets:

• pyproject.toml — psycopg2-binary, boto3 (the aws CLI isn’t available, and the CDC changes/ listing needs list_objects_v2 pagination, so boto3 beats CLI-subprocess). ruff dev dep.
• shared.py — copy verbatim (get_conn, get_logger, clean); rename logger ns to fleetfuel.
• run_migrations.py — copy; swap ledger to fuel.schema_migrations.
• .env.example — DATABASE_URL, RUSTFS_ENDPOINT/ACCESS_KEY/SECRET_KEY/REGION, FUEL_BUCKET=fuel.
• .gitignore — .env, __pycache__, .venv.
• README.md — what it owns vs. not (DB schema = ours; read-API = dashboard_api; frontend = fleetops).
• migrations/01_fuel_schema.sql — see Part B.
• import_fuel.py — the loader (below).
• s3util.py (optional) — thin boto3 client factory (endpoint_url + path-style addressing).

import_fuel.py

• boto3 S3 client from RUSTFS_* env.
• Default --snapshot: GET fuel_records/latest.json, upsert all records on id. At 1922 rows / hourly cadence this full reconcile is trivial and self-healing (picks up edits + soft-deletes) → simplest correct design.
• --changes (optional, lower-latency): list fuel_records/changes/, process files newer than a watermark in fuel.ingest_state.
• --file <path>: local JSON for dev/testing.
• --apply writes; default is a dry-run logging parsed/valid/skipped counts.
• Upsert via execute_values: INSERT … ON CONFLICT (id) DO UPDATE SET raw=EXCLUDED.raw, updated_at=now(). Derived/normalized columns populated by a DB trigger reading raw. Scrub JSON NaN → null first.

Bmigrations/01_fuel_schema.sql — the fuel schema new

Idempotent, lives in shared tracksolid_db:

• CREATE SCHEMA IF NOT EXISTS fuel; + reporting;
• Normalizer functions (IMMUTABLE, single source of truth):
  • fuel.norm_plate(text) → upper, strip non-alphanumeric (KCA 542Q→KCA542Q); null out junk.
  • fuel.canon_fuel_type(text) → map typos to PETROL / DIESEL / VPOWER / OTHER / NULL.
  • fuel.canon_department(text) → upper + collapse-ws + variant map to canonical set.
• fuel.records — raw-first + derived columns populated by a BEFORE INSERT/UPDATE trigger from raw: id, raw, record_datetime, plate, car_raw, liters, amount, fuel_type, department, driver, odometer, deleted_at, message_fingerprint, ingested_at, updated_at. Indexes on plate, record_datetime, department, partial WHERE deleted_at IS NULL.
• fuel.ingest_state — watermark for --changes mode.
• reporting.v_fuel_fills — read view: fuel.records (deleted_at IS NULL) LEFT JOIN tracksolid.devices d ON fuel.norm_plate(d.vehicle_number) = r.plate, exposing fuel_date, plate, vehicle_number, cost_centre, assigned_city, imei, department, driver, liters, amount, fuel_type, odometer. Same filter contract as reporting.v_daily_summary.
• reporting.v_fuel_efficiency (optional, high-value) — per-plate window over record_datetime: km = odometer − lag(odometer), km_per_litre = km / liters, with defensive bounds.
• Grants — USAGE + SELECT on the views to dashboard_ro (mirror tracksolid migration 18).

Cdashboard_api read endpoints edit

In dashboard_api_rev.py, add endpoints reusing _analytics_window + _dim_filters + RealDictCursor:

• GET /analytics/fuel-fills — period/start/end + dims + optional department, fuel_type. Returns: totals (litres, spend_kes, fills, avg_price_per_litre, vehicles_fuelled), rows (per-vehicle), by_department, trend (daily litres + spend), data_status (unmatched-plate count).
• GET /analytics/fuel-fills/recent — recent N fills for the detail table.
• Extend GET /analytics/filters to also return departments and fuel_types.

No business logic in the API — it only selects from the reporting.* views.

DFleetOps SPA — new “Fuel Log” tab edit

In 15_fleetops/src/index.html (single-file SPA, inline JS + Chart.js). Leave the existing fuel panel untouched; add a “Fuel Log” tab:

• KPI strip: total litres, total KES spend, fills, avg KES/litre, vehicles fuelled.
• Trend chart (spend + litres) — reuse the utilisation panel’s Chart.js setup.
• Per-vehicle table (litres, spend, fills, last odometer, km/l) + by-department breakdown.
• Recent-fills detail table from /analytics/fuel-fills/recent.
• Wire to the shared filter state + new department / fuel-type dropdowns; calls go through the existing API_BASE mechanism.

EGit & deploy

• git init in 17_fleetfuel, add repo.rahamafresh.com/kianiadee/fleetfuel.git as origin, push (repo is currently empty).
• Deploy like fleettickets: a Coolify container/cron in the stack runs run_migrations.py then import_fuel.py --snapshot --apply hourly (matching the CDC cadence).
• dashboard_api + fleetops ride their existing Coolify pipelines (feature → staging → main).

·Critical files

Reuse: 16_fleettickets/shared.py, run_migrations.py, the _scrub_nan/upsert shape; dashboard_api_rev.py:444 _dim_filters, _analytics_window; the stdlib SigV4 reader already proven this session (fallback if boto3 is undesirable).

✓Verification (end-to-end)

1. Bucket read — already proven this session (listed 14 objects, parsed latest.json = 1922 rows).
2. Ingestion dry-run — python import_fuel.py --snapshot (no --apply): logs parsed/valid/skipped, no DB writes.
3. Migrate + apply — python run_migrations.py then import_fuel.py --snapshot --apply. Spot-check: SELECT count(*), count(*) FILTER (WHERE deleted_at IS NULL) FROM fuel.records; (≈1922 / ≈1888) and plate-match rate in reporting.v_fuel_fills.
4. API — curl "$API/analytics/fuel-fills?period=90d" → totals/rows/by_department/trend non-empty; /analytics/filters includes departments.
5. Frontend — build & run fleetops locally, open the Fuel Log tab, confirm KPIs/chart/tables render and filters drive the queries.
6. /verify the fleetops change once wired.