Petrolyst/petrolyst-infra
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fuel-db: customer on every row, place and holder for site, licence tables

Browse Source 
Draft migrations rewritten in place (PD-16, PD-18), with 0004 for the licence
tables (PD-20) and rule tests in fuel-db/tests. Dashboards' column list
regenerated; coverage_detail selects operator_uid.

Checks: tests/run.sh 106 of 106; 18 views, 686 rows, 0 differences against a
reference from the same corpus. Codex's fixes to its inspection are not
re-inspected.

Bands: persistence, Superset. Rules: 6, 8.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Nasim SadatandClaude Opus 5.5 committed 2026-10-30 14:40:16 +01:00
1 parent e4806f5799
commit ab553526f6
11 files changed
+1197 -398
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

stacks/fuel-monitoring/fuel-db/migrations/0002_fm_derivations.sql
+58 -44
View File
@@ -4,6 +4,18 @@
BEGIN;
SET LOCAL search_path = fm, pg_catalog, public;
-- Each holder's customer, current place and local day. The current place is found by the customer's
-- local day; a holder without a place, or a place without its own zone, uses the customer's zone.
CREATE VIEW fm.v_holder_current AS
SELECT h.holder_uid, h.operator_uid, hp.place_uid,
coalesce(pl.timezone, o.timezone) AS timezone,
(now() AT TIME ZONE coalesce(pl.timezone, o.timezone))::date AS today
FROM fm.holder h
JOIN fm.operator o ON o.operator_uid = h.operator_uid
LEFT JOIN fm.holder_place hp ON hp.holder_uid = h.holder_uid
AND daterange(coalesce(hp.valid_from, '-infinity'::date), hp.valid_to, '[)') @> (now() AT TIME ZONE o.timezone)::date
LEFT JOIN fm.place pl ON pl.place_uid = hp.place_uid;
CREATE VIEW fm.v_source_rank AS
SELECT sha256, kind, CASE kind WHEN 'delivery_form' THEN 1 WHEN 'site_logbook' THEN 2
WHEN 'operator_record' THEN 3 WHEN 'operator_workbook' THEN 3 ELSE 9 END AS rank
@@ -74,7 +86,7 @@ HAVING count(*) FILTER (WHERE s.claim_status IN ('asserted','reviewed') AND s.li
-- v1: tank-level rows are summed only in a later version, and only when every placed tank has one.
CREATE VIEW fm.v_stock_point AS -- stock on arrival and on departure of each live visit
SELECT v.visit_id, v.visit_key, v.site_uid, v.visit_on, v.seq_in_day, v.kind,
SELECT v.visit_id, v.visit_key, v.holder_uid, v.visit_on, v.seq_in_day, v.kind,
d.delivered_l, coalesce(o.n_external, 0) AS n_other_movements,
coalesce(b.litres, st.litres, a.litres - d.delivered_l) AS arrive_l,
CASE WHEN coalesce(b.litres, st.litres) IS NOT NULL THEN 'stated'
@@ -104,10 +116,10 @@ LEFT JOIN fm.v_visit_stock st ON st.visit_id = v.visit_id AND st.phase = 'standa
CREATE VIEW fm.v_stock_interval AS -- consecutive stock points per site; no overlap by construction
WITH pts AS (
SELECT p.*, row_number() OVER (PARTITION BY p.site_uid ORDER BY p.visit_on, p.seq_in_day) AS n
SELECT p.*, row_number() OVER (PARTITION BY p.holder_uid ORDER BY p.visit_on, p.seq_in_day) AS n
FROM fm.v_stock_point p WHERE p.arrive_l IS NOT NULL OR p.depart_l IS NOT NULL),
pr AS (
SELECT o.site_uid, o.visit_key || '>' || c.visit_key AS interval_key,
SELECT o.holder_uid, o.visit_key || '>' || c.visit_key AS interval_key,
o.visit_id AS open_visit_id, c.visit_id AS close_visit_id,
o.visit_on AS start_on, c.visit_on AS end_on, c.visit_on - o.visit_on AS days,
o.depart_l AS opening_l, o.depart_basis AS opening_basis,
@@ -117,8 +129,8 @@ pr AS (
coalesce(o.depart_reviewed AND c.arrive_reviewed, false) AS endpoints_reviewed,
o.n_other_movements + c.n_other_movements AS n_endpoint_movements,
o.visit_on AS o_on, o.seq_in_day AS o_seq, c.visit_on AS c_on, c.seq_in_day AS c_seq
FROM pts o JOIN pts c ON c.site_uid = o.site_uid AND c.n = o.n + 1)
SELECT p.site_uid, p.interval_key, p.open_visit_id, p.close_visit_id, p.start_on, p.end_on, p.days,
FROM pts o JOIN pts c ON c.holder_uid = o.holder_uid AND c.n = o.n + 1)
SELECT p.holder_uid, p.interval_key, p.open_visit_id, p.close_visit_id, p.start_on, p.end_on, p.days,
p.opening_l, p.opening_basis, p.closing_l, p.closing_basis,
coalesce(mid.delivered_l, 0) AS deliveries_between_l, coalesce(mid.out_l, 0) AS out_l,
p.opening_l + coalesce(mid.delivered_l,0) - coalesce(mid.out_l,0) - p.closing_l AS consumed_l,
@@ -144,7 +156,7 @@ CROSS JOIN LATERAL (
FROM fm.v_visit_live v
LEFT JOIN fm.v_visit_delivery d ON d.visit_id = v.visit_id
LEFT JOIN fm.v_visit_other_movement x ON x.visit_id = v.visit_id
WHERE v.site_uid = p.site_uid
WHERE v.holder_uid = p.holder_uid
AND (v.visit_on, v.seq_in_day) > (p.o_on, p.o_seq)
AND (v.visit_on, v.seq_in_day) < (p.c_on, p.c_seq)) mid;
@@ -180,7 +192,7 @@ WITH gen AS ( -- generators placed at the site at any time in the
bool_or(daterange(coalesce(p.occupied_from,'-infinity'::date), p.occupied_to, '[)')
@> daterange(i.start_on, i.end_on, '[]')) AS full_cover
FROM fm.v_stock_interval i
JOIN fm.asset_placement p ON p.site_uid = i.site_uid
JOIN fm.asset_placement p ON p.holder_uid = i.holder_uid
JOIN fm.asset a ON a.asset_uid = p.asset_uid AND a.kind = 'generator'
WHERE daterange(coalesce(p.occupied_from,'-infinity'::date), p.occupied_to, '[)')
&& daterange(i.start_on, i.end_on, '[]')
@@ -201,13 +213,13 @@ pair AS ( -- every counter read at both endpoints, classified
o.all_reviewed AND c.all_reviewed AS reviewed,
(o.hours_max > o.hours_min OR c.hours_max > c.hours_min) AS conflict
FROM fm.v_stock_interval i
JOIN fm.runtime_meter m ON m.site_uid = i.site_uid
JOIN fm.runtime_meter m ON m.holder_uid = i.holder_uid
JOIN fm.v_visit_reading o ON o.visit_id = i.open_visit_id AND o.meter_id = m.meter_id
JOIN fm.v_visit_reading c ON c.visit_id = i.close_visit_id AND c.meter_id = m.meter_id),
idle AS ( -- at most one live generator_idle per asset and date (EXCLUDE)
SELECT g.interval_key, g.asset_uid, x.code, x.status
FROM gen g JOIN fm.v_stock_interval i USING (interval_key)
JOIN fm.assumption x ON x.asset_uid = g.asset_uid AND x.site_uid = i.site_uid -- this site's only
JOIN fm.assumption x ON x.asset_uid = g.asset_uid AND x.holder_uid = i.holder_uid -- this site's only
AND x.topic = 'generator_idle' AND x.status IN ('proposed','accepted')
AND daterange(x.applies_from, x.applies_to, '[]') @> daterange(i.start_on, i.end_on, '[]')),
per_gen AS (
@@ -234,7 +246,7 @@ loose AS ( -- counters read at both ends but not tied to a set
SELECT interval_key, array_agg(meter_id ORDER BY meter_id) AS meters
FROM pair WHERE tie = 'not_tied' GROUP BY interval_key),
cova AS ( -- live meter_coverage assumptions with their exact members
SELECT x.assumption_id, x.code, x.site_uid, x.status, x.applies_from, x.applies_to,
SELECT x.assumption_id, x.code, x.holder_uid, x.status, x.applies_from, x.applies_to,
coalesce(array_agg(m.meter_id ORDER BY m.meter_id)
FILTER (WHERE m.member_role = 'counter'), '{}') AS meters,
coalesce(array_agg(m.asset_uid ORDER BY m.asset_uid)
@@ -254,7 +266,7 @@ unm AS ( -- at most one live coverage assumption per site an
FROM loose l
JOIN fm.v_stock_interval i USING (interval_key)
LEFT JOIN cov cv USING (interval_key)
LEFT JOIN cova c ON c.site_uid = i.site_uid
LEFT JOIN cova c ON c.holder_uid = i.holder_uid
AND daterange(c.applies_from, c.applies_to, '[]') @> daterange(i.start_on, i.end_on, '[]')),
sel AS ( -- the validated population; nothing else is summed
SELECT p.* FROM pair p
@@ -343,7 +355,7 @@ FROM x;
-- M4 (L per calendar day) per stock interval: no runtime, so no generator population. The
-- population is the site's whole stock. Input to the L/day proposal path only; never a KPI.
CREATE VIEW fm.v_m4_interval AS
SELECT i.*, 'stock:' || i.site_uid AS population_key,
SELECT i.*, 'stock:' || i.holder_uid AS population_key,
CASE WHEN i.days > 0 THEN i.consumed_l / i.days END AS l_per_day,
CASE WHEN i.stock_status = 'ok' THEN 'eligible' ELSE 'excluded' END AS eligibility,
CASE WHEN i.stock_status <> 'ok' THEN 'stock interval: ' || i.stock_status ELSE '' END AS exclusion_reasons,
@@ -359,7 +371,7 @@ WITH span AS ( -- the endpoint visits and every visit between them
FROM fm.v_stock_interval i
JOIN fm.visit o ON o.visit_id = i.open_visit_id
JOIN fm.visit c ON c.visit_id = i.close_visit_id
JOIN fm.visit v ON v.site_uid = i.site_uid
JOIN fm.visit v ON v.holder_uid = i.holder_uid
AND (v.visit_on, v.seq_in_day) >= (o.visit_on, o.seq_in_day)
AND (v.visit_on, v.seq_in_day) <= (c.visit_on, c.seq_in_day)),
stock_deps AS (
@@ -379,27 +391,27 @@ runtime_deps AS (
FROM fm.hour_meter_reading x WHERE x.visit_id IN (i.open_visit_id, i.close_visit_id)
UNION ALL
SELECT 'meter:' || m.meter_id || ':' || m.mapping_status || ':' || coalesce(m.asset_uid, '-')
FROM fm.runtime_meter m WHERE m.site_uid = i.site_uid
FROM fm.runtime_meter m WHERE m.holder_uid = i.holder_uid
UNION ALL
SELECT 'epoch:' || e.meter_id || ':' || e.epoch_id || ':' || coalesce(e.starts_on::text, '-')
|| ':' || e.start_reason || ':' || e.continuity || ':' || coalesce(e.rollover_at::text, '-')
FROM fm.meter_epoch e JOIN fm.runtime_meter m ON m.meter_id = e.meter_id
WHERE m.site_uid = i.site_uid
WHERE m.holder_uid = i.holder_uid
UNION ALL -- placements at this site, and anywhere of the assets its meters name
SELECT 'placement:' || p.asset_uid || ':' || p.site_uid || ':' || coalesce(p.occupied_from::text, '-')
SELECT 'placement:' || p.asset_uid || ':' || coalesce(p.holder_uid, p.place_uid) || ':' || coalesce(p.occupied_from::text, '-')
|| ':' || coalesce(p.occupied_to::text, '-')
FROM fm.asset_placement p
WHERE (p.site_uid = i.site_uid
OR p.asset_uid IN (SELECT m.asset_uid FROM fm.runtime_meter m WHERE m.site_uid = i.site_uid))
WHERE (p.holder_uid = i.holder_uid
OR p.asset_uid IN (SELECT m.asset_uid FROM fm.runtime_meter m WHERE m.holder_uid = i.holder_uid))
AND daterange(coalesce(p.occupied_from, '-infinity'::date), p.occupied_to, '[)')
&& daterange(i.start_on, i.end_on, '[]')
UNION ALL
SELECT 'assumption:' || x.code || ':' || x.fingerprint || ':' || x.status
FROM fm.assumption x
WHERE x.site_uid = i.site_uid AND x.topic IN ('generator_idle','meter_coverage')
WHERE x.holder_uid = i.holder_uid AND x.topic IN ('generator_idle','meter_coverage')
AND daterange(x.applies_from, x.applies_to, '[]') && daterange(i.start_on, i.end_on, '[]')) d
GROUP BY i.interval_key)
SELECT i.interval_key, i.site_uid, i.open_visit_id, i.close_visit_id,
SELECT i.interval_key, i.holder_uid, i.open_visit_id, i.close_visit_id,
coalesce(s.deps, '[]'::jsonb) AS stock_deps, coalesce(r.deps, '[]'::jsonb) AS runtime_deps,
encode(sha256(convert_to(coalesce(s.deps, '[]'::jsonb)::text, 'UTF8')), 'hex') AS stock_dep_sha256,
encode(sha256(convert_to(coalesce(r.deps, '[]'::jsonb)::text, 'UTF8')), 'hex') AS runtime_dep_sha256
@@ -424,11 +436,11 @@ GROUP BY b.baseline_id, b.metric;
-- Consecutive readings per meter (live visits), giving hours per day: R1 / public.v_runtime.
CREATE VIEW fm.v_runtime_pair AS
WITH r AS (
SELECT v.site_uid, v.visit_id, v.visit_on, v.seq_in_day, x.meter_id, x.epoch_id, x.n_epochs,
SELECT v.holder_uid, v.visit_id, v.visit_on, v.seq_in_day, x.meter_id, x.epoch_id, x.n_epochs,
x.n_unassigned_epoch, x.hours, x.all_reviewed,
row_number() OVER (PARTITION BY x.meter_id ORDER BY v.visit_on, v.seq_in_day) AS n
FROM fm.v_visit_reading x JOIN fm.v_visit_live v ON v.visit_id = x.visit_id)
SELECT o.site_uid, o.meter_id, o.visit_id AS open_visit_id, c.visit_id AS close_visit_id,
SELECT o.holder_uid, o.meter_id, o.visit_id AS open_visit_id, c.visit_id AS close_visit_id,
o.visit_on AS start_on, c.visit_on AS end_on, c.visit_on - o.visit_on AS days,
c.hours - o.hours AS runtime_h,
CASE WHEN c.visit_on > o.visit_on THEN (c.hours - o.hours) / (c.visit_on - o.visit_on) END AS h_per_day,
@@ -443,7 +455,7 @@ FROM r o JOIN r c ON c.meter_id = o.meter_id AND c.n = o.n + 1;
-- assumption's dates, divided by that independently sourced runtime assumption (never one derived
-- from fuel). The window is the matched one. Reference tab only; never a KPI (design §4.1).
CREATE VIEW fm.v_modelled_lph AS
SELECT x.site_uid, x.code, x.status, x.basis,
SELECT x.holder_uid, x.code, x.status, x.basis,
x.value_num AS h_per_day, x.low_num AS h_per_day_low, x.high_num AS h_per_day_high,
m4.window_start, m4.window_end, m4.days AS matched_days, m4.fuel_l / NULLIF(m4.days, 0) AS l_per_day,
m4.fuel_l / NULLIF(m4.days, 0) / NULLIF(x.value_num, 0) AS modelled_l_per_h,
@@ -454,46 +466,48 @@ CROSS JOIN LATERAL (
SELECT sum(i.consumed_l) AS fuel_l, sum(i.days) AS days,
min(i.start_on) AS window_start, max(i.end_on) AS window_end
FROM fm.v_m4_interval i
WHERE i.site_uid = x.site_uid AND i.eligibility = 'eligible'
WHERE i.holder_uid = x.holder_uid AND i.eligibility = 'eligible'
AND daterange(x.applies_from, x.applies_to, '[]') @> daterange(i.start_on, i.end_on, '[]')) m4
WHERE x.topic = 'runtime_hours_per_day' AND NOT x.derived_from_fuel AND x.status IN ('proposed','accepted')
AND m4.days > 0;
-- Review entries whose fingerprint matches no live row: the value changed after it was reviewed,
-- or the subject was superseded or retracted. The site comes from every row the subject key has
-- had, in any status; for an asset's specification or engine total, from every site the asset was
-- placed at. So one entry can give one row per site, and none is lost when its subject leaves.
-- or the subject was superseded or retracted. The holder comes from every row the subject key has
-- had, in any status; for an asset's specification or engine total, from every holder the asset was
-- placed on; for a location, from every holder that has stood at the place. So one entry can give
-- one row per holder, and none is lost when its subject leaves.
CREATE VIEW fm.v_review_stale AS
WITH subj AS (
SELECT 'claim' AS kind, source_key AS subject_key, assertion_sha256, site_uid,
SELECT operator_uid, 'claim' AS kind, source_key AS subject_key, assertion_sha256, holder_uid,
claim_status IN ('asserted','reviewed','disputed') AS live
FROM fm.fuel_movement
UNION ALL SELECT 'claim', source_key, assertion_sha256, site_uid, claim_status IN ('asserted','reviewed','disputed')
UNION ALL SELECT operator_uid, 'claim', source_key, assertion_sha256, holder_uid, claim_status IN ('asserted','reviewed','disputed')
FROM fm.stock_observation
UNION ALL SELECT 'claim', h.source_key, h.assertion_sha256, m.site_uid, h.claim_status IN ('asserted','reviewed','disputed')
UNION ALL SELECT h.operator_uid, 'claim', h.source_key, h.assertion_sha256, m.holder_uid, h.claim_status IN ('asserted','reviewed','disputed')
FROM fm.hour_meter_reading h JOIN fm.runtime_meter m ON m.meter_id = h.meter_id
UNION ALL SELECT 'claim', source_key, assertion_sha256, site_uid, claim_status IN ('asserted','reviewed','disputed')
UNION ALL SELECT operator_uid, 'claim', source_key, assertion_sha256, holder_uid, claim_status IN ('asserted','reviewed','disputed')
FROM fm.electricity_bill
UNION ALL SELECT 'claim', source_key, assertion_sha256, site_uid, claim_status IN ('asserted','reviewed','disputed')
UNION ALL SELECT operator_uid, 'claim', source_key, assertion_sha256, holder_uid, claim_status IN ('asserted','reviewed','disputed')
FROM fm.imported_calculation
UNION ALL SELECT 'claim', e.source_key, e.assertion_sha256, p.site_uid, e.claim_status IN ('asserted','reviewed','disputed')
UNION ALL SELECT e.operator_uid, 'claim', e.source_key, e.assertion_sha256, p.holder_uid, e.claim_status IN ('asserted','reviewed','disputed')
FROM fm.equipment_spec e LEFT JOIN fm.asset_placement p ON p.asset_uid = e.asset_uid
UNION ALL SELECT 'claim', source_key, assertion_sha256, site_uid, claim_status IN ('asserted','reviewed','disputed')
UNION ALL SELECT operator_uid, 'claim', source_key, assertion_sha256, holder_uid, claim_status IN ('asserted','reviewed','disputed')
FROM fm.runtime_interval
UNION ALL SELECT 'claim', t.source_key, t.assertion_sha256, p.site_uid, t.claim_status IN ('asserted','reviewed','disputed')
UNION ALL SELECT t.operator_uid, 'claim', t.source_key, t.assertion_sha256, p.holder_uid, t.claim_status IN ('asserted','reviewed','disputed')
FROM fm.engine_fuel_total t LEFT JOIN fm.asset_placement p ON p.asset_uid = t.asset_uid
UNION ALL SELECT 'location', source_key, assertion_sha256, site_uid, acceptance IN ('pending','accepted','rejected')
FROM fm.site_location
UNION ALL SELECT 'assumption', code, fingerprint, site_uid, status <> 'superseded'
UNION ALL SELECT l.operator_uid, 'location', l.source_key, l.assertion_sha256, hp.holder_uid, l.acceptance IN ('pending','accepted','rejected')
FROM fm.place_location l LEFT JOIN fm.holder_place hp ON hp.place_uid = l.place_uid
UNION ALL SELECT operator_uid, 'assumption', code, fingerprint, holder_uid, status <> 'superseded'
FROM fm.assumption)
SELECT r.review_id, r.entry_key, r.subject_kind, r.subject_key, r.assertion_sha256, r.verdict,
r.reviewed_on, r.reviewer_role, st.site_uid
SELECT r.operator_uid, r.review_id, r.entry_key, r.subject_kind, r.subject_key, r.assertion_sha256, r.verdict,
r.reviewed_on, r.reviewer_role, st.holder_uid
FROM fm.review r
LEFT JOIN LATERAL (SELECT DISTINCT s.site_uid FROM subj s
WHERE s.kind = r.subject_kind AND s.subject_key = r.subject_key AND s.site_uid IS NOT NULL) st ON true
LEFT JOIN LATERAL (SELECT DISTINCT s.holder_uid FROM subj s
WHERE s.operator_uid = r.operator_uid AND s.kind = r.subject_kind
AND s.subject_key = r.subject_key AND s.holder_uid IS NOT NULL) st ON true
WHERE NOT EXISTS (SELECT 1 FROM subj l
WHERE l.live AND l.kind = r.subject_kind AND l.subject_key = r.subject_key
AND l.assertion_sha256 = r.assertion_sha256);
WHERE l.live AND l.operator_uid = r.operator_uid AND l.kind = r.subject_kind
AND l.subject_key = r.subject_key AND l.assertion_sha256 = r.assertion_sha256);
-- Visits kept for provenance only: every claim superseded or retracted.
CREATE VIEW fm.v_visit_provenance_only AS