diff --git a/bmspy/ups.py b/bmspy/ups.py index 4643259..e70be6b 100644 --- a/bmspy/ups.py +++ b/bmspy/ups.py @@ -312,49 +312,9 @@ def main() -> None: ) ) - if ( - charge_ratio <= args.critical_threshold - and comparison_1_charge_ratio <= args.critical_threshold - ): - if debug > 0: - print("Below critical threshold, shutting down") - handle_shutdown(action="shutdown", delay=args.shutdown_delay, debug=debug) - if critical_sent is False: - handle_email( - text="remaining capacity below {}%, shutting down".format( - args.critical_threshold - ), - level="Critical alert", - recipient=args.recipient, - mailserver=args.mailserver, - port=args.port, - mailuser=args.mailuser, - mailpass=args.mailpass, - debug=debug, - ) - critical_sent = True - - elif ( - charge_ratio <= args.warning_threshold - and comparison_1_charge_ratio <= args.warning_threshold - ): - if debug > 0: - print("Below warning threshold") - if warning_sent is False: - handle_email( - text="remaining capacity below {}%".format(args.warning_threshold), - level="Warning", - recipient=args.recipient, - mailserver=args.mailserver, - port=args.port, - mailuser=args.mailuser, - mailpass=args.mailpass, - debug=debug, - ) - warning_sent = True - + # A change of the power state is reported at every charge level. # Current needs to be negative for two consecutive reads - elif ( + if ( args.alert and current_amps < 0 and comparison_1_current_amps < 0 @@ -399,6 +359,55 @@ def main() -> None: warning_sent = False alert_sent = False + # The charge level matters only while the battery supplies the load, + # which needs two consecutive reads, as above. With mains power, a low + # battery charges again. A device that reports no current is judged by + # its charge alone. + has_current = _get_field_value(device_obj, "bms_current_amps") is not None + discharging = current_amps < 0 and comparison_1_current_amps < 0 + if has_current and not discharging: + pass + elif ( + charge_ratio <= args.critical_threshold + and comparison_1_charge_ratio <= args.critical_threshold + ): + if debug > 0: + print("Below critical threshold, shutting down") + handle_shutdown(action="shutdown", delay=args.shutdown_delay, debug=debug) + if critical_sent is False: + handle_email( + text="remaining capacity below {}%, shutting down".format( + args.critical_threshold + ), + level="Critical alert", + recipient=args.recipient, + mailserver=args.mailserver, + port=args.port, + mailuser=args.mailuser, + mailpass=args.mailpass, + debug=debug, + ) + critical_sent = True + + elif ( + charge_ratio <= args.warning_threshold + and comparison_1_charge_ratio <= args.warning_threshold + ): + if debug > 0: + print("Below warning threshold") + if warning_sent is False: + handle_email( + text="remaining capacity below {}%".format(args.warning_threshold), + level="Warning", + recipient=args.recipient, + mailserver=args.mailserver, + port=args.port, + mailuser=args.mailuser, + mailpass=args.mailpass, + debug=debug, + ) + warning_sent = True + time.sleep(DAEMON_UPDATE_PERIOD) diff --git a/tests/test_ups.py b/tests/test_ups.py index a6d638e..73acb6c 100644 --- a/tests/test_ups.py +++ b/tests/test_ups.py @@ -360,7 +360,7 @@ class TestUpsMain: data_seq = [] # Populate history with 5 readings all at 20% (below 30% critical) for _ in range(5): - data_seq.append(self._make_data(0.0, 0.20)) + data_seq.append(self._make_data(-1.0, 0.20)) mock_shutdown = MagicMock() mock_email = MagicMock() @@ -381,7 +381,7 @@ class TestUpsMain: data_seq = [] # 5 readings at 60% (below 75% warning, above 30% critical) for _ in range(5): - data_seq.append(self._make_data(0.0, 0.60)) + data_seq.append(self._make_data(-1.0, 0.60)) mock_email = MagicMock() with patch("sys.argv", ["bmspy-ups", "--warning", "75", "--critical", "30"]): @@ -397,7 +397,7 @@ class TestUpsMain: def test_main_mail_port_option_is_a_number(self): """A text port never equals 587, so the STARTTLS check would not match.""" ups_mod.warning_sent = False - data_seq = [self._make_data(0.0, 0.60)] * 5 + data_seq = [self._make_data(-1.0, 0.60)] * 5 mock_smtp = MagicMock() server = mock_smtp.return_value.__enter__.return_value with patch("sys.argv", ["bmspy-ups", "--port", "587"]): @@ -514,6 +514,46 @@ class TestUpsMain: (14, "shutdown"), (14, "Critical alert"), (15, "shutdown"), ] + OUTAGE = [(1.0, 0.95)] * 8 + [(-5.0, 0.90)] * 4 # mains power fails at reading 9 + + def test_power_return_below_the_warning_level_is_reported(self): + """The charge is 60% when mains power returns at reading 17.""" + readings = self.OUTAGE + [(-5.0, 0.60)] * 4 + [(3.0, 0.60)] * 8 + [(3.0, 0.80)] * 4 + assert self._events(readings) == [ + (10, "Power loss alert"), (14, "Warning"), (18, "cancel"), (18, "Recovery alert"), + ] + # The next power failure can send its emails again. + assert (ups_mod.alert_sent, ups_mod.warning_sent, ups_mod.critical_sent) == (False, False, False) + + def test_power_return_below_the_critical_level_cancels_the_shutdown(self): + """The charge is 25% when mains power returns at reading 17.""" + readings = self.OUTAGE + [(-5.0, 0.25)] * 4 + [(3.0, 0.25)] * 8 + events = self._events(readings) + assert events[:3] == [(10, "Power loss alert"), (14, "shutdown"), (14, "Critical alert")] + # The shutdown is canceled, and the low charge on mains power does not start it again. + assert events[-2:] == [(18, "cancel"), (18, "Recovery alert")] + assert [number for number, event in events if event == "shutdown"] == [14, 15, 16] + + def test_second_power_failure_is_reported_again(self): + readings = self.OUTAGE + [(-5.0, 0.60)] * 4 + [(3.0, 0.60)] * 6 + readings = readings + [(-5.0, 0.60)] * 4 + assert [event for _, event in self._events(readings)] == [ + "Power loss alert", "Warning", "cancel", "Recovery alert", "Power loss alert", "Warning", + ] + + def test_one_discharging_reading_on_mains_power_does_not_shut_down(self): + """One reading can be wrong. A shutdown needs two discharging readings.""" + events = self._events([(3.0, 0.20)] * 5 + [(-0.1, 0.20)] + [(3.0, 0.20)] * 4) + assert [event for _, event in events if event in ("shutdown", "Critical alert", "Warning")] == [] + + def test_low_battery_on_mains_power_does_not_shut_down(self): + """At a restart after a long power failure, the battery is low and it charges.""" + assert self._events([(3.0, 0.20)] * 8) == [] + + def test_device_without_current_is_judged_by_its_charge(self): + """The River models report the charge but no battery current.""" + events = self._events([(None, 0.95)] * 5 + [(None, 0.20)] * 2) + assert events == [(7, "shutdown"), (7, "Critical alert")] def test_main_not_enough_history_debug_2(self, capsys): """With debug=2 and < 4 readings, prints 'not enough readings'.""" @@ -532,7 +572,7 @@ class TestUpsMain: def test_main_below_warning_debug_1(self, capsys): """With debug=1, warning message is printed.""" ups_mod.warning_sent = False - data_seq = [self._make_data(0.0, 0.60)] * 5 + data_seq = [self._make_data(-1.0, 0.60)] * 5 with patch("sys.argv", ["bmspy-ups", "-v", "--warning", "75", "--critical", "30"]): with pytest.raises(StopIteration): with patch("bmspy.ups.client.read_data", side_effect=[data_seq[0]] + data_seq + [StopIteration("done")]), \ @@ -603,7 +643,7 @@ class TestUpsMain: def test_main_debug_1_prints_thresholds(self, capsys): """With debug=1 and below threshold, threshold messages are printed.""" ups_mod.critical_sent = False - data_seq = [self._make_data(0.0, 0.20)] * 5 + data_seq = [self._make_data(-1.0, 0.20)] * 5 with patch("sys.argv", ["bmspy-ups", "-v", "--critical", "30"]): with pytest.raises(StopIteration): with patch("bmspy.ups.client.read_data", side_effect=[data_seq[0]] + data_seq + [StopIteration("done")]), \