From 953f8a6482cceffce655772324448c22b9b4684f Mon Sep 17 00:00:00 2001 From: Timothy Allen Date: Sat, 10 Oct 2026 09:03:34 +0200 Subject: [PATCH] Read each device in a background thread A failed JBD read retried forever inside a client request and blocked every other device and client. The server now starts one polling thread for each device, and a request returns the last snapshot. Only the first request after startup waits, until each device completes one read attempt. One lock makes sure that only one Bluetooth collection runs at a time. A serial BMS retries a failed read after 1 second and keeps its last snapshot for 30 seconds. An error in a collection does not end the polling of that device. --- README.md | 8 ++- bmspy/server.py | 84 ++++++++++++++++++++++----- tests/test_server.py | 19 +++++- tests/test_server_ecoflow.py | 109 ++++++++++++++++++++++++++--------- 4 files changed, 175 insertions(+), 45 deletions(-) diff --git a/README.md b/README.md index 0bd30af..1a3e675 100644 --- a/README.md +++ b/README.md @@ -94,8 +94,12 @@ Omit `--ups` for all devices. Add `--ups river2` to select one. `--ups all` mean device named `all`, not all devices. To include a serial BMS, add a server argument such as `-d network:/dev/ttyUSB0`. -The server caches Bluetooth data for 60 seconds. Failed reads wait 60 seconds -before retrying. Unavailable devices are omitted from the response. +The server reads each device in its own background thread, so a slow or offline +device does not delay the other devices or the clients. It reads one Bluetooth +device at a time and waits 60 seconds after each Bluetooth read, including a +failed one. It reads a serial BMS every 5 seconds and retries a failed serial +read after 1 second. Unavailable devices are omitted from the response. Only the +first request after startup waits, until each device completes one read attempt. #### Diagnostics diff --git a/bmspy/server.py b/bmspy/server.py index d676ef7..114772d 100755 --- a/bmspy/server.py +++ b/bmspy/server.py @@ -13,8 +13,9 @@ import signal import json import re import struct +import threading import serial -from dataclasses import asdict as dataclass_asdict, dataclass +from dataclasses import asdict as dataclass_asdict, dataclass, field from types import FrameType from typing import Any, NoReturn @@ -34,6 +35,18 @@ class DeviceState: data: UPS | None = None timestamp: float = 0.0 # Last collection attempt ecoflow: EcoFlowBTConfig | None = None # EcoFlow BT config + # Set after the first collection attempt, successful or not. + polled: threading.Event = field(default_factory=threading.Event, repr=False, compare=False) + + +SERIAL_POLL_SECONDS = 5 +SERIAL_RETRY_SECONDS = 1 +SERIAL_STALE_SECONDS = 30 +BLUETOOTH_POLL_SECONDS = 60 +FIRST_POLL_TIMEOUT = 120 + +# One adapter: collect from one Bluetooth device at a time. +bluetooth_lock = threading.Lock() # Expected kernel log output when the USB-serial adapter is plugged in: # usb 1-1.4: new full-speed USB device number 4 using xhci_hcd @@ -172,23 +185,57 @@ def collect_device_data(device: DeviceState, debug: int = 0) -> UPS | None: if device.ser is not None: return collect_data(device.ser, debug) or None if device.ecoflow is not None: - return ecoflow_collect_data(device.ecoflow, debug) or None + with bluetooth_lock: + return ecoflow_collect_data(device.ecoflow, debug) or None return None -def get_device_data(device: DeviceState, debug: int = 0) -> UPS | None: - """Return cached data, refreshing it when the backend's interval expires.""" - min_age = 60 if device.ecoflow is not None else 5 - if device.timestamp <= time.time() - min_age: - device.data = collect_device_data(device, debug) - if device.ecoflow is None: - # Preserve serial retries. BLE already retries within collect_data; - # retrying it forever here would block all clients when it is offline. - while device.data is None: - time.sleep(1) - device.data = collect_device_data(device, debug) +def poll_device(device: DeviceState, stop: threading.Event, debug: int = 0) -> None: + """Keep *device.data* current until *stop* is set. + + Each device has its own thread, so a slow or offline device delays neither + the clients nor the other devices. A collection replaces the snapshot and + never changes one that clients can already read, so no lock is needed. + """ + last_success = 0.0 + while not stop.is_set(): + try: + data = collect_device_data(device, debug) + except Exception as e: + # A failed read must not end polling for this device. + debugger("server: collection error: {}".format(e)) + data = None device.timestamp = time.time() - return device.data + if data is not None: + device.data = data + last_success = device.timestamp + delay = BLUETOOTH_POLL_SECONDS if device.ecoflow is not None else SERIAL_POLL_SECONDS + elif device.ecoflow is not None: + # BLE already retries within collect_data. Do not serve stale data. + device.data = None + delay = BLUETOOTH_POLL_SECONDS + else: + # Serial reads fail briefly at times. Retry soon, and keep the last + # snapshot for a short time so that one failed read stays invisible. + if device.timestamp - last_success > SERIAL_STALE_SECONDS: + device.data = None + delay = SERIAL_RETRY_SECONDS + device.polled.set() + stop.wait(delay) + + +def start_polling( + ups_devices: dict[str, DeviceState], stop: threading.Event, debug: int = 0, +) -> list[threading.Thread]: + """Start one polling thread per device.""" + threads = [] + for name, device in ups_devices.items(): + thread = threading.Thread( + target=poll_device, args=(device, stop, debug), name="poll-" + name, daemon=True, + ) + thread.start() + threads.append(thread) + return threads def main(): @@ -342,6 +389,10 @@ def main(): sock.listen(1) + stop_polling = threading.Event() + atexit.register(stop_polling.set) + start_polling(ups_devices, stop_polling, debug) + while True: connection = None client_address = None @@ -398,7 +449,10 @@ def main(): name, device.timestamp, time.time() ) ) - data = get_device_data(device, debug) + # Wait for the first collection attempt only, so that a + # client at startup sees every device that can answer. + device.polled.wait(FIRST_POLL_TIMEOUT) + data = device.data if data is not None: result[name] = data diff --git a/tests/test_server.py b/tests/test_server.py index 1f0d009..ad893f9 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -11,7 +11,17 @@ import serial from bmspy.classes import BMSScalarField, BMSInfoField, UPS from bmspy.jbd_bms import JBDBMS -from bmspy.server import DeviceState, parse_device, read_request, send_response +from bmspy.server import ( + DeviceState, collect_device_data, parse_device, read_request, send_response, +) + + +def _poll_once(devices, stop, debug=0): + """Collect once in the server thread. main() stops its polling threads only + at process exit, so real ones would outlive a test and call later mocks.""" + for device in devices.values(): + device.data = collect_device_data(device, debug) + device.polled.set() # --------------------------------------------------------------------------- @@ -340,6 +350,7 @@ class TestServerMain: patch("bmspy.server.signal.signal"), \ patch("bmspy.server.initialise_serial", return_value=MagicMock()), \ patch("bmspy.server.collect_data", return_value=fake_bms), \ + patch("bmspy.server.start_polling", _poll_once), \ patch("bmspy.server.time.sleep"), \ patch("bmspy.server.os.path.isdir", return_value=True), \ patch("bmspy.server.os.path.exists", return_value=False), \ @@ -525,6 +536,7 @@ class TestServerMain: patch("bmspy.server.signal.signal"), \ patch("bmspy.server.initialise_serial", return_value=MagicMock()), \ patch("bmspy.server.collect_data", return_value=fake_bms), \ + patch("bmspy.server.start_polling", _poll_once), \ patch("bmspy.server.time.sleep"), \ patch("bmspy.server.os.path.isdir", return_value=True), \ patch("bmspy.server.os.path.exists", return_value=False), \ @@ -576,6 +588,7 @@ class TestServerMain: patch("bmspy.server.signal.signal"), \ patch("bmspy.server.initialise_serial", return_value=MagicMock()), \ patch("bmspy.server.collect_data", return_value=fake_bms), \ + patch("bmspy.server.start_polling", _poll_once), \ patch("bmspy.server.time.sleep"), \ patch("bmspy.server.os.path.exists", return_value=False), \ patch("bmspy.server.os.path.isdir", return_value=False), \ @@ -686,6 +699,7 @@ class TestServerMain: patch("bmspy.server.signal.signal"), \ patch("bmspy.server.initialise_serial", return_value=MagicMock()), \ patch("bmspy.server.collect_data", return_value=fake_bms), \ + patch("bmspy.server.start_polling", _poll_once), \ patch("bmspy.server.time.sleep"), \ patch("bmspy.server.os.path.isdir", return_value=True), \ patch("bmspy.server.os.path.exists", return_value=False), \ @@ -777,6 +791,7 @@ class TestServerMain: patch("bmspy.server.signal.signal"), \ patch("bmspy.server.initialise_serial", return_value=MagicMock()), \ patch("bmspy.server.collect_data", return_value=fake_bms), \ + patch("bmspy.server.start_polling", _poll_once), \ patch("bmspy.server.time.sleep"), \ patch("bmspy.server.os.path.isdir", return_value=False), \ patch("bmspy.server.os.makedirs"), \ @@ -838,6 +853,7 @@ class TestServerMain: patch("bmspy.server.signal.signal"), \ patch("bmspy.server.initialise_serial", return_value=MagicMock()), \ patch("bmspy.server.collect_data", return_value=JBDBMS()), \ + patch("bmspy.server.start_polling", _poll_once), \ patch("bmspy.server.time.sleep"), \ patch("bmspy.server.os.path.isdir", return_value=True), \ patch("bmspy.server.os.path.exists", return_value=False), \ @@ -901,6 +917,7 @@ class TestServerMain: patch("bmspy.server.signal.signal"), \ patch("bmspy.server.initialise_serial", return_value=MagicMock()), \ patch("bmspy.server.collect_data", return_value=fake_bms), \ + patch("bmspy.server.start_polling", _poll_once), \ patch("bmspy.server.time.sleep"), \ patch("bmspy.server.os.path.isdir", return_value=True), \ patch("bmspy.server.os.path.exists", return_value=False), \ diff --git a/tests/test_server_ecoflow.py b/tests/test_server_ecoflow.py index f2ab9bd..00d1675 100644 --- a/tests/test_server_ecoflow.py +++ b/tests/test_server_ecoflow.py @@ -1,5 +1,6 @@ """Server routing and caching tests with synthetic device identifiers.""" +import threading from unittest.mock import Mock import pytest @@ -74,46 +75,100 @@ def test_collection_dispatches_to_the_selected_backend(monkeypatch): assert server.collect_device_data(server.DeviceState()) is None +class StopAfter: + """Stand-in for the stop event: record each wait and stop after *loops* polls.""" + + def __init__(self, loops, observe=lambda: None): + self.loops, self.observe = loops, observe + self.waits, self.seen = [], [] + + def is_set(self): + return len(self.waits) >= self.loops + + def wait(self, delay): + self.waits.append(delay) + self.seen.append(self.observe()) + + @pytest.mark.parametrize("bluetooth,interval", [(False, 5), (True, 60)]) -def test_cache_interval_depends_on_backend(monkeypatch, bluetooth, interval): - old, fresh = JBDBMS(), JBDBMS() +def test_poll_interval_depends_on_backend(monkeypatch, bluetooth, interval): + fresh = JBDBMS() config = initialise_device(ADDRESS, SERIAL, "test-user") if bluetooth else None - state = server.DeviceState(data=old, timestamp=100, ecoflow=config) + state = server.DeviceState(ser=None if bluetooth else Mock(), ecoflow=config) read = Mock(return_value=fresh) monkeypatch.setattr(server, "collect_device_data", read) - monkeypatch.setattr(server.time, "time", lambda: 100 + interval - 1) - assert server.get_device_data(state) is old - read.assert_not_called() - monkeypatch.setattr(server.time, "time", lambda: 100 + interval) - assert server.get_device_data(state) is fresh - read.assert_called_once_with(state, 0) + monkeypatch.setattr(server.time, "time", lambda: 100) + stop = StopAfter(1) + server.poll_device(state, stop, 2) + assert state.data is fresh + assert state.timestamp == 100 + assert state.polled.is_set() + assert stop.waits == [interval] + read.assert_called_once_with(state, 2) -def test_offline_bluetooth_does_not_retry_forever_or_serve_stale_data(monkeypatch): +def test_offline_bluetooth_does_not_retry_at_once_or_serve_stale_data(monkeypatch): state = server.DeviceState( data=JBDBMS(), timestamp=1, ecoflow=initialise_device(ADDRESS, SERIAL, "test-user"), ) - read = Mock(return_value=None) - monkeypatch.setattr(server, "collect_device_data", read) + monkeypatch.setattr(server, "collect_device_data", Mock(return_value=None)) monkeypatch.setattr(server.time, "time", lambda: 100) - assert server.get_device_data(state) is None + stop = StopAfter(1) + server.poll_device(state, stop) + assert state.data is None assert state.timestamp == 100 - # A failed poll also has a backoff, rather than reconnecting on every GET. - assert server.get_device_data(state) is None - read.assert_called_once_with(state, 0) - monkeypatch.setattr(server.time, "time", lambda: 160) - assert server.get_device_data(state) is None - assert read.call_count == 2 + # A client must not wait for a device that is known to be offline. + assert state.polled.is_set() + assert stop.waits == [60] -def test_serial_collection_still_retries(monkeypatch): +def test_failed_serial_read_retries_soon_and_hides_a_short_failure(monkeypatch): data = JBDBMS() state = server.DeviceState(ser=Mock()) - read = Mock(side_effect=[None, data]) - sleep = Mock() + monkeypatch.setattr(server, "collect_device_data", Mock(side_effect=[data, None, None])) + clock = iter([100, 110, 131]) + monkeypatch.setattr(server.time, "time", lambda: next(clock)) + stop = StopAfter(3, observe=lambda: state.data) + server.poll_device(state, stop) + # The last snapshot stays for 30 seconds, then the device reads as absent. + assert stop.seen == [data, data, None] + assert stop.waits == [5, 1, 1] + + +def test_collection_error_does_not_end_polling(monkeypatch, capsys): + data = JBDBMS() + state = server.DeviceState(ser=Mock()) + monkeypatch.setattr(server, "collect_device_data", Mock(side_effect=[RuntimeError("boom"), data])) + stop = StopAfter(2) + server.poll_device(state, stop) + assert state.data is data + assert stop.waits == [1, 5] + assert "collection error: boom" in capsys.readouterr().out + + +def test_polling_threads_do_not_wait_for_each_other_and_stop(monkeypatch): + data = JBDBMS() + offline_started, release = threading.Event(), threading.Event() + working, offline = server.DeviceState(ser=Mock()), server.DeviceState(ser=Mock()) + + def read(device, debug=0): + if device is offline: + offline_started.set() + release.wait(5) + return None + return data + monkeypatch.setattr(server, "collect_device_data", read) - monkeypatch.setattr(server.time, "time", lambda: 100) - monkeypatch.setattr(server.time, "sleep", sleep) - assert server.get_device_data(state) is data - assert read.call_count == 2 - sleep.assert_called_once_with(1) + stop = threading.Event() + threads = server.start_polling({"offline": offline, "working": working}, stop) + try: + assert offline_started.wait(5) + assert working.polled.wait(5) + assert working.data is data + assert not offline.polled.is_set() + finally: + stop.set() + release.set() + for thread in threads: + thread.join(5) + assert not any(thread.is_alive() for thread in threads)