#!/usr/bin/env python3 """Generate ATOMA's dashboards for the Petrolyst app as one Superset 6.1 import bundle. Implements SPEC.md in this folder: the two dashboards the Petrolyst app embeds, "ATOMA · Overview" (atoma-overview) and "ATOMA · Site" (atoma-site, the four detail tabs), one dataset per public.v_* view and every chart. Nothing in the bundle names a site. build_bundle.py --database-uuid [--out atoma-dashboards.zip] The database uuid differs per Superset instance; `import_bundle.py --show-database` prints it. Output is deterministic: the same inputs give a byte-identical zip, and stable uuid5 ids make a re-import update the objects instead of copying them. Import with import_bundle.py, which overwrites datasets, charts and dashboards (the stock dashboard import leaves existing datasets and charts unchanged). Before writing, the bundle is checked: every column a chart or filter uses exists in its dataset, every chart in a Site filter's scope has site_label, the Overview sums recorded values only and money only per currency, and no row is wider than 12. Requires PyYAML. """ from __future__ import annotations import argparse import io import json import re import sys import uuid import zipfile from pathlib import Path from typing import Any import yaml HERE = Path(__file__).resolve().parent VERSION = "1.0.0" NAMESPACE = uuid.uuid5(uuid.NAMESPACE_URL, "https://petrolyst.com/superset/atoma-dashboards") DATABASE_NAME = "Fuel" # As superset-init registers it: fuel-db directly, as the read-only role fuel_ro. The password # comes from a secret file at connect time (superset_config.py), never from the bundle. DATABASE_URI = "postgresql+psycopg2://fuel_ro@fuel-db:5432/fuel" CATALOG, SCHEMA = "fuel", "public" SITE_FILTER = "NATIVE_FILTER-site" PERIOD_FILTER = "NATIVE_FILTER-period" # SPEC §3: neutral hues, no red or green, keyed by v_site.baseline_status_label. The # Petrolyst app's map uses the same colours (petrolyst-platform app/src/app/core/status.ts). STATUS_COLOURS = { "Approved baseline": "#009688", "Eligible - not yet approved": "#3987e5", "Provisional": "#c98500", "Insufficient evidence": "#64748B", } # SPEC §4 F3: a stock interval that is not "ok" is greyed; its status names the reason. EXCLUDED_GREY = "#A3A9B5" STOCK_STATUS_COLOURS = {status: EXCLUDED_GREY for status in ( "no_opening_or_closing", "endpoint_movement_order_unknown", "unquantified_movement", "same_day", "negative")} # Datasets with a time axis: the column the Viewing period applies to. MAIN_DTTM = { "v_delivery": "delivery_date", "v_stock_reading": "observed_on", "v_interval": "start_on", "v_runtime": "start_on", "v_hour_meter_evidence": "reading_date", "v_electricity_cost": "bill_month", } # Calculated columns: (name, PostgreSQL SQL, type, label). The Petrolyst server's site_status # (petrolyst-platform server/) shows the same two texts. CALCULATED = { "v_site": [ ("lph_text", "COALESCE(CAST(lph_estimate AS varchar), 'not available')", "VARCHAR", "L/h"), # SPEC §4: "Eligible" and "Provisional" carry "not approved". ("baseline_status_text", "CASE WHEN baseline_status = 'provisional' THEN baseline_status_label" " || ' (not approved)' ELSE baseline_status_label END", "VARCHAR", "Baseline status"), ], # PD-12 Overview: money is grouped by currency; a missing one is its own row. "v_delivery": [ ("currency_text", "COALESCE(currency, 'currency not recorded')", "VARCHAR", "Currency"), ], "v_electricity_cost": [ ("currency_text", "COALESCE(currency, 'currency not recorded')", "VARCHAR", "Currency"), ("currency_basis_text", "COALESCE(currency_basis, 'basis not recorded')", "VARCHAR", "Currency basis"), ], "v_stock_reading": [ ("reading_series", "reading_type || ', ' || value_kind || CASE WHEN seq_in_day > 1" " THEN ' (visit ' || CAST(seq_in_day AS varchar) || ')' ELSE '' END", "VARCHAR", "Reading"), ], } # Virtual datasets: name -> (view it extends, extra output columns, PostgreSQL SQL). # coverage_detail feeds S3. The view's lane names start with a sort number ("1 Stock # intervals"), which reads as a count on the chart, so it adds lane_name without the number # (the chart still sorts by lane). Its amount column is the tooltip: the litres delivered in a # stock interval (from its opening visit up to, not including, its closing visit; the opening # delivery is what the interval then uses), or a bill's amount. # # v_coverage carries dates, not visits or bills, so an amount is withheld wherever dates cannot # place it (Codex inspection round 3, ATOMA-08/09): a delivery on a boundary day that has more # than one visit could belong to either neighbouring interval, and bills that start on the same # day cannot be told apart. Everywhere else the date window is exact: on a single-visit day the # opening visit's delivery is included and the closing visit's excluded. VIRTUAL = { "coverage_detail": ("v_coverage", [("lane_name", "character varying(60)"), ("amount", "character varying(120)")], """\ SELECT c.site_uid, c.site_code, c.site_name, c.site_label, c.lane, c.status, c.start_on, c.end_on, c.start_ts, c.end_ts, regexp_replace(c.lane, '^[0-9]+ ', '') AS lane_name, CASE WHEN c.lane = '1 Stock intervals' THEN CASE WHEN f.ambiguous THEN 'not attributed: a delivery on a day with several visits (see Deliveries)' WHEN f.filled_l IS NULL AND coalesce(f.n_unknown, 0) = 0 THEN 'no delivery' WHEN f.filled_l IS NULL THEN 'filled, quantity not recorded' ELSE to_char(round(f.filled_l), 'FM999,999,999,990') || ' L filled' END || CASE WHEN f.n_unknown > 0 AND f.filled_l IS NOT NULL AND NOT f.ambiguous THEN ', plus ' || CAST(f.n_unknown AS varchar) || ' of unknown quantity' ELSE '' END WHEN c.lane = '4 Electricity bills' THEN CASE WHEN b.n_bills > 1 THEN 'several bills start this day (see Electricity bills)' WHEN b.disputed THEN 'bill disputed, amount withheld' WHEN b.amount IS NULL THEN 'bill amount not recorded' ELSE to_char(round(b.amount), 'FM999,999,999,990') || ' AFN billed (inferred)' END END AS amount FROM public.v_coverage c LEFT JOIN (SELECT s.site_uid, s.start_on, s.end_on, sum(CASE WHEN d.delivery_date < s.end_on THEN d.delivered_l END) AS filled_l, count(*) FILTER (WHERE d.delivery_date < s.end_on AND d.delivered_l IS NULL) AS n_unknown, bool_or(d.delivery_date IN (s.start_on, s.end_on) AND m.site_uid IS NOT NULL) AS ambiguous FROM (SELECT DISTINCT site_uid, start_on, end_on FROM public.v_coverage WHERE lane = '1 Stock intervals') s JOIN public.v_delivery d ON d.site_uid = s.site_uid AND d.delivery_date >= s.start_on AND d.delivery_date <= s.end_on LEFT JOIN (SELECT site_uid, day FROM (SELECT site_uid, delivery_date AS day, seq_in_day FROM public.v_delivery UNION ALL SELECT site_uid, observed_on, seq_in_day FROM public.v_stock_reading) x GROUP BY site_uid, day HAVING max(seq_in_day) > 1) m ON m.site_uid = d.site_uid AND m.day = d.delivery_date GROUP BY s.site_uid, s.start_on, s.end_on) f ON c.lane = '1 Stock intervals' AND f.site_uid = c.site_uid AND f.start_on = c.start_on AND f.end_on = c.end_on LEFT JOIN (SELECT site_uid, coalesce(period_start, bill_month) AS start_on, sum(amount) AS amount, bool_or(claim_status = 'disputed') AS disputed, count(*) AS n_bills FROM public.v_electricity_cost GROUP BY 1, 2) b ON c.lane = '4 Electricity bills' AND b.site_uid = c.site_uid AND b.start_on = c.start_on """), } # Saved metrics: (name, SQL, label). Every dataset also gets count. METRICS = { "v_delivery": [("litres_delivered", "SUM(delivered_l)", "Litres delivered")], "v_electricity_cost": [("amount", "SUM(amount)", "Amount (AFN, inferred)")], # SPEC §5: one eligibility filter on numerator and denominator. Group or filter # by metric and population_key wherever this is used. "v_interval": [( "pooled_l_per_h", "SUM(CASE WHEN eligibility <> 'excluded' THEN fuel_used_l END)" " / NULLIF(SUM(CASE WHEN eligibility <> 'excluded' THEN runtime_h END), 0)", "Pooled L/h (usable intervals)", )], } LABELS = { "site_code": "Code", "site_name": "Site name", "site_label": "Site", "site_uid": "Site uid", "pilot_status": "Pilot status", "baseline_status_label": "Baseline status", "baseline_status": "Baseline status code", "baseline_validity_note": "Validity note", "lph_estimate": "L/h", "lph_method": "L/h method", "lph_window_start": "L/h window start", "lph_window_end": "L/h window end", "baseline_decision_ref": "Decision", "fuel_l_per_calendar_day": "L/day, whole record", "latest_record_on": "Latest record", "location_status": "Location status", "location_precision_m": "Location precision (m)", "delivered_l": "Delivered (L)", "fuel_l_per_day": "L per calendar day", "l_per_h": "L/h", "l_per_h_low": "L/h low", "l_per_h_high": "L/h high", "current_baseline_l_per_h": "Approved baseline L/h", "fuel_used_l": "Used (L)", "fuel_used_unc_l": "Used, uncertainty (L)", "opening_l": "Opening (L)", "closing_l": "Closing (L)", "deliveries_l": "Deliveries (L)", "withdrawals_l": "Withdrawals (L)", "runtime_h": "Runtime (h)", "runtime_h_per_day": "Runtime (h/day)", "h_per_day": "h/day", "h_per_day_low": "h/day low", "h_per_day_high": "h/day high", "l_per_day": "L/day", "modelled_l_per_h": "Modelled L/h", "modelled_l_per_h_low": "Modelled L/h low", "modelled_l_per_h_high": "Modelled L/h high", "stock_balance_l": "Stock depletion, site (L)", "engine_l": "Engine-measured, this generator (L)", "engine_all_generators_l": "Engine-measured, all generators (L)", "all_generators_metered": "All generators metered", "covered_h": "Hours covered", "interval_h": "Interval hours", "difference_l": "Difference, site (L)", "difference_pct": "Difference, site (%)", "fuel_l": "Fuel (L)", "kwh": "kWh", "amount": "Amount", "value_kind": "Value kind", "claim_status": "Claim status", "interval_label": "Interval", "exclusion_reasons": "Exclusion reasons", "in_current_baseline": "In current baseline", "cached_value": "Workbook value", "unit_as_labelled": "Unit (as labelled)", "source_label": "Workbook label", "our_reading": "Our reading", "baseline_use": "Use", "make_model": "Make and model", "load_pct": "Load (%)", "value_text": "Value (text)", "assumption_code": "Assumption", "matched_days": "Matched days", "document_date": "Date", "used_for": "Used for", "check_code": "Check code", "label": "Check", "measure": "Measure", "lane_name": "Evidence", "seq_in_day": "Visit of the day", "event_key": "Event", "n_sources": "Sources", "sources_disagree": "Sources disagree", "delivery_date": "Date", } def label_for(column: str) -> str: return LABELS.get(column) or column.replace("_", " ").capitalize() def uid(kind: str, key: str) -> str: return str(uuid.uuid5(NAMESPACE, f"{kind}/{key}")) # --- Datasets ------------------------------------------------------------------- def read_columns(path: Path) -> dict[str, list[tuple[str, str]]]: views: dict[str, list[tuple[str, str]]] = {} for line in path.read_text(encoding="utf-8").splitlines(): if not line.strip() or line.startswith("#"): continue view, column, data_type = line.split("\t") views.setdefault(view, []).append((column, data_type)) return views def superset_type(data_type: str) -> str: return data_type.upper().replace(",", ", ") def dataset_yaml(view: str, columns: list[tuple[str, str]], database_uuid: str) -> dict[str, Any]: cols = [] for name, data_type in columns: temporal = data_type.startswith(("date", "timestamp")) cols.append(column_entry(name, superset_type(data_type), temporal, None)) for name, expression, data_type, verbose in CALCULATED.get(view, []): cols.append(column_entry(name, data_type, False, expression, verbose)) metrics = [metric_entry("count", "COUNT(*)", "Count")] metrics += [metric_entry(*m) for m in METRICS.get(view, [])] return { "table_name": view, "main_dttm_col": MAIN_DTTM.get(view), "currency_code_column": None, "description": None, "default_endpoint": None, "offset": 0, "cache_timeout": None, "catalog": CATALOG, "schema": SCHEMA, "sql": VIRTUAL[view][2] if view in VIRTUAL else None, "params": None, "template_params": None, "filter_select_enabled": True, "fetch_values_predicate": None, "extra": None, "normalize_columns": False, "always_filter_main_dttm": False, "folders": None, "uuid": uid("dataset", view), "metrics": metrics, "columns": cols, "version": VERSION, "database_uuid": database_uuid, } def column_entry(name: str, data_type: str, temporal: bool, expression: str | None, verbose: str | None = None) -> dict[str, Any]: return { "column_name": name, "verbose_name": verbose or label_for(name), "is_dttm": temporal, "is_active": True, "type": data_type, "advanced_data_type": None, "groupby": True, "filterable": True, "expression": expression, "description": None, "python_date_format": None, "datetime_format": None, "extra": None, } def metric_entry(name: str, expression: str, verbose: str) -> dict[str, Any]: return { "metric_name": name, "verbose_name": verbose, "metric_type": None, "expression": expression, "description": None, "d3format": None, "currency": None, "extra": None, "warning_text": None, } # --- Chart building blocks ---------------------------------------------------------- def sql_metric(expression: str, label: str) -> dict[str, Any]: return { "expressionType": "SQL", "sqlExpression": expression, "label": label, "hasCustomLabel": True, "optionName": "metric_" + re.sub(r"\W+", "_", label.lower()).strip("_"), "column": None, "aggregate": None, "datasourceWarning": False, } def sql_filter(expression: str) -> dict[str, Any]: return { "expressionType": "SQL", "sqlExpression": expression, "clause": "WHERE", "subject": None, "operator": None, "comparator": None, "isExtra": False, "isNew": False, "datasourceWarning": False, "filterOptionName": "filter_" + uuid.uuid5(NAMESPACE, expression).hex[:12], } def time_filter(column: str) -> dict[str, Any]: """An open TEMPORAL_RANGE filter: the Viewing period replaces its range.""" return { "expressionType": "SIMPLE", "subject": column, "operator": "TEMPORAL_RANGE", "comparator": "No filter", "clause": "WHERE", "sqlExpression": None, "isExtra": False, "isNew": False, "datasourceWarning": False, "filterOptionName": f"filter_period_{column}", } def order(*columns: str, ascending: bool = True) -> list[str]: return [json.dumps([c, ascending]) for c in columns] class Chart: """One chart: key (SPEC id), title, dataset, viz type, params, description.""" def __init__(self, key: str, title: str, view: str, viz: str, params: dict[str, Any], description: str | None = None, columns_used: list[str] | None = None) -> None: self.key, self.title, self.view, self.viz = key, title, view, viz self.params = {"viz_type": viz, "datasource": "0__table", **params} self.description = description self.columns_used = columns_used or [] @property def uuid(self) -> str: return uid("chart", self.key) def yaml(self, dataset_uuid: str) -> dict[str, Any]: return { "slice_name": self.title, "description": self.description, "certified_by": None, "certification_details": None, "viz_type": self.viz, "params": self.params, "query_context": None, "cache_timeout": None, "uuid": self.uuid, "version": VERSION, "dataset_uuid": dataset_uuid, } def table(key: str, title: str, view: str, columns: list[str], *, order_by: list[str] | None = None, filters: list[dict[str, Any]] | None = None, period: str | None = None, server_pages: bool = False, search: bool = False, html: bool = False, description: str | None = None) -> Chart: adhoc = list(filters or []) if period: adhoc.append(time_filter(period)) params = { "query_mode": "raw", "all_columns": columns, "groupby": [], "metrics": [], "percent_metrics": [], "order_by_cols": order_by or [], "adhoc_filters": adhoc, "row_limit": 10000, "server_pagination": server_pages, "server_page_length": 25, "page_length": 25, "include_search": search, "show_cell_bars": False, "color_pn": False, "allow_render_html": html, "table_timestamp_format": "%Y-%m-%d", "column_config": {}, "conditional_formatting": [], } used = columns + [json.loads(o)[0] for o in order_by or []] + ([period] if period else []) return Chart(key, title, view, "table", params, description, used) def bars(key: str, title: str, view: str, x: str, metrics: list[Any], *, groupby: list[str] | None = None, grain: str | None = None, period: str | None = None, y_title: str = "", filters: list[dict[str, Any]] | None = None, description: str | None = None) -> Chart: adhoc = list(filters or []) if period: adhoc.append(time_filter(period)) params = { **timeseries_common(x, metrics, groupby or [], grain, adhoc, y_title), "orientation": "vertical", "stack": "Stack" if groupby else None, "only_total": True, "show_value": False, } used = [x] + list(groupby or []) + ([period] if period else []) return Chart(key, title, view, "echarts_timeseries_bar", params, description, used) def summary_table(key: str, title: str, view: str, groupby: list[str], metrics: list[Any], *, period: str | None = None, description: str | None = None) -> Chart: """An aggregated table: one row per group, one column per metric.""" params = { "query_mode": "aggregate", "groupby": groupby, "metrics": metrics, "all_columns": [], "percent_metrics": [], "order_by_cols": [], "order_desc": True, "adhoc_filters": [time_filter(period)] if period else [], "row_limit": 1000, "server_pagination": False, "include_search": False, "show_cell_bars": False, "color_pn": False, "allow_render_html": False, "table_timestamp_format": "%Y-%m-%d", "column_config": {}, "conditional_formatting": [], } return Chart(key, title, view, "table", params, description, groupby + ([period] if period else [])) def big_number(key: str, title: str, view: str, metric: dict[str, Any], *, period: str | None = None, subheader: str = "", description: str | None = None) -> Chart: params = { "metric": metric, "adhoc_filters": [time_filter(period)] if period else [], "subheader": subheader, "header_font_size": 0.4, "subheader_font_size": 0.15, "y_axis_format": "SMART_NUMBER", "time_format": "smart_date", "force_timestamp_formatting": False, } return Chart(key, title, view, "big_number_total", params, description, [period] if period else []) def timeseries_common(x: str, metrics: list[Any], groupby: list[str], grain: str | None, adhoc: list[dict[str, Any]], y_title: str) -> dict[str, Any]: return { "x_axis": x, "time_grain_sqla": grain, "metrics": metrics, "groupby": groupby, "adhoc_filters": adhoc, "row_limit": 10000, "truncate_metric": True, "show_empty_columns": True, "comparison_type": "values", "annotation_layers": [], "color_scheme": "supersetColors", "show_legend": bool(groupby), "legendType": "scroll", "legendOrientation": "top", "x_axis_title": "", "x_axis_title_margin": 15, "y_axis_title": y_title, "y_axis_title_margin": 40, "y_axis_title_position": "Left", "x_axis_time_format": "%Y-%m-%d", "y_axis_format": "SMART_NUMBER", "rich_tooltip": True, "tooltipTimeFormat": "%Y-%m-%d", "truncateXAxis": True, "y_axis_bounds": [None, None], "zoomable": False, } # --- Overview (PD-12, embedded in the Petrolyst app) --------------------------------------- OVERVIEW_SLUG = "atoma-overview" SITE_SLUG = "atoma-site" # one embedded dashboard with the DETAIL_TABS as its tabs # SPEC §10: only sums and counts of recorded values, never an average of per-site rates; money # only grouped by currency. The value says whether the metric is money. OVERVIEW_METRICS = { "COUNT(*)": False, "SUM(delivered_l)": False, "COUNT(unit_price)": False, "SUM(delivered_l * unit_price)": True, "SUM(CASE WHEN COALESCE(claim_status, '') <> 'disputed' THEN amount END)": True, "SUM(CASE WHEN claim_status = 'disputed' THEN amount END)": True, } def overview_charts() -> list[Chart]: def m(expression: str, label: str) -> dict[str, Any]: assert expression in OVERVIEW_METRICS, expression return sql_metric(expression, label) return [ big_number("O3", "Litres delivered", "v_delivery", m("SUM(delivered_l)", "Litres delivered"), period="delivery_date", subheader="all sites, in the viewing period", description="Recorded deliveries. Deliveries are not consumption."), summary_table("O1", "Sites by pilot status", "v_site", ["pilot_status"], [m("COUNT(*)", "Sites")]), summary_table("O2", "Sites by baseline status", "v_site", ["baseline_status_label"], [m("COUNT(*)", "Sites")]), summary_table("O4", "Fuel cost by currency", "v_delivery", ["currency_text"], [m("SUM(delivered_l * unit_price)", "Cost (priced deliveries)"), m("COUNT(unit_price)", "Priced deliveries"), m("COUNT(*)", "Deliveries")], period="delivery_date", description="Litres times the recorded unit price, only where a price was " "recorded. Never added up across currencies."), summary_table("O5", "Electricity billed by currency", "v_electricity_cost", ["currency_text", "currency_basis_text"], [m("SUM(CASE WHEN COALESCE(claim_status, '') <> 'disputed' THEN amount END)", "Billed"), m("SUM(CASE WHEN claim_status = 'disputed' THEN amount END)", "Disputed"), m("COUNT(*)", "Bills")], period="bill_month", description="Grid cost as billed; disputed bills apart. Never converted to " "fuel or hours, never added up across currencies."), bars("O6", "Litres delivered per month", "v_delivery", "delivery_date", [m("SUM(delivered_l)", "Litres delivered")], grain="P1M", period="delivery_date", y_title="Litres", description="All sites' recorded deliveries per calendar month."), ] OVERVIEW_NOTES = ("**Reading this page.** All sites together, from recorded values only: " "deliveries are not consumption, and site rates (L/h, L/day) are never averaged " "here; each site's own page has them. Money is shown per currency and never added " "up across currencies. The viewing period narrows the deliveries, the costs and " "the monthly bars; the site counts are always the current state.") OVERVIEW_ROWS = [ [("MARKDOWN:notes-overview", 12, 14)], [("O3", 4, 40), ("O1", 4, 40), ("O2", 4, 40)], [("O4", 6, 40), ("O5", 6, 40)], [("O6", 12, 60)], ] OVERVIEW_PERIOD_SCOPE = {"O3", "O4", "O5", "O6"} # --- Site Detail ----------------------------------------------------------------------- def detail_charts() -> list[Chart]: return [ # Handlebars would render these as cards, but it compiles templates with # eval, which Superset's own CSP blocks. One-row tables need no CSP change. table("S1", "Fuel per running hour", "v_site", [ "lph_text", "lph_method", "lph_window_start", "lph_window_end", "baseline_status_text", "baseline_validity_note", "baseline_decision_ref", "baseline_conditions", "fuel_l_per_calendar_day"], description="L/h comes only from an approved or proposed baseline version. " "L/day is per calendar day over the whole record, not per running hour."), table("S2", "Supporting measures", "v_kpi_measures", [ "measure", "value", "unit", "value_kind", "window_start", "window_end", "note"], order_by=order("measure_order")), Chart("S3", "Evidence coverage", "coverage_detail", "gantt_chart", { "start_time": "start_ts", "end_time": "end_ts", "y_axis": "lane_name", "series": "status", "subcategories": False, "tooltip_metrics": [], "order_by_cols": order("lane"), "adhoc_filters": [], "row_limit": 10000, "color_scheme": "supersetColors", "show_legend": True, "legendType": "scroll", "legendOrientation": "top", "zoomable": True, "x_axis_time_format": "%Y-%m-%d", "tooltip_time_format": "%Y-%m-%d", "tooltip_columns": ["amount"], "y_axis_title": "Evidence", "y_axis_title_margin": 150, "y_axis_title_position": "Left", }, "Periods covered by each kind of evidence. Gaps are not interpolated.", ["start_ts", "end_ts", "lane", "lane_name", "status", "amount"]), bars("F1", "Litres delivered per day (all deliveries that day)", "v_delivery", "delivery_date", ["litres_delivered"], grain="P1D", period="delivery_date", y_title="Litres", description="Deliveries are not consumption. F5 lists each delivery."), table("F5", "Deliveries", "v_delivery", [ "delivery_date", "seq_in_day", "event_key", "delivered_l", "unit_price", "currency", "n_sources", "sources_disagree", "reviewed", "source_ref"], order_by=order("delivery_date", "seq_in_day"), period="delivery_date", search=True, description="Every delivery event, including one whose quantity is not known."), Chart("F2", "Recorded stock", "v_stock_reading", "echarts_timeseries_scatter", { # A second visit on the same day is its own series, so no reading is hidden # behind the day's MAX. **timeseries_common("observed_on", [sql_metric("MAX(litres)", "Litres")], ["reading_series"], "P1D", [time_filter("observed_on")], "Litres"), "markerSize": 6, }, "Points only; nothing is drawn between readings.", ["observed_on", "litres", "reading_series", "seq_in_day"]), bars("F3", "Fuel per calendar day, by stock interval", "v_interval", "interval_label", [sql_metric("MAX(fuel_l_per_day)", "L per calendar day")], groupby=["stock_status"], period="start_on", y_title="L/day", description="One bar per observed stock interval; nothing is spread over months. " "F4 lists the intervals with their status."), table("F4", "Stock intervals", "v_interval", [ "interval_label", "start_on", "end_on", "days", "opening_l", "opening_basis", "deliveries_l", "withdrawals_l", "closing_l", "closing_basis", "fuel_used_l", "fuel_l_per_day", "stock_status"], order_by=order("start_on"), period="start_on"), bars("R1", "Runtime per day, by meter interval", "v_runtime", "interval_label", [sql_metric("MAX(h_per_day)", "h/day")], groupby=["meter_label"], period="start_on", y_title="h/day", description="Empty until hour-meter readings exist. An invalid reading pair " "(two epochs, a decreasing timer, over 24 h/day) shows no bar."), table("R1T", "Runtime pairs and their validity", "v_runtime", [ "meter_label", "mapped_set", "interval_label", "start_on", "end_on", "days", "runtime_h", "h_per_day", "pair_valid", "same_epoch", "monotonic", "over_24h_per_day", "reviewed"], order_by=order("meter_label", "start_on"), period="start_on", description="Every pair of hour-meter readings, valid or not. R1 shows no bar " "for an invalid pair."), table("R2", "Hour-meter evidence", "v_hour_meter_evidence", [ "meter_label", "mapped_set", "reading_date", "date_basis", "hours", "reading_state", "epoch_no", "claim_status"], order_by=order("reading_date"), period="reading_date", description="Empty until hour-meter readings exist."), table("R3", "Grid energy (kWh)", "v_kpi_measures", [ "measure", "value", "unit", "value_kind", "note"], filters=[sql_filter("unit = 'kWh'")], description="Empty value: no kWh records. Bills are cost only and are not " "converted to fuel or hours."), bars("R4", "Electricity cost per bill month (AFN, inferred)", "v_electricity_cost", "bill_month", ["amount"], grain="P1M", period="bill_month", y_title="AFN", description="Grid cost only; never converted to fuel or hours."), table("R4T", "Electricity bills", "v_electricity_cost", [ "bill_month", "period_start", "period_end", "amount", "currency", "currency_basis", "kwh", "claim_status", "source_ref"], order_by=order("bill_month"), period="bill_month", description="A disputed bill is listed without amount or kWh."), table("B1", "Baseline versions", "v_baseline_version", [ "code", "metric", "status", "value", "range_low", "range_high", "unit", "window_start", "window_end", "matched_days", "meets_policy", "policy_check", "evidence_grade", "conditions", "approved_on", "approved_by_role", "decision_ref"], order_by=order("window_start"), description="Empty while no baseline version is defined."), table("B2", "Inputs of each version", "v_baseline_input", [ "baseline_code", "interval_label", "included", "exclusion_reason", "fuel_l", "runtime_h", "days", "eligibility_at_computation", "dependencies_changed"], order_by=order("baseline_code", "start_on"), description="Empty until a baseline version exists."), table("B3", "All intervals: eligibility and exclusions", "v_interval", [ "interval_label", "eligibility", "exclusion_reasons", "evidence_grade", "conditions", "metric", "population_key", "in_current_baseline"], order_by=order("start_on")), Chart("B4", "Interval L/h against the approved baseline", "v_interval", "mixed_timeseries", { "x_axis": "interval_label", "time_grain_sqla": None, "metrics": [sql_metric("MAX(l_per_h)", "Interval L/h")], "groupby": [], "adhoc_filters": [sql_filter("eligibility <> 'excluded'")], "row_limit": 10000, "truncate_metric": True, "comparison_type": "values", "metrics_b": [sql_metric("MAX(current_baseline_l_per_h)", "Approved baseline L/h")], "groupby_b": [], "adhoc_filters_b": [sql_filter("baseline_comparable = true")], "row_limit_b": 10000, "truncate_metric_b": True, "comparison_type_b": "values", "seriesType": "bar", "seriesTypeB": "line", "markerEnabledB": True, "markerSizeB": 6, # One y-axis for both (no dual axis). "yAxisIndex": 0, "yAxisIndexB": 0, "annotation_layers": [], "color_scheme": "supersetColors", "show_legend": True, "legendType": "scroll", "legendOrientation": "top", "y_axis_format": "SMART_NUMBER", "rich_tooltip": True, "x_axis_time_format": "%Y-%m-%d", "y_axis_title": "L/h", "y_axis_title_margin": 40, "y_axis_title_position": "Left", }, "Bars only for usable intervals; the line only for an approved baseline of the same " "metric and population. See S1 for the site's status.", ["interval_label", "l_per_h", "eligibility", "current_baseline_l_per_h", "baseline_comparable"]), table("B5", "Assumptions", "v_assumption", [ "code", "topic", "value", "value_low", "value_high", "unit", "applies_from", "applies_to", "basis", "derived_from_fuel", "status", "decision_ref", "rationale"], order_by=order("code")), table("B6", "Missing data and quality checks", "v_data_quality", [ "label", "status", "detail"], order_by=order("check_order")), table("B7", "Evidence references", "v_evidence", [ "title", "kind", "issuer", "document_date", "used_for", "document_key"], order_by=order("document_date")), table("B8", "Workbook calculations, assessed", "v_imported_calculation", [ "source_label", "formula", "cached_value", "unit_as_labelled", "our_reading", "baseline_use", "exclusion_reason", "claim_status", "cell_ref"], order_by=order("cell_ref")), table("B9", "Stock against engine-measured fuel", "v_reconciliation", [ "interval_label", "asset_uid", "engine_l", "covered_h", "interval_h", "placed_throughout", "stock_balance_l", "engine_all_generators_l", "all_generators_metered", "difference_l", "difference_pct", "stock_status", "endpoint_times_known"], order_by=order("start_on"), description="Empty until flow-meter data exist. The difference is shown only " "with full coverage and known visit times."), table("B10", "Equipment specifications: manufacturer and nameplate data, not site performance", "v_spec_reference", [ "slot", "make_model", "attribute", "load_pct", "value", "value_text", "unit", "basis", "applicability", "applicability_note", "claim_status"], order_by=order("slot", "attribute"), description="Manufacturer and nameplate data, not site performance."), table("B11", "Modelled L/h under an assumed runtime: not measured", "v_modelled_lph", [ "assumption_code", "status", "basis", "h_per_day", "h_per_day_low", "h_per_day_high", "window_start", "window_end", "matched_days", "l_per_day", "modelled_l_per_h", "modelled_l_per_h_low", "modelled_l_per_h_high", "value_kind"], order_by=order("assumption_code"), description="Modelled from an assumed runtime, not measured. It never reaches " "S1, the map or the sites list."), ] # Reading notes shown at the top of each tab (SPEC §4 "when data are missing"). TAB_NOTES = { "summary": "**Reading this tab.** L/h comes only from a baseline version; while no runtime " "evidence exists it is *not available*. L/day is per calendar day over the whole " "record, not per running hour. The coverage chart has up to four lanes: Stock " "intervals, Runtime (one per meter), Baseline window, Electricity bills. A " "lane missing from the chart means no coverage span could be formed: no evidence of " "that kind, or too few readings (see Recorded stock and Hour-meter evidence); gaps " "are not interpolated. Missing data and quality checks (Baseline & Evidence) list what " "is missing.", "fuel": "**Reading this tab.** Deliveries are not consumption; the bars add up each day's " "deliveries, and the Deliveries table lists each one, including any of unknown quantity. Recorded stock is shown as points only; nothing is drawn between " "readings. Fuel per calendar day is one bar per observed stock interval, never spread " "over months; a grey bar is an interval whose stock balance is not usable, and its " "status (legend, tooltip, Stock intervals table) names the reason. The Viewing period narrows every chart here (by delivery, reading or " "interval start date).", "runtime": "**Reading this tab.** Runtime and hour-meter evidence stay empty until hour-meter " "readings exist; an invalid reading pair (two epochs, a decreasing timer, over " "24 h/day) shows no bar and is listed with its flags in the pair table. Grid energy without a value means no kWh records. " "Electricity cost is grid cost only (AFN, inferred) and is never converted to fuel " "or hours; a disputed bill is listed without amount or kWh. The Viewing period " "narrows every chart here except grid energy.", "baseline": "**Reading this tab.** Baseline versions and their inputs stay empty until a " "version is proposed. Interval L/h shows bars only for usable intervals, and the " "line only for an approved baseline of the same metric and population. Stock " "against engine-measured fuel stays empty until flow-meter data exist; the " "difference is shown only with full coverage and known visit times. Modelled L/h " "comes from an assumed runtime, is not measured, and never reaches the site's L/h, " "the map or the sites list. The Viewing period does not apply to this tab.", } DETAIL_TABS = [ ("summary", "Summary", [ [("MARKDOWN:notes-summary", 12, 17)], [("S1", 12, 30)], [("S2", 12, 60)], [("S3", 12, 50)], ]), ("fuel", "Fuel & Deliveries", [ [("MARKDOWN:notes-fuel", 12, 17)], [("F1", 6, 60), ("F2", 6, 60)], [("F5", 12, 50)], [("F3", 12, 60)], [("F4", 12, 60)], ]), ("runtime", "Runtime & Electricity", [ [("MARKDOWN:notes-runtime", 12, 14)], [("R1", 6, 55), ("R2", 6, 55)], [("R1T", 12, 45)], [("R3", 6, 55), ("COLUMN:R4", 6, [("R4", 50), ("R4T", 50)])], ]), ("baseline", "Baseline & Evidence", [ [("MARKDOWN:notes-baseline", 12, 14)], [("B1", 6, 50), ("B2", 6, 50)], [("B3", 12, 50)], [("B4", 12, 55)], [("B5", 6, 50), ("B6", 6, 50)], [("B7", 6, 50), ("B8", 6, 50)], [("B9", 6, 50), ("B10", 6, 50)], [("B11", 12, 45)], ]), ] # SPEC §4: the Viewing period applies to the time-series charts of these two tabs only. PERIOD_SCOPE = {"F1", "F2", "F3", "F4", "F5", "R1", "R1T", "R2", "R4", "R4T"} def detail_filters(ids: dict[str, int]) -> list[dict[str, Any]]: site = select_filter(SITE_FILTER, "Site", "v_site", "site_label", multi=False, required=True, excluded=[]) # SPEC §4 fallback: "required" alone still lets the charts query every site while # the filter is empty (local test, 24 Oct 2026), so the first site is preselected. site["controlValues"]["defaultToFirstItem"] = True # What the filter editor writes with that option: charts wait for this filter's # value, and the first value is applied without "Apply filters". site["requiredFirst"] = True site["description"] = "Exactly one site. Search by code or name." period = period_filter(ids, PERIOD_SCOPE, "Narrows the fuel, runtime and electricity charts. " "Summary and Baseline & Evidence keep their fixed windows.") return [site, period] def period_filter(ids: dict[str, int], scope: set[str], description: str) -> dict[str, Any]: """The Viewing period, applied only to the charts named in scope.""" return { "id": PERIOD_FILTER, "name": "Viewing period", "filterType": "filter_time", "type": "NATIVE_FILTER", "targets": [{}], "controlValues": {"enableEmptyFilter": False}, "defaultDataMask": {"extraFormData": {}, "filterState": {}, "ownState": {}}, "cascadeParentIds": [], "scope": {"rootPath": ["ROOT_ID"], "excluded": sorted(i for k, i in ids.items() if k not in scope)}, "description": description, } def select_filter(fid: str, name: str, view: str, column: str, *, multi: bool, required: bool, excluded: list[int]) -> dict[str, Any]: return { "id": fid, "name": name, "filterType": "filter_select", "type": "NATIVE_FILTER", "targets": [{"datasetUuid": uid("dataset", view), "column": {"name": column}}], "controlValues": { "enableEmptyFilter": required, "defaultToFirstItem": False, "multiSelect": multi, "searchAllOptions": False, "inverseSelection": False, "creatable": False, "sortAscending": True, }, "defaultDataMask": {"extraFormData": {}, "filterState": {}, "ownState": {}}, "cascadeParentIds": [], # The frontend derives chartsInScope from this; excluded must be a list. "scope": {"rootPath": ["ROOT_ID"], "excluded": excluded}, "description": "", } # --- Dashboards --------------------------------------------------------------------------- class Layout: """Builds a dashboard position (v2) from rows of (key, width, height).""" def __init__(self, title: str, charts: dict[str, Chart]) -> None: self.charts = charts self.ids: dict[str, int] = {} self.nodes: dict[str, Any] = { "DASHBOARD_VERSION_KEY": "v2", "ROOT_ID": {"children": [], "id": "ROOT_ID", "type": "ROOT"}, "GRID_ID": {"children": [], "id": "GRID_ID", "parents": ["ROOT_ID"], "type": "GRID"}, "HEADER_ID": {"id": "HEADER_ID", "meta": {"text": title}, "type": "HEADER"}, } self.nodes["ROOT_ID"]["children"].append("GRID_ID") self.markdown: dict[str, str] = {} def add(self, node_id: str, node_type: str, parent: str, meta: dict[str, Any]) -> str: parents = self.nodes[parent].get("parents", []) + [parent] self.nodes[node_id] = {"children": [], "id": node_id, "meta": meta, "parents": parents, "type": node_type} self.nodes[parent]["children"].append(node_id) return node_id def chart(self, key: str, width: int, height: int, parent: str) -> None: chart = self.charts[key] self.ids[key] = len(self.ids) + 1 # placeholder id; the importer maps it by uuid self.add(f"CHART-{key}", "CHART", parent, { "chartId": self.ids[key], "height": height, "sliceName": chart.title, "uuid": chart.uuid, "width": width}) def rows(self, rows: list[list[tuple]], parent: str, prefix: str) -> None: for n, row in enumerate(rows, 1): if sum(item[1] for item in row) > 12: raise SystemExit(f"layout: row {prefix}-{n} is wider than 12") row_id = self.add(f"ROW-{prefix}-{n}", "ROW", parent, {"background": "BACKGROUND_TRANSPARENT"}) for key, width, rest in row: if key.startswith("MARKDOWN:"): name = key.split(":", 1)[1] self.add(f"MARKDOWN-{name}", "MARKDOWN", row_id, {"width": width, "height": rest, "code": self.markdown[name]}) elif key.startswith("COLUMN:"): col_id = self.add(f"COLUMN-{key.split(':', 1)[1]}", "COLUMN", row_id, {"background": "BACKGROUND_TRANSPARENT", "width": width}) for sub_key, height in rest: self.chart(sub_key, width, height, col_id) else: self.chart(key, width, rest, row_id) def tab(self, tab_id: str, title: str, tabs_id: str) -> str: return self.add(tab_id, "TAB", tabs_id, {"text": title, "defaultText": "Tab title", "placeholder": "Tab title"}) def tabs(self, tabs: list[tuple[str, str, list]], parent: str, tabs_id: str) -> None: tabs_id = self.add(tabs_id, "TABS", parent, {}) for slug, title, rows in tabs: self.rows(rows, self.tab(f"TAB-{slug}", title, tabs_id), slug) def dashboard_yaml(key: str, title: str, slug: str, layout: Layout, metadata: dict[str, Any], css: str = "") -> dict[str, Any]: return { "dashboard_title": title, "description": None, "css": css, "slug": slug, "certified_by": None, "certification_details": None, "published": True, "uuid": uid("dashboard", key), "position": layout.nodes, "metadata": metadata, "version": VERSION, } def base_metadata(layout: Layout, filters: list[dict[str, Any]], cross_filters: bool) -> dict[str, Any]: return { "color_scheme": "supersetColors", "refresh_frequency": 0, # Expanded descriptions never render in 6.1 (the dashboard loads description_markeddown, # the chart holder reads description_markdown), so each tab has a notes block instead. "expanded_slices": {}, "label_colors": dict(STATUS_COLOURS), "shared_label_colors": [], "map_label_colors": {}, "color_scheme_domain": [], "cross_filters_enabled": cross_filters, # Filters in Superset's left column (owner, 29 Oct); the Petrolyst app hides the column # when it embeds a dashboard and sets the filters itself. "filter_bar_orientation": "VERTICAL", "native_filter_configuration": filters, "chart_configuration": {}, "global_chart_configuration": {"scope": {"rootPath": ["ROOT_ID"], "excluded": []}}, "timed_refresh_immune_slices": [], "default_filters": "{}", "filter_scopes": {}, } def build(database_uuid: str, columns_path: Path) -> dict[str, Any]: """Return {path inside the bundle: yaml object}.""" views = read_columns(columns_path) files: dict[str, Any] = { "metadata.yaml": {"version": VERSION, "type": "Dashboard", "timestamp": "2026-10-24T00:00:00+00:00"}, "databases/Fuel.yaml": { "database_name": DATABASE_NAME, "sqlalchemy_uri": DATABASE_URI, "cache_timeout": None, "expose_in_sqllab": True, "allow_run_async": False, "allow_ctas": False, "allow_cvas": False, "allow_dml": False, "allow_file_upload": False, "extra": {"metadata_params": {}, "engine_params": {}, "metadata_cache_timeout": {}, "schemas_allowed_for_file_upload": []}, "impersonate_user": False, "uuid": database_uuid, "version": VERSION, }, } for name, (base, extra, _) in VIRTUAL.items(): views[name] = views[base] + extra for view in sorted(views): files[f"datasets/Fuel/{view}.yaml"] = dataset_yaml(view, views[view], database_uuid) # PD-12: the Petrolyst app embeds the Overview and one site dashboard, whose four detail # tabs are Superset's own (owner, 29 Oct). The app shows the site header and chooses site # and period. Admins open the same dashboards in Superset, with the filter column. overview = {c.key: c for c in overview_charts()} all_charts = {**{c.key: c for c in detail_charts()}, **overview} dashboards: list[tuple[Layout, dict[str, Chart], dict[str, Any]]] = [] for key, title in [(OVERVIEW_SLUG, "Overview"), (SITE_SLUG, "Site")]: page = Layout(f"ATOMA · {title}", all_charts) page.markdown.update({f"notes-{k}": v for k, v in TAB_NOTES.items()}) page.markdown["notes-overview"] = OVERVIEW_NOTES if key == OVERVIEW_SLUG: page.rows(OVERVIEW_ROWS, "GRID_ID", "overview") filters = [period_filter(page.ids, OVERVIEW_PERIOD_SCOPE, "Narrows the deliveries, the costs and the monthly bars.")] else: page.tabs(DETAIL_TABS, "GRID_ID", "TABS-detail") filters = detail_filters(page.ids) placed = {k: all_charts[k] for k in page.ids} page_metadata = base_metadata(page, filters, cross_filters=False) page_metadata["label_colors"].update(STOCK_STATUS_COLOURS) files[f"dashboards/{key}.yaml"] = dashboard_yaml(key, f"ATOMA · {title}", key, page, page_metadata) dashboards.append((page, placed, page_metadata)) for chart in all_charts.values(): name = re.sub(r"\W+", "_", f"{chart.key}_{chart.title}").strip("_") files[f"charts/{name}.yaml"] = chart.yaml(uid("dataset", chart.view)) check(files, views, dashboards, overview) return files def check(files: dict[str, Any], views: dict[str, list[tuple[str, str]]], dashboards: list[tuple[Layout, dict[str, Chart], dict[str, Any]]], overview: dict[str, Chart]) -> None: problems = [] columns = {v: {c for c, _ in cols} | {c[0] for c in CALCULATED.get(v, [])} for v, cols in views.items()} for view in list(MAIN_DTTM) + list(CALCULATED) + list(METRICS): if view not in columns: problems.append(f"{view}: configured but not in the column snapshot") for view, col in MAIN_DTTM.items(): if col not in columns.get(view, set()): problems.append(f"{view}: main time column {col} missing") for layout, charts, metadata in dashboards: placed = set(layout.ids) if placed != set(charts): problems.append(f"charts not placed or unknown: {sorted(placed ^ set(charts))}") for chart in charts.values(): missing = set(chart.columns_used) - columns.get(chart.view, set()) if missing: problems.append(f"{chart.key}: {sorted(missing)} not in {chart.view}") for f in metadata["native_filter_configuration"]: in_scope = [k for k, i in layout.ids.items() if i not in f["scope"]["excluded"]] for target in f["targets"]: col = target.get("column", {}).get("name") if col is None: continue for key in in_scope: if col not in columns.get(charts[key].view, set()): problems.append(f"{f['id']}: {key} ({charts[key].view}) lacks {col}") # SPEC §10: the Overview sums recorded values only, and money only per currency. for chart in overview.values(): metrics = chart.params.get("metrics") or [chart.params["metric"]] for metric in metrics: expression = metric["sqlExpression"] if expression not in OVERVIEW_METRICS: problems.append(f"{chart.key}: metric {expression!r} is not an allowed total") elif (OVERVIEW_METRICS[expression] and "currency_text" not in chart.params.get("groupby", [])): problems.append(f"{chart.key}: money must be grouped by currency_text") uuids = [obj["uuid"] for path, obj in files.items() if path != "metadata.yaml"] if len(uuids) != len(set(uuids)): problems.append("duplicate uuids") if problems: raise SystemExit("bundle check failed:\n " + "\n ".join(problems)) def write_zip(files: dict[str, Any], out: Path) -> None: buffer = io.BytesIO() with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as bundle: for path in sorted(files): text = yaml.safe_dump(files[path], sort_keys=False, allow_unicode=True, width=100) info = zipfile.ZipInfo(f"atoma_dashboards/{path}", date_time=(1980, 1, 1, 0, 0, 0)) info.compress_type = zipfile.ZIP_DEFLATED info.external_attr = 0o644 << 16 bundle.writestr(info, text) out.write_bytes(buffer.getvalue()) def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) parser.add_argument("--database-uuid", required=True, type=uuid.UUID, help='uuid of the "Fuel" database in the target Superset') parser.add_argument("--columns", type=Path, default=HERE / "columns.tsv") parser.add_argument("--out", type=Path, default=Path("atoma-dashboards.zip")) args = parser.parse_args(argv) files = build(str(args.database_uuid), args.columns) write_zip(files, args.out) counts = {k: sum(p.startswith(k + "/") for p in files) for k in ("datasets", "charts", "dashboards")} print(f"{args.out}: {counts['dashboards']} dashboards, {counts['charts']} charts, " f"{counts['datasets']} datasets") return 0 if __name__ == "__main__": sys.exit(main())