Add a Grafana dashboard generator for the three UPS devices
tools/generate_grafana_ups.py writes grafana-ups.json, which shows the JBD network UPS, the River 2 desktop UPS, and the River 3 NAS UPS in three columns. Git ignores the generated file. Each EcoFlow column shows one time-remaining panel in hours, for the direction in use. The River 2 histories add pack current and remaining capacity. The River 3 reports neither, so its histories add battery power and charge.
This commit is contained in:
@@ -0,0 +1,229 @@
|
||||
"""Regression tests for the generated Grafana v2 dashboard, without Grafana I/O."""
|
||||
|
||||
import importlib.util
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def generator():
|
||||
spec = importlib.util.spec_from_file_location(
|
||||
"generate_grafana_ups", ROOT / "tools/generate_grafana_ups.py",
|
||||
)
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def panels(generator):
|
||||
return {
|
||||
panel["spec"]["title"]: panel["spec"]
|
||||
for panel in generator.build_dashboard()["elements"].values()
|
||||
}
|
||||
|
||||
|
||||
def queries(panel):
|
||||
return {
|
||||
query["spec"]["refId"]: query["spec"]["query"]["spec"]["query"]
|
||||
for query in panel["data"]["spec"]["queries"]
|
||||
}
|
||||
|
||||
|
||||
def properties_for(panel, ref):
|
||||
overrides = panel["vizConfig"]["spec"]["fieldConfig"]["overrides"]
|
||||
return {
|
||||
prop["id"]: prop["value"]
|
||||
for override in overrides
|
||||
if override["matcher"] == {"id": "byFrameRefID", "options": ref}
|
||||
for prop in override["properties"]
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("title,primary,label,unit,secondary,secondary_unit,display_name", [
|
||||
("Network · Cell voltage history", "bms_voltage_cells_volts", "cell", "volt",
|
||||
"bms_current_amps", "amp", "Current"),
|
||||
("Network · Temperature history", "bms_temperature_celsius", "sensor", "celsius",
|
||||
"bms_capacity_remaining_ah", "amph", "Remaining Capacity"),
|
||||
])
|
||||
def test_network_histories_keep_original_series_and_separate_axes(
|
||||
panels, title, primary, label, unit, secondary, secondary_unit, display_name,
|
||||
):
|
||||
panel = panels[title]
|
||||
query = queries(panel)
|
||||
assert set(query) == {"A", "B"}
|
||||
assert f'r["_measurement"] == "{primary}"' in query["A"]
|
||||
assert f'r["_measurement"] == "{secondary}"' in query["B"]
|
||||
# Cell and sensor identities survive grouping and column selection.
|
||||
keys = ["ups", "_measurement", "_field", label]
|
||||
assert f'group(columns: {json.dumps(keys)})' in query["A"]
|
||||
assert f'keep(columns: {json.dumps(["_time", "_value"] + keys)})' in query["A"]
|
||||
defaults = panel["vizConfig"]["spec"]["fieldConfig"]["defaults"]
|
||||
assert defaults["unit"] == unit
|
||||
assert "${__field.labels." + label + "}" in defaults["displayName"]
|
||||
secondary_properties = properties_for(panel, "B")
|
||||
assert secondary_properties["unit"] == secondary_unit
|
||||
assert secondary_properties["displayName"] == display_name
|
||||
assert secondary_properties["custom.axisPlacement"] == "right"
|
||||
assert secondary_properties["custom.axisLabel"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("title", [
|
||||
"Network · Cell voltage history", "Network · Temperature history",
|
||||
])
|
||||
def test_network_histories_do_not_invent_empty_intervals(panels, title):
|
||||
panel = panels[title]
|
||||
for query in queries(panel).values():
|
||||
assert 'from(bucket: "${bucket:raw}")' in query
|
||||
assert 'r["ups"] == "network"' in query
|
||||
assert 'r["_field"] == "value"' in query
|
||||
assert 'range(start: v.timeRangeStart, stop: v.timeRangeStop)' in query
|
||||
assert 'sort(columns: ["_time"])' in query
|
||||
assert 'aggregateWindow(' not in query
|
||||
assert 'last()' not in query
|
||||
assert 'fill(' not in query
|
||||
assert 'map(' not in query # In particular, do not change the current's sign.
|
||||
custom = panel["vizConfig"]["spec"]["fieldConfig"]["defaults"]["custom"]
|
||||
assert custom["insertNulls"] is False
|
||||
assert custom["spanNulls"] is False # Keep actual nulls, rather than hiding them.
|
||||
assert custom["drawStyle"] == "line"
|
||||
assert custom["lineInterpolation"] == "linear"
|
||||
assert panel["data"]["spec"]["transformations"] == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize("device,title", [("desktop", "Desktop"), ("nas", "NAS")])
|
||||
def test_ecoflow_histories_average_each_display_interval(panels, device, title):
|
||||
for suffix, refs in (("Temperature history", {"A", "B"}), ("Input / output history", {"A", "B", "C"})):
|
||||
panel = panels[f"{title} · {suffix}"]
|
||||
assert set(queries(panel)) == refs
|
||||
for query in queries(panel).values():
|
||||
assert f'r["ups"] == "{device}"' in query
|
||||
assert 'aggregateWindow(every: v.windowPeriod, fn: mean, createEmpty: true)' in query
|
||||
assert 'float(v:' not in query # An empty interval stays null.
|
||||
|
||||
|
||||
@pytest.mark.parametrize("title,ref,measurement,divisor,expected", [
|
||||
# River 2 reports pack current and remaining capacity. River 3 reports
|
||||
# neither, so its panels use battery power and charge.
|
||||
("Desktop · Temperature history", "B", "ef_river2_bms_remain_cap", "1000.0", {
|
||||
"displayName": "Remaining Capacity", "unit": "amph",
|
||||
"custom.axisPlacement": "right", "custom.axisLabel": "Remaining capacity (Ah)"}),
|
||||
("NAS · Temperature history", "B", "bms_capacity_charge_ratio", None, {
|
||||
"displayName": "Charge", "unit": "percentunit",
|
||||
"custom.axisPlacement": "right", "custom.axisLabel": "Charge (%)"}),
|
||||
("Desktop · Input / output history", "C", "ef_river2_bms_amp", None, {
|
||||
"displayName": "Current", "unit": "amp",
|
||||
"custom.axisPlacement": "right", "custom.axisLabel": "Current (A)"}),
|
||||
("NAS · Input / output history", "C", "ef_river3_display_pow_get_bms", None, {
|
||||
"displayName": "Battery"}),
|
||||
])
|
||||
def test_ecoflow_histories_add_a_battery_series(panels, title, ref, measurement, divisor, expected):
|
||||
panel = panels[title]
|
||||
query = queries(panel)[ref]
|
||||
assert f'r["_measurement"] == "{measurement}"' in query
|
||||
assert ("r._value / " + divisor in query) if divisor else ("map(" not in query)
|
||||
properties = properties_for(panel, ref)
|
||||
assert expected.items() <= properties.items()
|
||||
# Battery power stays on the watt axis of the panel.
|
||||
if "unit" not in expected:
|
||||
assert "unit" not in properties and "custom.axisPlacement" not in properties
|
||||
|
||||
|
||||
@pytest.mark.parametrize("title,series", [
|
||||
# The River 2 decoder omits the estimate of the idle direction.
|
||||
("Desktop", {"A": ("ef_discharge_remaining_time_minutes", "To empty"),
|
||||
"B": ("ef_charge_remaining_time_minutes", "To full")}),
|
||||
# River 3 sends both estimates at all times. Use the one in use only.
|
||||
("NAS", {"A": ("ef_remaining_time_minutes", "Time remaining")}),
|
||||
])
|
||||
def test_time_remaining_shows_the_direction_in_use_in_hours(panels, title, series):
|
||||
panel = panels[f"{title} · Time remaining"]
|
||||
assert set(queries(panel)) == set(series)
|
||||
for ref, (measurement, name) in series.items():
|
||||
query = queries(panel)[ref]
|
||||
assert f'r["_measurement"] == "{measurement}"' in query
|
||||
assert 'range(start: -5m, stop: now())' in query
|
||||
assert query.index("last()") < query.index("float(v: r._value) / 60.0")
|
||||
assert properties_for(panel, ref) == {"displayName": name}
|
||||
defaults = panel["vizConfig"]["spec"]["fieldConfig"]["defaults"]
|
||||
# A fixed unit: Grafana's duration units round 76 hours to "3 days".
|
||||
assert defaults["unit"] == "suffix: h"
|
||||
assert defaults["decimals"] == 1
|
||||
assert defaults["noValue"] == "No estimate"
|
||||
assert not any("estimate" in name for name in panels if name != f"{title} · Time remaining")
|
||||
|
||||
|
||||
def test_only_the_time_remaining_panels_change_the_no_value_text(panels):
|
||||
for title, panel in panels.items():
|
||||
expected = "No estimate" if title.endswith("Time remaining") else "No recent data"
|
||||
assert panel["vizConfig"]["spec"]["fieldConfig"]["defaults"]["noValue"] == expected
|
||||
|
||||
|
||||
def test_latest_readings_still_use_recent_samples(generator):
|
||||
query = generator.flux("network", "bms_voltage_cells_volts", label="cell")
|
||||
assert 'range(start: -5m, stop: now())' in query
|
||||
assert 'sort(columns: ["_time"])' in query
|
||||
assert 'last()' in query
|
||||
assert 'aggregateWindow(' not in query
|
||||
|
||||
|
||||
def test_generator_writes_the_dashboard_file(generator, tmp_path, monkeypatch):
|
||||
output = tmp_path / "dashboard.json"
|
||||
monkeypatch.setattr("sys.argv", ["generate_grafana_ups.py", "--output", str(output)])
|
||||
generator.main()
|
||||
assert output.read_text(encoding="utf-8") == generator.render()
|
||||
|
||||
|
||||
def test_local_dashboard_file_is_current(generator):
|
||||
"""Git ignores the generated file, so a clone has none until the generator runs."""
|
||||
dashboard = ROOT / "grafana-ups.json"
|
||||
if not dashboard.exists():
|
||||
pytest.skip("grafana-ups.json is not generated")
|
||||
assert generator.render() == dashboard.read_text(encoding="utf-8")
|
||||
|
||||
|
||||
def test_api_wrapper_is_optional(generator):
|
||||
dashboard = json.loads(generator.render())
|
||||
assert "apiVersion" not in dashboard
|
||||
resource = json.loads(generator.render(api_resource=True))
|
||||
assert resource["spec"] == dashboard
|
||||
assert resource["apiVersion"] == "dashboard.grafana.app/v2"
|
||||
|
||||
|
||||
def test_top_row_has_charge_and_capacity_or_flow_without_reading_age(generator):
|
||||
dashboard = generator.build_dashboard()
|
||||
top = {}
|
||||
for item in dashboard["layout"]["spec"]["items"]:
|
||||
spec = item["spec"]
|
||||
if spec["y"] == 0:
|
||||
title = dashboard["elements"][spec["element"]["name"]]["spec"]["title"]
|
||||
top[title] = (spec["x"], spec["width"])
|
||||
expected = {}
|
||||
for col, (_, title, model) in enumerate(generator.DEVICES):
|
||||
expected[f"{title} · Charge ({model})"] = (col * 8, 5)
|
||||
secondary = "Capacity" if model == "JBD" else "Battery power flow"
|
||||
expected[f"{title} · {secondary}"] = (col * 8 + 5, 3)
|
||||
assert top == expected
|
||||
assert all("Last export" not in element["spec"]["title"]
|
||||
for element in dashboard["elements"].values())
|
||||
for element in dashboard["elements"].values():
|
||||
assert all("int(v: now())" not in query for query in queries(element["spec"]).values())
|
||||
|
||||
|
||||
def test_three_column_layout_stays_within_original_height(generator):
|
||||
dashboard = generator.build_dashboard()
|
||||
occupied = set()
|
||||
for item in dashboard["layout"]["spec"]["items"]:
|
||||
spec = item["spec"]
|
||||
assert spec["element"]["name"] in dashboard["elements"]
|
||||
assert 0 <= spec["x"] < spec["x"] + spec["width"] <= 24
|
||||
assert 0 <= spec["y"] < spec["y"] + spec["height"] <= 24
|
||||
cells = {(x, y) for x in range(spec["x"], spec["x"] + spec["width"])
|
||||
for y in range(spec["y"], spec["y"] + spec["height"])}
|
||||
assert not occupied.intersection(cells)
|
||||
occupied.update(cells)
|
||||
Reference in New Issue
Block a user