Petrolyst/petrolyst-infra
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
petrolyst-infra/stacks/fuel-monitoring/fuel-db/migrations/0002_fm_derivations.sql
T
Nasim SadatandClaude Opus 5.5 ab553526f6 fuel-db: customer on every row, place and holder for site, licence tables 
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>
2026-10-30 14:40:16 +01:00

518 lines
34 KiB
PL/PgSQL
Raw Permalink Blame History

-- 0002_fm_derivations.sql: internal derivations in schema `fm` (not readable by fuel_ro).
-- DRAFT, not applied. Applied after 0001 by a loader acting as fuel_owner (README in this folder).
-- Design: fuel-monitoring .context/plans/2026-10-24-fuel-db-baseline-dashboard.md, §4 and §5.1.
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
FROM fm.source_document;
-- The analytical visit set: visits that still carry a live claim. A visit whose claims were all
-- superseded or retracted stays in fm.visit for provenance only and is invisible from here on.
CREATE VIEW fm.v_visit_live AS
SELECT v.* FROM fm.visit v
WHERE EXISTS (SELECT 1 FROM fm.fuel_movement m WHERE m.visit_id = v.visit_id
AND m.claim_status IN ('asserted','reviewed','disputed'))
OR EXISTS (SELECT 1 FROM fm.stock_observation s WHERE s.visit_id = v.visit_id
AND s.claim_status IN ('asserted','reviewed','disputed'))
OR EXISTS (SELECT 1 FROM fm.hour_meter_reading h WHERE h.visit_id = v.visit_id
AND h.claim_status IN ('asserted','reviewed','disputed'));
-- One row per physical movement (event_key; trigger rule 10 keeps kind and tanks equal within it).
-- Value = the best-ranked asserted or reviewed claim. The spread covers every live claim, and a
-- disputed claim makes the event unreviewed. No value = an unquantified movement.
CREATE VIEW fm.v_movement_event AS
SELECT m.visit_id, m.event_key, m.kind, m.from_tank_uid, m.to_tank_uid,
(array_agg(m.litres ORDER BY r.rank, m.movement_id)
FILTER (WHERE m.claim_status IN ('asserted','reviewed') AND m.litres IS NOT NULL))[1] AS litres,
coalesce(max(m.litres) - min(m.litres), 0) AS spread_l,
bool_and(m.claim_status = 'reviewed') AS all_reviewed,
count(*) AS n_claims
FROM fm.fuel_movement m JOIN fm.v_source_rank r ON r.sha256 = m.source_sha256
WHERE m.claim_status IN ('asserted','reviewed','disputed')
GROUP BY m.visit_id, m.event_key, m.kind, m.from_tank_uid, m.to_tank_uid;
-- Quantified site-level deliveries per visit, summed over events (never over claims).
CREATE VIEW fm.v_visit_delivery AS
SELECT visit_id, sum(litres) AS delivered_l, sum(spread_l) AS delivered_spread_l,
bool_and(all_reviewed) AS all_reviewed, count(*) AS n_events
FROM fm.v_movement_event
WHERE kind = 'delivery' AND to_tank_uid IS NULL AND litres IS NOT NULL
GROUP BY visit_id;
-- Every other external movement event (internal transfers cancel out at site level). Between two
-- endpoints they are booked; AT an endpoint visit their order relative to the stock readings is
-- unknown, so the interval is excluded instead of guessed (see v_stock_interval).
CREATE VIEW fm.v_visit_other_movement AS
SELECT visit_id,
sum(litres) FILTER (WHERE kind IN ('withdrawal','drain','spill','loss_reported')) AS out_l,
coalesce(sum(spread_l) FILTER (WHERE kind IN ('withdrawal','drain','spill','loss_reported')), 0) AS out_spread_l,
count(*) AS n_external,
count(*) FILTER (WHERE litres IS NULL OR kind = 'external_in'
OR (kind = 'delivery' AND to_tank_uid IS NOT NULL)) AS n_unquantified,
bool_and(all_reviewed) AS all_reviewed
FROM fm.v_movement_event
WHERE kind <> 'internal_transfer'
AND NOT (kind = 'delivery' AND to_tank_uid IS NULL AND litres IS NOT NULL)
GROUP BY visit_id;
CREATE VIEW fm.v_visit_stock AS -- site-level stock per visit and phase; same claim rules
SELECT s.visit_id, s.phase,
(array_agg(s.litres ORDER BY r.rank, s.stock_obs_id)
FILTER (WHERE s.claim_status IN ('asserted','reviewed') AND s.litres IS NOT NULL))[1] AS litres,
max(s.litres) - min(s.litres) AS spread_l,
max(s.resolution_l) FILTER (WHERE s.litres IS NOT NULL) AS resolution_l,
bool_and(s.claim_status = 'reviewed') AS all_reviewed, -- every live claim, with a number or not
count(*) AS n_claims
FROM fm.stock_observation s JOIN fm.v_source_rank r ON r.sha256 = s.source_sha256
WHERE s.claim_status IN ('asserted','reviewed','disputed') AND s.scope = 'site_total'
GROUP BY s.visit_id, s.phase
HAVING count(*) FILTER (WHERE s.claim_status IN ('asserted','reviewed') AND s.litres IS NOT NULL) > 0;
-- no asserted or reviewed number (only disputed, illegible or placeholder claims): no reading
-- 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.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'
WHEN a.litres - d.delivered_l IS NOT NULL THEN 'derived' END AS arrive_basis,
CASE WHEN b.litres IS NOT NULL THEN b.resolution_l WHEN st.litres IS NOT NULL THEN st.resolution_l
ELSE a.resolution_l END AS arrive_res_l,
CASE WHEN b.litres IS NOT NULL THEN b.spread_l WHEN st.litres IS NOT NULL THEN st.spread_l
ELSE a.spread_l + d.delivered_spread_l END AS arrive_spread_l,
CASE WHEN b.litres IS NOT NULL THEN b.all_reviewed WHEN st.litres IS NOT NULL THEN st.all_reviewed
ELSE a.all_reviewed AND d.all_reviewed END AS arrive_reviewed,
coalesce(a.litres, st.litres, b.litres + d.delivered_l) AS depart_l,
CASE WHEN coalesce(a.litres, st.litres) IS NOT NULL THEN 'stated'
WHEN b.litres + d.delivered_l IS NOT NULL THEN 'derived' END AS depart_basis,
CASE WHEN a.litres IS NOT NULL THEN a.resolution_l WHEN st.litres IS NOT NULL THEN st.resolution_l
ELSE b.resolution_l END AS depart_res_l,
CASE WHEN a.litres IS NOT NULL THEN a.spread_l WHEN st.litres IS NOT NULL THEN st.spread_l
ELSE b.spread_l + d.delivered_spread_l END AS depart_spread_l,
CASE WHEN a.litres IS NOT NULL THEN a.all_reviewed WHEN st.litres IS NOT NULL THEN st.all_reviewed
ELSE b.all_reviewed AND d.all_reviewed END AS depart_reviewed,
a.litres - (b.litres + d.delivered_l) AS after_vs_before_plus_delivered_l -- QA
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 o ON o.visit_id = v.visit_id
LEFT JOIN fm.v_visit_stock b ON b.visit_id = v.visit_id AND b.phase = 'before_delivery'
LEFT JOIN fm.v_visit_stock a ON a.visit_id = v.visit_id AND a.phase = 'after_delivery'
LEFT JOIN fm.v_visit_stock st ON st.visit_id = v.visit_id AND st.phase = 'standalone';
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.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.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,
c.arrive_l AS closing_l, c.arrive_basis AS closing_basis,
coalesce(o.depart_res_l,0) + coalesce(c.arrive_res_l,0) AS endpoint_res_l,
coalesce(o.depart_spread_l,0) + coalesce(c.arrive_spread_l,0) AS endpoint_spread_l,
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.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,
p.endpoint_res_l + p.endpoint_spread_l + coalesce(mid.spread_l,0) AS consumed_unc_l,
(p.endpoint_spread_l + coalesce(mid.spread_l,0)) > 0 AS inputs_conflict, -- any disagreement, as v_data_quality
p.endpoints_reviewed AND coalesce(mid.all_reviewed, true) AS inputs_reviewed,
CASE WHEN p.opening_l IS NULL OR p.closing_l IS NULL THEN 'no_opening_or_closing'
WHEN p.n_endpoint_movements > 0 THEN 'endpoint_movement_order_unknown'
WHEN mid.n_unknown > 0 THEN 'unquantified_movement'
WHEN p.days = 0 THEN 'same_day'
WHEN p.opening_l + coalesce(mid.delivered_l,0) - coalesce(mid.out_l,0) - p.closing_l
< -(p.endpoint_res_l + p.endpoint_spread_l + coalesce(mid.spread_l,0)) THEN 'negative'
ELSE 'ok' END AS stock_status
-- mid: live visits strictly between the endpoints. A visit counts only through its live movement
-- events, never through its kind.
FROM pr p
CROSS JOIN LATERAL (
SELECT sum(d.delivered_l) AS delivered_l,
sum(coalesce(d.delivered_spread_l, 0) + coalesce(x.out_spread_l, 0)) AS spread_l,
sum(x.out_l) AS out_l,
bool_and(coalesce(d.all_reviewed, true) AND coalesce(x.all_reviewed, true)) AS all_reviewed,
coalesce(sum(x.n_unquantified), 0) AS n_unknown
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.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;
CREATE VIEW fm.v_visit_reading AS -- one reading per visit and meter, with the claims' spread
SELECT h.visit_id, h.meter_id,
min(h.epoch_id) FILTER (WHERE h.hours IS NOT NULL) AS epoch_id,
count(DISTINCT h.epoch_id) FILTER (WHERE h.hours IS NOT NULL) AS n_epochs,
count(*) FILTER (WHERE h.hours IS NOT NULL AND h.epoch_id IS NULL) AS n_unassigned_epoch,
(array_agg(h.hours ORDER BY r.rank, h.reading_id)
FILTER (WHERE h.claim_status IN ('asserted','reviewed') AND h.hours IS NOT NULL))[1] AS hours,
min(h.hours) AS hours_min, max(h.hours) AS hours_max,
max(h.resolution_h) FILTER (WHERE h.hours IS NOT NULL) AS resolution_h,
bool_or(h.date_basis <> 'stated') FILTER (WHERE h.hours IS NOT NULL) AS date_inferred,
bool_and(h.claim_status = 'reviewed') AS all_reviewed -- every live claim, with a number or not
FROM fm.hour_meter_reading h JOIN fm.v_source_rank r ON r.sha256 = h.source_sha256
WHERE h.claim_status IN ('asserted','reviewed','disputed')
GROUP BY h.visit_id, h.meter_id
HAVING count(*) FILTER (WHERE h.claim_status IN ('asserted','reviewed') AND h.hours IS NOT NULL) > 0;
-- no asserted or reviewed number (only disputed or illegible claims): no reading
-- M2 per stock interval. Only a validated population is summed (CTE sel):
-- * a counter CONFIRMED for a generator placed at the site in the interval, when that generator
-- has exactly one such counter and no idle assumption;
-- * counters not tied to a set (unmapped, or mapping only proposed), only when the one live
-- meter_coverage assumption for the dates names exactly these counters, exactly the generators
-- left uncovered and exactly the idle sets. Such intervals give only the summed-set-hour metric.
-- A counter confirmed for a generator NOT placed at the site in the interval is never summed and
-- excludes the interval. Runtime and L/h are reported only for a valid population.
CREATE VIEW fm.v_lph_interval AS
WITH gen AS ( -- generators placed at the site at any time in the interval; full_cover =
-- one placement spans the whole interval, both endpoint dates included
SELECT i.interval_key, p.asset_uid,
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.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, '[]')
GROUP BY i.interval_key, p.asset_uid),
pair AS ( -- every counter read at both endpoints, classified
SELECT i.interval_key, m.meter_id,
CASE WHEN m.mapping_status = 'confirmed' THEN m.asset_uid END AS asset_uid,
CASE WHEN m.mapping_status <> 'confirmed' THEN 'not_tied'
WHEN EXISTS (SELECT 1 FROM gen g
WHERE g.interval_key = i.interval_key AND g.asset_uid = m.asset_uid) THEN 'tied'
ELSE 'tied_elsewhere' END AS tie,
(o.n_epochs = 1 AND c.n_epochs = 1 AND o.n_unassigned_epoch = 0 AND c.n_unassigned_epoch = 0
AND o.epoch_id = c.epoch_id) AS same_epoch,
c.hours - o.hours AS runtime_h,
c.hours_min - o.hours_max - o.resolution_h - c.resolution_h AS runtime_low_h,
c.hours_max - o.hours_min + o.resolution_h + c.resolution_h AS runtime_high_h,
o.date_inferred OR c.date_inferred AS date_inferred,
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.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.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 (
SELECT g.interval_key, g.asset_uid, g.full_cover,
count(DISTINCT p.meter_id) AS n_counters,
count(DISTINCT d.code) > 0 AS is_idle,
coalesce(bool_or(d.status = 'proposed'), false) AS idle_proposed
FROM gen g
LEFT JOIN pair p ON p.interval_key = g.interval_key AND p.asset_uid = g.asset_uid AND p.tie = 'tied'
LEFT JOIN idle d ON d.interval_key = g.interval_key AND d.asset_uid = g.asset_uid
GROUP BY g.interval_key, g.asset_uid, g.full_cover),
cov AS (
SELECT interval_key, count(*) AS n_generators,
count(*) FILTER (WHERE NOT full_cover) AS n_partial,
count(*) FILTER (WHERE n_counters > 1) AS n_double_counted,
count(*) FILTER (WHERE n_counters >= 1 AND is_idle) AS n_metered_and_idle,
count(*) FILTER (WHERE n_counters = 0 AND NOT is_idle) AS n_uncovered,
coalesce(array_agg(asset_uid ORDER BY asset_uid)
FILTER (WHERE n_counters = 0 AND NOT is_idle), '{}') AS uncovered_set,
coalesce(array_agg(asset_uid ORDER BY asset_uid) FILTER (WHERE is_idle), '{}') AS idle_set,
bool_or(idle_proposed) AS idle_proposed
FROM per_gen GROUP BY interval_key),
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.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)
FILTER (WHERE m.member_role = 'covered_generator'), '{}') AS covered,
coalesce(array_agg(m.asset_uid ORDER BY m.asset_uid)
FILTER (WHERE m.member_role = 'idle_generator'), '{}') AS idle_set
FROM fm.assumption x JOIN fm.assumption_member m ON m.assumption_id = x.assumption_id
WHERE x.topic = 'meter_coverage' AND x.status IN ('proposed','accepted')
GROUP BY x.assumption_id),
unm AS ( -- at most one live coverage assumption per site and date (EXCLUDE)
SELECT l.interval_key, cardinality(l.meters) AS n_unmapped, c.code AS cov_code, c.status AS cov_status,
c.covered AS cov_covered,
coalesce(c.meters = l.meters
AND c.covered = coalesce(cv.uncovered_set, '{}')
AND c.idle_set = coalesce(cv.idle_set, '{}')
AND cardinality(c.meters) = cardinality(c.covered), false) AS cov_matches
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.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
JOIN per_gen g ON g.interval_key = p.interval_key AND g.asset_uid = p.asset_uid
WHERE p.tie = 'tied' AND g.n_counters = 1 AND NOT g.is_idle
UNION ALL
SELECT p.* FROM pair p JOIN unm u ON u.interval_key = p.interval_key
WHERE p.tie = 'not_tied' AND u.cov_matches),
agg AS (
SELECT interval_key, count(*) AS n_meters, sum(runtime_h) AS runtime_h,
sum(runtime_low_h) AS runtime_low_h, sum(runtime_high_h) AS runtime_high_h,
bool_and(same_epoch) AS same_epoch, bool_and(runtime_h >= 0) AS monotonic,
max(runtime_h) AS max_meter_h, bool_or(date_inferred) AS date_inferred,
bool_and(reviewed) AS runtime_reviewed, bool_or(conflict) AS runtime_conflict,
string_agg(coalesce(asset_uid, 'meter:' || meter_id), ','
ORDER BY coalesce(asset_uid, 'meter:' || meter_id)) AS population_key
FROM sel GROUP BY interval_key),
rd AS ( -- what was read at both ends, before validation
SELECT interval_key, count(*) FILTER (WHERE tie = 'tied_elsewhere') AS n_tied_elsewhere
FROM pair GROUP BY interval_key),
pv AS ( -- valid = the summed population covers every placed generator exactly once
SELECT rd.interval_key,
coalesce(cv.n_generators > 0 AND rd.n_tied_elsewhere = 0 AND cv.n_partial = 0
AND cv.n_double_counted = 0 AND cv.n_metered_and_idle = 0
AND CASE WHEN u.interval_key IS NULL THEN cv.n_uncovered = 0 ELSE u.cov_matches END
AND a.n_meters >= 1, false) AS population_valid
FROM rd LEFT JOIN cov cv USING (interval_key) LEFT JOIN unm u USING (interval_key)
LEFT JOIN agg a USING (interval_key)),
x AS (
SELECT i.*,
coalesce(pv.population_valid, false) AS population_valid,
a.n_meters, u.cov_code,
CASE WHEN pv.population_valid THEN a.runtime_h END AS runtime_h,
CASE WHEN pv.population_valid THEN a.runtime_low_h END AS runtime_low_h,
CASE WHEN pv.population_valid THEN a.runtime_high_h END AS runtime_high_h,
CASE WHEN pv.population_valid THEN -- counters not tied to a set also name the sets they cover
(SELECT string_agg(k, ',' ORDER BY k)
FROM unnest(string_to_array(a.population_key, ',')
|| CASE WHEN u.cov_matches THEN array(SELECT 'covers:' || g FROM unnest(u.cov_covered) g)
ELSE '{}'::text[] END) k) END AS population_key,
a.date_inferred, a.runtime_reviewed, a.runtime_conflict,
coalesce(cv.n_generators, 0) AS n_generators, coalesce(u.n_unmapped, 0) AS n_unmapped,
coalesce(cv.idle_proposed, false)
OR (coalesce(u.cov_matches, false) AND u.cov_status = 'proposed') AS depends_on_proposed,
concat_ws('; ',
CASE WHEN i.stock_status <> 'ok' THEN 'stock interval: ' || i.stock_status END,
CASE WHEN rd.interval_key IS NULL THEN 'no timer reading at both ends' END,
CASE WHEN coalesce(cv.n_generators, 0) = 0 THEN 'no generator placed at the site' END,
CASE WHEN cv.n_partial > 0
THEN 'a generator was placed at or removed from the site within the interval' END,
CASE WHEN rd.n_tied_elsewhere > 0
THEN 'a counter is confirmed for a generator not placed here in the interval' END,
CASE WHEN cv.n_double_counted > 0 THEN 'two counters for one generator' END,
CASE WHEN cv.n_metered_and_idle > 0 THEN 'a generator is both metered and assumed idle' END,
CASE WHEN u.interval_key IS NULL AND cv.n_uncovered > 0
THEN 'a generator has neither a counter nor an idle assumption' END,
CASE WHEN u.interval_key IS NOT NULL AND u.cov_code IS NULL
THEN 'counters not tied to sets, and no meter-coverage assumption for the whole interval' END,
CASE WHEN u.cov_code IS NOT NULL AND NOT u.cov_matches
THEN 'meter-coverage assumption ' || u.cov_code
|| ' does not name exactly these counters, uncovered sets and idle sets' END,
CASE WHEN NOT a.same_epoch THEN 'readings not in one meter epoch' END,
CASE WHEN NOT a.monotonic THEN 'timer went backwards' END,
CASE WHEN a.max_meter_h > 24 * i.days THEN 'over 24 h/day on one meter' END,
CASE WHEN a.runtime_h = 0 AND i.consumed_l > 0 THEN 'fuel used with zero runtime' END,
CASE WHEN a.runtime_h = 0 AND i.consumed_l <= 0 THEN 'no operation in the interval' END,
CASE WHEN i.days < 7 THEN 'shorter than 7 days' END) AS exclusion_reasons
FROM fm.v_stock_interval i
LEFT JOIN rd USING (interval_key) LEFT JOIN agg a USING (interval_key)
LEFT JOIN cov cv USING (interval_key) LEFT JOIN unm u USING (interval_key)
LEFT JOIN pv USING (interval_key))
SELECT x.*,
CASE WHEN NOT population_valid THEN NULL
WHEN n_unmapped = 0 AND n_meters = 1 THEN 'apparent_l_per_generator_h'
ELSE 'apparent_l_per_set_h' END AS metric,
CASE WHEN runtime_h > 0 THEN consumed_l / runtime_h END AS l_per_h,
CASE WHEN runtime_high_h > 0 THEN (consumed_l - consumed_unc_l) / runtime_high_h END AS l_per_h_low,
CASE WHEN runtime_low_h > 0 THEN (consumed_l + consumed_unc_l) / runtime_low_h END AS l_per_h_high,
-- NULL beside a non-NULL l_per_h = unbounded
CASE WHEN exclusion_reasons <> '' OR NOT population_valid THEN 'excluded'
WHEN depends_on_proposed THEN 'conditional' ELSE 'eligible' END AS eligibility,
CASE WHEN opening_basis = 'stated' AND closing_basis = 'stated' AND inputs_reviewed AND NOT inputs_conflict
AND NOT date_inferred AND runtime_reviewed AND NOT runtime_conflict THEN 'G1' ELSE 'G2' END AS evidence_grade
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.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,
CASE WHEN i.inputs_reviewed AND NOT i.inputs_conflict THEN 'G1' ELSE 'G2' END AS evidence_grade
FROM fm.v_stock_interval i;
-- Everything an interval's value and eligibility rest on, as sorted lists, with their hashes. Claims
-- are listed in every status, so a review, supersession or retraction changes the hash. `baseline
-- propose` snapshots both lists; fm.v_baseline_validity compares them with today's state.
CREATE VIEW fm.v_interval_dependency AS
WITH span AS ( -- the endpoint visits and every visit between them, live or not
SELECT i.interval_key, v.visit_id
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.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 (
SELECT s.interval_key, jsonb_agg(d.item ORDER BY d.item) AS deps
FROM span s CROSS JOIN LATERAL (
SELECT 'stock:' || x.source_key || ':' || x.assertion_sha256 || ':' || x.claim_status AS item
FROM fm.stock_observation x WHERE x.visit_id = s.visit_id
UNION ALL
SELECT 'movement:' || x.event_key || ':' || x.assertion_sha256 || ':' || x.claim_status
FROM fm.fuel_movement x WHERE x.visit_id = s.visit_id) d
GROUP BY s.interval_key),
runtime_deps AS (
SELECT i.interval_key, jsonb_agg(d.item ORDER BY d.item) AS deps
FROM fm.v_stock_interval i CROSS JOIN LATERAL (
SELECT 'reading:' || x.meter_id || ':' || x.assertion_sha256 || ':' || x.claim_status
|| ':' || coalesce(x.epoch_id::text, '-') AS item
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.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.holder_uid = i.holder_uid
UNION ALL -- placements at this site, and anywhere of the assets its meters name
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.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.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.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
FROM fm.v_stock_interval i
LEFT JOIN stock_deps s USING (interval_key) LEFT JOIN runtime_deps r USING (interval_key);
-- Does each version still rest on what it was computed from? An input whose interval no longer
-- exists, or whose dependency hash differs, makes the whole version "inputs changed".
CREATE VIEW fm.v_baseline_validity AS
SELECT b.baseline_id,
count(bi.baseline_id) > 0
AND coalesce(bool_and(d.interval_key IS NOT NULL
AND d.stock_dep_sha256 = bi.stock_dep_sha256
AND (b.metric = 'fuel_l_per_calendar_day'
OR d.runtime_dep_sha256 = bi.runtime_dep_sha256)), false) AS inputs_unchanged
FROM fm.baseline_version b
LEFT JOIN fm.baseline_input bi ON bi.baseline_id = b.baseline_id
LEFT JOIN fm.v_interval_dependency d
ON d.open_visit_id = bi.open_visit_id AND d.close_visit_id = bi.close_visit_id
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.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.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,
(o.n_epochs = 1 AND c.n_epochs = 1 AND o.n_unassigned_epoch = 0 AND c.n_unassigned_epoch = 0
AND o.epoch_id = c.epoch_id) AS same_epoch,
c.hours >= o.hours AS monotonic,
c.hours - o.hours > 24 * (c.visit_on - o.visit_on) AS over_24h_per_day,
o.all_reviewed AND c.all_reviewed AS reviewed
FROM r o JOIN r c ON c.meter_id = o.meter_id AND c.n = o.n + 1;
-- M3, modelled L/h: L per calendar day over the usable intervals that lie wholly inside the
-- 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.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,
m4.fuel_l / NULLIF(m4.days, 0) / NULLIF(x.high_num, 0) AS modelled_l_per_h_low,
m4.fuel_l / NULLIF(m4.days, 0) / NULLIF(x.low_num, 0) AS modelled_l_per_h_high
FROM fm.assumption x
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.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 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 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 operator_uid, 'claim', source_key, assertion_sha256, holder_uid, claim_status IN ('asserted','reviewed','disputed')
FROM fm.stock_observation
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 operator_uid, 'claim', source_key, assertion_sha256, holder_uid, claim_status IN ('asserted','reviewed','disputed')
FROM fm.electricity_bill
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 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 operator_uid, 'claim', source_key, assertion_sha256, holder_uid, claim_status IN ('asserted','reviewed','disputed')
FROM fm.runtime_interval
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 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.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.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.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
SELECT v.* FROM fm.visit v
WHERE NOT EXISTS (SELECT 1 FROM fm.v_visit_live l WHERE l.visit_id = v.visit_id);
COMMIT;
Reference in new issue View Git Blame Copy Permalink