"""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"] } def defaults_for(panel): return panel["vizConfig"]["spec"]["fieldConfig"]["defaults"] def positions(generator): dashboard = generator.build_dashboard() return { dashboard["elements"][item["spec"]["element"]["name"]]["spec"]["title"]: tuple(item["spec"][key] for key in ("x", "y", "width", "height")) for item in dashboard["layout"]["spec"]["items"] } def test_network_cell_history_keeps_original_series_and_separate_axes(panels): panel = panels["Network · Cell voltage history"] query = queries(panel) assert set(query) == {"A", "B"} assert 'r["_measurement"] == "bms_voltage_cells_volts"' in query["A"] assert 'r["_measurement"] == "bms_current_amps"' in query["B"] # Cell identities survive grouping and column selection. keys = ["ups", "_measurement", "_field", "cell"] assert f'group(columns: {json.dumps(keys)})' in query["A"] assert f'keep(columns: {json.dumps(["_time", "_value"] + keys)})' in query["A"] assert defaults_for(panel)["unit"] == "volt" assert "${__field.labels.cell}" in defaults_for(panel)["displayName"] current = properties_for(panel, "B") assert current["unit"] == "amp" assert current["displayName"] == "Current" assert current["custom.axisPlacement"] == "right" assert current["custom.axisLabel"] def test_network_cell_history_does_not_invent_empty_intervals(panels): panel = panels["Network · Cell voltage history"] 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 = defaults_for(panel)["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,measurement,expected", [ # River 2 reports pack current. River 3 reports none, so its panel uses # battery power, which stays on the watt axis. ("desktop", "Desktop", "ef_river2_bms_amp", { "displayName": "Current", "unit": "amp", "custom.axisPlacement": "right", "custom.axisLabel": "Current (A)"}), ("nas", "NAS", "ef_river3_display_pow_get_bms", {"displayName": "Battery"}), ]) def test_ecoflow_power_histories_add_a_battery_series(panels, device, title, measurement, expected): panel = panels[f"{title} · Input / output history"] assert set(queries(panel)) == {"A", "B", "C"} 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 'map(' not in query # An empty interval stays null. assert f'r["_measurement"] == "{measurement}"' in queries(panel)["C"] properties = properties_for(panel, "C") assert expected.items() <= properties.items() if "unit" not in expected: assert "unit" not in properties and "custom.axisPlacement" not in properties def test_charge_history_adds_pack_current_on_its_own_axis(panels): panel = panels["Charge history"] expected = { "A": ("network", "bms_capacity_charge_ratio", "Network", "green"), "B": ("desktop", "bms_capacity_charge_ratio", "Desktop", "yellow"), "C": ("nas", "bms_capacity_charge_ratio", "NAS", "blue"), # River 3 reports no battery current, so NAS has no current series. "D": ("network", "bms_current_amps", "Network current", "green"), "E": ("desktop", "ef_river2_bms_amp", "Desktop current", "yellow"), } assert set(queries(panel)) == set(expected) for ref, (device, measurement, name, color) in expected.items(): query = queries(panel)[ref] assert f'r["ups"] == "{device}"' in query assert f'r["_measurement"] == "{measurement}"' in query assert 'aggregateWindow(every: v.windowPeriod, fn: mean, createEmpty: true)' in query assert 'map(' not in query # Do not change the current's sign. properties = properties_for(panel, ref) assert properties["displayName"] == name # A device has one color for its charge and its current. assert properties["color"] == {"mode": "fixed", "fixedColor": color} if ref in "ABC": assert properties["max"] == 1 and properties["custom.axisSoftMin"] == 0.5 assert "unit" not in properties and "custom.axisPlacement" not in properties else: assert properties["unit"] == "amp" assert properties["custom.axisPlacement"] == "right" assert properties["custom.axisLabel"] == "Current (A)" assert properties["custom.lineStyle"]["fill"] == "dash" # The charge limits must not apply to the current axis. assert "max" not in properties and "custom.axisSoftMin" not in properties defaults = defaults_for(panel) assert defaults["unit"] == "percentunit" assert "max" not in defaults and "axisSoftMin" not in defaults["custom"] @pytest.mark.parametrize("title,queries_by_name", [ # The JBD reports battery current only, so it has no output power. ("Output power", {"Desktop": ("desktop", "ef_output_power_watts"), "NAS": ("nas", "ef_output_power_watts")}), ("Temperature", {"Network sensor ${__field.labels.sensor}": ("network", "bms_temperature_celsius"), "Desktop": ("desktop", "bms_temperature_celsius"), "NAS": ("nas", "bms_temperature_celsius")}), ]) def test_shared_histories_have_one_query_for_each_device(panels, title, queries_by_name): panel = panels[title] assert len(queries(panel)) == len(queries_by_name) for (ref, query), (name, (device, measurement)) in zip(queries(panel).items(), queries_by_name.items()): assert f'r["ups"] == "{device}"' in query assert f'r["_measurement"] == "{measurement}"' in query assert 'aggregateWindow(every: v.windowPeriod, fn: mean, createEmpty: true)' in query assert properties_for(panel, ref) == {"displayName": name} # Temperature sensors stay separate series. if title == "Temperature": assert all('"sensor"])' in query for query in queries(panel).values()) @pytest.mark.parametrize("title,measurement,rule", [ # A JBD pack discharges on battery power. A River has no input power then. ("Network", '== "bms_current_amps"', "if r._value < 0.0 then 0.0 else 1.0"), # An idle River 2 draws no input power on mains, so the AC input voltage counts too. ("Desktop", "=~ /^(ef_input_power_watts|ef_river2_inv_inv_in_vol)$/", "if r._value > 0.0 then 1.0 else 0.0"), ("NAS", '== "ef_input_power_watts"', "if r._value > 0.0 then 1.0 else 0.0"), ]) def test_power_source_uses_the_same_rule_now_and_over_time(generator, panels, title, measurement, rule): mappings = [{"type": "value", "options": { "0": {"text": "On battery", "color": "red"}, "1": {"text": "Mains", "color": "green"}}}] now = panels[f"{title} · Source"] (query,) = queries(now).values() assert f'r["_measurement"] {measurement}' in query assert query.index("last()") < query.index(f"_value: {rule}") assert defaults_for(now)["mappings"] == mappings timeline = panels["Power source"] assert timeline["vizConfig"]["group"] == "state-timeline" ref = "ABC"[[name for _, name, _ in generator.DEVICES].index(title)] query = queries(timeline)[ref] assert f'r["_measurement"] {measurement}' in query # The lowest reading keeps a short failure visible. The rule cannot test # a null reading, so empty intervals are left out, and the panel ends a # state after five minutes with no reading. assert 'aggregateWindow(every: v.windowPeriod, fn: min, createEmpty: false)' in query assert query.index("aggregateWindow(") < query.index(f"_value: {rule}") assert properties_for(timeline, ref) == {"displayName": title} assert defaults_for(timeline)["mappings"] == mappings assert defaults_for(timeline)["custom"]["insertNulls"] == 300000 def test_river_source_takes_the_highest_of_its_measurements(panels): """Mains if one of the measurements is above zero, at each time.""" (now,) = queries(panels["Desktop · Source"]).values() steps = [now.index(step) for step in ( "last()", "toFloat()", 'group(columns: ["ups"])', "max()", "_value: if r._value > 0.0")] assert steps == sorted(steps) history = queries(panels["Power source"])["B"] steps = [history.index(step) for step in ( "aggregateWindow(", "toFloat()", 'group(columns: ["ups", "_time"])', "max()", 'group(columns: ["ups"])', 'sort(columns: ["_time"])', "_value: if r._value > 0.0")] assert steps == sorted(steps) # A second text column would show as a second band in the timeline. assert 'keep(columns: ["_time", "_value", "ups"])' in history @pytest.mark.parametrize("title,measurement", [ ("Desktop · Charge (River 2)", "ef_river2_ems_max_charge_soc"), ("NAS · Charge (River 3)", "ef_river3_display_cms_max_chg_soc"), ]) def test_ecoflow_charge_shows_the_charge_limit(panels, title, measurement): panel = panels[title] assert set(queries(panel)) == {"A", "B"} assert f'r["_measurement"] == "{measurement}"' in queries(panel)["B"] # The limit is a percentage. The charge is a ratio. assert "float(v: r._value) / 100.0" in queries(panel)["B"] assert properties_for(panel, "A") == {"displayName": "Charge"} assert properties_for(panel, "B")["displayName"] == "Limit" assert set(queries(panels["Network · Charge (JBD)"])) == {"A"} @pytest.mark.parametrize("title,ref,pattern", [ ("Desktop · Errors", "F", "/^ef_river2_((bms|inv)_err_code|mppt_fault_code|bms_bms_fault)$/"), ("NAS · Errors", "A", "/^ef_river3_display_[a-z0-9_]*err_?code$/"), ]) def test_ecoflow_errors_show_the_highest_numeric_code(panels, title, ref, pattern): panel = panels[title] query = queries(panel)[ref] assert f'r["_measurement"] =~ {pattern}' in query # One type before the series join into one table. steps = [query.index(step) for step in ("last()", "toFloat()", 'group(columns: ["ups"])', "max()")] assert steps == sorted(steps) (mapping,) = defaults_for(panel)["mappings"] assert {value: entry["text"] for value, entry in mapping["options"].items()} == {"0": "OK"} # No data must not read as OK. assert defaults_for(panel)["noValue"] == "No recent data" def test_river2_device_error_code_is_text(panels): """bmspy exports the four bytes as "0x00000000", which is not a number.""" panel = panels["Desktop · Errors"] device = queries(panel)["A"] assert 'r["_measurement"] == "ef_river2_pd_err_code"' in device assert '_value: if r._value == "0x00000000" then 0.0 else 1.0' in device assert "toFloat()" not in device assert all("pd_err_code" not in queries(panel)[ref] for ref in "BCDEF") assert properties_for(panel, "A")["displayName"] == "Device" (mapping,) = properties_for(panel, "A")["mappings"] assert {value: entry["text"] for value, entry in mapping["options"].items()} == {"0": "OK", "1": "Error"} # The solar controller reports 4096 with no solar input. That is not an error. NORMAL_SOLAR_CODE = '_value: if r._measurement == "ef_river2_mppt_fault_code" and r._value == 4096.0 then 0.0 else r._value' def test_river2_errors_name_the_module_that_reports_a_code(panels): panel = panels["Desktop · Errors"] modules = { "B": ("ef_river2_bms_err_code", "Battery"), "C": ("ef_river2_bms_bms_fault", "Battery fault"), "D": ("ef_river2_inv_err_code", "Inverter"), "E": ("ef_river2_mppt_fault_code", "Solar"), } assert set(queries(panel)) == {"A", "F"} | set(modules) for ref, (measurement, name) in modules.items(): query = queries(panel)[ref] assert f'r["_measurement"] == "{measurement}"' in query # A code of zero gives no series, so only a module with a code shows. steps = [query.index(step) for step in ("last()", "toFloat()", NORMAL_SOLAR_CODE, "filter(fn: (r) => r._value != 0.0)")] assert steps == sorted(steps) properties = properties_for(panel, ref) assert properties["displayName"] == name assert properties["unit"] == "hex0x" # The meaning of a code is not known, so its color is not the fault color. assert properties["color"] == {"mode": "fixed", "fixedColor": "orange"} # "Modules OK" needs a reading of every code, and every code at zero. clear = queries(panel)["F"] steps = [clear.index(step) for step in ("toFloat()", NORMAL_SOLAR_CODE, "max()", "filter(fn: (r) => r._value == 0.0)")] assert steps == sorted(steps) assert properties_for(panel, "F") == {"displayName": "Modules"} assert panel["vizConfig"]["spec"]["options"]["textMode"] == "value_and_name" def test_network_protection_names_the_flags_that_are_set(panels): panel = panels["Network · Protection"] active, clear = queries(panel)["A"], queries(panel)["B"] for query in (active, clear): assert 'r["_measurement"] =~ /^bms_protection_[a-z]+_bool$/' in query assert "if r._value then 1.0 else 0.0" in query # One series for each set flag, named by the description of the flag. assert '"help"])' in active assert active.index("last()") < active.index("filter(fn: (r) => r._value == 1.0)") assert defaults_for(panel)["displayName"] == "${__field.labels.help}" assert panel["vizConfig"]["spec"]["options"]["textMode"] == "name" # "None" needs a reading of every flag. No data must not read as None. steps = [clear.index(step) for step in ( "last()", 'group(columns: ["ups"])', "max()", "filter(fn: (r) => r._value == 0.0)")] assert steps == sorted(steps) assert properties_for(panel, "B") == {"displayName": "None"} assert defaults_for(panel)["noValue"] == "No recent data" @pytest.mark.parametrize("pattern,matches,rejects", [ ("/^bms_protection_[a-z]+_bool$/", ["bms_protection_sop_bool", "bms_protection_slmos_bool"], ["bms_charge_is_charging"]), ("/^ef_river2_((bms|inv)_err_code|mppt_fault_code|bms_bms_fault)$/", ["ef_river2_bms_err_code", "ef_river2_inv_err_code", "ef_river2_mppt_fault_code", "ef_river2_bms_bms_fault"], # The device code is text ("0x00000000"), and has a query of its own. ["ef_river2_pd_err_code", "ef_river2_pd_decode_error", "ef_river2_ems_bms_warning_state"]), ("/^ef_river3_display_[a-z0-9_]*err_?code$/", ["ef_river3_display_errcode", "ef_river3_display_bms_err_code", "ef_river3_display_plug_in_info_4p8_1_err_code"], # A nested list and a text field. Neither is a number. ["ef_river3_display_err_code_record_list", "ef_river3_display_decode_error"]), ]) def test_fault_patterns_select_numeric_fields_only(pattern, matches, rejects): import re expression = re.compile(pattern.strip("/")) assert all(expression.search(name) for name in matches) assert not any(expression.search(name) for name in rejects) @pytest.mark.parametrize("title,expected", [ ("Network", {"Cycles": ("bms_charge_cycles", None), "Cell spread": ("bms_voltage_cells_volts", "volt"), "Remaining": ("bms_capacity_remaining_ah", "amph"), "Nominal": ("bms_capacity_nominal_ah", "amph")}), ("Desktop", {"Cycles": ("ef_river2_bms_cycles", None), "Health": ("ef_river2_bms_soh", "percent"), "Full": ("ef_river2_bms_full_cap", "amph"), "Design": ("ef_river2_bms_design_cap", "amph")}), ("NAS", {"Health": ("ef_river3_display_cms_batt_soh", "percent"), "Design": ("ef_river3_display_bms_design_cap", "amph")}), ]) def test_battery_panels_show_the_slow_values_of_each_model(panels, title, expected): panel = panels[f"{title} · Battery"] assert len(queries(panel)) == len(expected) for (ref, query), (name, (measurement, unit)) in zip(queries(panel).items(), expected.items()): assert f'r["_measurement"] == "{measurement}"' in query properties = properties_for(panel, ref) assert properties["displayName"] == name assert properties.get("unit") == unit # The River models report mAh. assert ("float(v: r._value) / 1000.0" in query) is (unit == "amph" and title != "Network") if title == "Network": spread = queries(panel)["B"] assert spread.index("last()") < spread.index('group(columns: ["ups"])') < spread.index("spread()") def test_network_time_remaining_is_calculated_during_discharge_only(panels): panel = panels["Network · Time remaining"] query = queries(panel)["A"] assert 'r["_measurement"] =~ /^bms_(capacity_remaining_ah|current_amps)$/' in query steps = [query.index(step) for step in ( "last()", 'pivot(rowKey: ["_field"], columnKey: ["_measurement"], valueColumn: "_value")', 'filter(fn: (r) => r["bms_current_amps"] < 0.0)', '_value: r["bms_capacity_remaining_ah"] / (0.0 - r["bms_current_amps"])', )] assert steps == sorted(steps) assert defaults_for(panel)["unit"] == "suffix: h" assert panel["description"].startswith("Calculated, not reported") @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"] # River 2 has one more query, which names the state with no estimate. assert set(queries(panel)) == set(series) | ({"C"} if title == "Desktop" else set()) 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 not any("estimate" in name for name in panels) def state_text(panel, ref): """Give the name of a state query, and the text for each of its values.""" properties = properties_for(panel, ref) (mapping,) = properties["mappings"] assert properties["color"] == {"mode": "fixed", "fixedColor": "text"} return properties["displayName"], {value: entry["text"] for value, entry in mapping["options"].items()} def test_network_time_remaining_names_the_state_on_mains(panels): panel = panels["Network · Time remaining"] assert set(queries(panel)) == {"A", "B"} query = queries(panel)["B"] assert 'r["_measurement"] == "bms_current_amps"' in query # The opposite of the discharge rule, so one of the two queries has a value. steps = [query.index(step) for step in ("last()", "filter(fn: (r) => r._value >= 0.0)", "_value: 0.0")] assert steps == sorted(steps) assert state_text(panel, "B") == ("State", {"0": "On mains"}) assert panel["vizConfig"]["spec"]["options"]["textMode"] == "value" def test_river2_time_remaining_names_the_state_with_no_estimate(panels): panel = panels["Desktop · Time remaining"] query = queries(panel)["C"] # The raw reports keep the values that the decoder omits from A and B. assert 'r["_measurement"] =~ /^ef_river2_ems_(chg|dsg)_remain_time$/' in query steps = [query.index(step) for step in ( "last()", "toFloat()", "if r._value == 5999.0 or r._value == 4294967295.0 then 1.0 else 0.0", 'group(columns: ["ups"])', "min()", "filter(fn: (r) => r._value == 1.0)")] assert steps == sorted(steps) assert state_text(panel, "C") == ("Battery", {"1": "Idle"}) def test_only_the_river3_time_remaining_changes_the_no_value_text(panels): """The other two panels name the state with a query, so no data stays no data.""" for title, panel in panels.items(): expected = "No estimate" if title == "NAS · 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_each_device_column_starts_with_its_present_state(generator): position = positions(generator) expected = {} for col, (_, title, model) in enumerate(generator.DEVICES): x = col * 8 first, second = ("Pack voltage", "Pack current") if model == "JBD" else ("Input", "Output") expected.update({ f"{title} · Charge ({model})": (x, 0, 5, 4), f"{title} · Source": (x + 5, 0, 3, 4), f"{title} · {first}": (x, 4, 4, 4), f"{title} · {second}": (x + 4, 4, 4, 4), # The names of the protection flags need more width than an error code. f"{title} · Time remaining": (x, 8, 4 if model == "JBD" else 5, 4), f"{title} · {'Protection' if model == 'JBD' else 'Errors'}": (x + 4, 8, 4, 4) if model == "JBD" else (x + 5, 8, 3, 4), }) assert {title: place for title, place in position.items() if place[1] < 12} == expected def test_failures_and_charge_follow_the_present_state_at_full_width(generator): position = positions(generator) assert position["Power source"] == (0, 12, 24, 4) assert position["Charge history"] == (0, 16, 24, 7) # Everything else is detail, below these two. assert all(y >= 23 for title, (_, y, _, _) in position.items() if y >= 12 and title not in ("Power source", "Charge history")) def test_no_panel_shows_the_age_of_a_reading(generator): dashboard = generator.build_dashboard() 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_layout_has_no_overlap_and_no_hole(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"] 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) height = max(y for _, y in occupied) + 1 assert occupied == {(x, y) for x in range(24) for y in range(height)}