Show the present state of each device first in the Grafana dashboard
The dashboard started with gauges and histories of equal weight, so a power failure and its effect were hard to find. The first rows now show the charge, the power source, the time remaining, and the errors of each device. Below them, one band shows each period on battery power, and one chart shows the charge of the three devices. The charge history adds the pack current of the JBD and the River 2 on a right axis, as dashed lines in the color of each device. The charge limits apply to the charge lines only. A time-remaining panel with no estimate now names the state. The JBD panel shows "On mains" while the pack does not discharge. The River 2 panel shows "Idle" while the device reports 5999 minutes for the two directions. The River 2 errors panel names the module that reports a code and shows the code in hexadecimal. The solar controller reports 4096 with no solar input, and the panel counts that value as zero.
This commit is contained in:
+365
-77
@@ -45,39 +45,40 @@ def properties_for(panel, ref):
|
||||
}
|
||||
|
||||
|
||||
@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]
|
||||
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 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 '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"]
|
||||
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"]
|
||||
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"]
|
||||
|
||||
|
||||
@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]
|
||||
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
|
||||
@@ -88,7 +89,7 @@ def test_network_histories_do_not_invent_empty_intervals(panels, title):
|
||||
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"]
|
||||
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"
|
||||
@@ -96,44 +97,282 @@ def test_network_histories_do_not_invent_empty_intervals(panels, title):
|
||||
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, {
|
||||
@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 · Input / output history", "C", "ef_river3_display_pow_get_bms", None, {
|
||||
"displayName": "Battery"}),
|
||||
("nas", "NAS", "ef_river3_display_pow_get_bms", {"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)
|
||||
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()
|
||||
# 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
|
||||
|
||||
|
||||
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"),
|
||||
@@ -143,7 +382,8 @@ def test_ecoflow_histories_add_a_battery_series(panels, title, ref, measurement,
|
||||
])
|
||||
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)
|
||||
# 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
|
||||
@@ -154,13 +394,45 @@ def test_time_remaining_shows_the_direction_in_use_in_hours(panels, title, serie
|
||||
# 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")
|
||||
assert not any("estimate" in name for name in panels)
|
||||
|
||||
|
||||
def test_only_the_time_remaining_panels_change_the_no_value_text(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.endswith("Time remaining") else "No recent data"
|
||||
expected = "No estimate" if title == "NAS · Time remaining" else "No recent data"
|
||||
assert panel["vizConfig"]["spec"]["fieldConfig"]["defaults"]["noValue"] == expected
|
||||
|
||||
|
||||
@@ -195,35 +467,51 @@ def test_api_wrapper_is_optional(generator):
|
||||
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"])
|
||||
def test_each_device_column_starts_with_its_present_state(generator):
|
||||
position = positions(generator)
|
||||
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
|
||||
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_three_column_layout_stays_within_original_height(generator):
|
||||
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"] <= 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)}
|
||||
|
||||
Reference in New Issue
Block a user