From 0c179d29e5717474ad6773f21df28277b436a27a Mon Sep 17 00:00:00 2001 From: Timothy Allen Date: Sat, 10 Oct 2026 11:47:16 +0200 Subject: [PATCH] Disconnect from Bluetooth devices before the server stops The polling threads are daemon threads, so a stop ended the server in the middle of a Bluetooth read. The read got no disconnect, and the adapter kept an open connection or a scan in progress. The server now sets the stop event at exit, closes its socket, and waits at most 20 seconds for the polling threads. A Bluetooth read that sees the stop event cancels its current step, disconnects, and returns. A read that waits for the adapter does not connect after a stop. --- README.md | 5 +++ bmspy/ecoflow_bt.py | 35 ++++++++++++++++- bmspy/server.py | 64 ++++++++++++++++++++++--------- tests/test_ecoflow_bt.py | 74 ++++++++++++++++++++++++++++++++++++ tests/test_server.py | 63 +++++++++++++++++++++++++++++- tests/test_server_ecoflow.py | 55 +++++++++++++++++++++++++-- 6 files changed, 272 insertions(+), 24 deletions(-) diff --git a/README.md b/README.md index 35747ac..bd7539d 100644 --- a/README.md +++ b/README.md @@ -104,6 +104,11 @@ first request after startup waits, until each device completes one read attempt. The server answers each client in its own thread, so a slow client does not delay the other clients. It closes a connection that stays silent for 5 seconds. +The server holds a Bluetooth connection only during a read, and disconnects +after each one. When the server stops, it ends a read in progress and +disconnects from that device before it exits. It waits at most 20 seconds for +the disconnect. + Each Bluetooth read listens for 10 seconds. If a required report is still missing, the read continues until the report arrives, for at most 30 seconds in total. A read that reaches this limit returns a partial snapshot. Its diff --git a/bmspy/ecoflow_bt.py b/bmspy/ecoflow_bt.py index 86511d9..7518e69 100644 --- a/bmspy/ecoflow_bt.py +++ b/bmspy/ecoflow_bt.py @@ -16,6 +16,7 @@ import contextlib import hashlib import logging import struct +import threading from collections.abc import Callable from dataclasses import dataclass from enum import StrEnum, auto @@ -38,6 +39,8 @@ MANUFACTURER_KEY = 0xB5B5 # How often a collection checks for the required reports. _COMPLETION_POLL_SECONDS = 0.5 +# How often a collection looks at its stop event. +_STOP_POLL_SECONDS = 0.2 # Missing reports of the last collection per address, to log changes only. _last_missing: dict[str, set[str]] = {} @@ -1052,7 +1055,9 @@ def _record_missing(address: str, snapshot: EcoFlowSnapshot, missing: set[str]) )) -def collect_data(cfg: EcoFlowBTConfig, debug: int = 0) -> EcoFlowSnapshot | bool: +def collect_data( + cfg: EcoFlowBTConfig, debug: int = 0, stop: threading.Event | None = None, +) -> EcoFlowSnapshot | bool: """Connect to an EcoFlow device, collect push data, disconnect, return snapshot. Analogous to :func:`bmspy.jbd_bms.collect_data`. Each call opens a fresh @@ -1065,14 +1070,23 @@ def collect_data(cfg: EcoFlowBTConfig, debug: int = 0) -> EcoFlowSnapshot | bool Each failure is logged with its reason, at every debug level. A snapshot from a collection that reached the limit is partial, and its ``ef_missing_measurements`` field names the reports that did not arrive. + + If *stop* is set before the collection, no connection is made. If it is + set during the collection, the collection ends early, disconnects, and + returns ``False``. """ + if stop is not None and stop.is_set(): + return False + disconnecting = False + def failed(reason: str) -> bool: # Always logged: without this line, a device that is out of reach # only shows as missing data. debugger(f"ecoflow_bt [{cfg.address}]: read failed: {reason}") return False - async def _run() -> EcoFlowSnapshot | bool: + async def _collect() -> EcoFlowSnapshot | bool: + nonlocal disconnecting device = EcoFlowBTDevice( address=cfg.address, serial_number=cfg.serial_number, @@ -1093,8 +1107,25 @@ def collect_data(cfg: EcoFlowBTConfig, debug: int = 0) -> EcoFlowSnapshot | bool _record_missing(cfg.address, device.snapshot, missing) return device.snapshot finally: + disconnecting = True await device.disconnect() + async def _run() -> EcoFlowSnapshot | bool: + if stop is None: + return await _collect() + task = asyncio.ensure_future(_collect()) + while not task.done() and not stop.is_set(): + await asyncio.wait({task}, timeout=_STOP_POLL_SECONDS) + # Cancel once, and never during the disconnect: the cancellation ends + # the current step, and the disconnect in _collect then runs in full. + if not task.done() and not disconnecting: + task.cancel() + try: + return await task + except asyncio.CancelledError: + debugger(f"ecoflow_bt [{cfg.address}]: read stopped, device disconnected") + return False + try: return asyncio.run(_run()) except Exception as exc: diff --git a/bmspy/server.py b/bmspy/server.py index 5f3a4ba..a91102d 100755 --- a/bmspy/server.py +++ b/bmspy/server.py @@ -44,6 +44,8 @@ SERIAL_RETRY_SECONDS = 1 SERIAL_STALE_SECONDS = 30 BLUETOOTH_POLL_SECONDS = 60 FIRST_POLL_TIMEOUT = 120 +# Time limit at exit for the reads in progress to end and to disconnect. +POLL_STOP_TIMEOUT = 20 # Time limit for each read from a client and each write to it. CLIENT_TIMEOUT_SECONDS = 5 # A request is a short JSON object. A larger length prefix is an error. @@ -189,13 +191,18 @@ def parse_ecoflow_spec(spec: str) -> tuple[str, str, str]: return parts[0].upper(), parts[1], parts[2] -def collect_device_data(device: DeviceState, debug: int = 0) -> UPS | None: - """Collect fresh data from whichever device type is in *device*.""" +def collect_device_data( + device: DeviceState, debug: int = 0, stop: threading.Event | None = None, +) -> UPS | None: + """Collect fresh data from whichever device type is in *device*. + + A Bluetooth read ends early and disconnects when *stop* is set. + """ if device.ser is not None: return collect_data(device.ser, debug) or None if device.ecoflow is not None: with bluetooth_lock: - return ecoflow_collect_data(device.ecoflow, debug) or None + return ecoflow_collect_data(device.ecoflow, debug, stop) or None return None @@ -209,7 +216,7 @@ def poll_device(device: DeviceState, stop: threading.Event, debug: int = 0) -> N last_success = 0.0 while not stop.is_set(): try: - data = collect_device_data(device, debug) + data = collect_device_data(device, debug, stop) except Exception as e: # A failed read must not end polling for this device. debugger("server: collection error: {}".format(e)) @@ -247,6 +254,24 @@ def start_polling( return threads +def stop_polling(threads: list[threading.Thread], stop: threading.Event) -> None: + """End the polling threads, and wait for a Bluetooth read to disconnect. + + The threads are daemon threads, so the process does not wait for them. A + Bluetooth read that ends without a disconnect leaves the adapter with an + open connection or a scan in progress. + """ + stop.set() + deadline = time.monotonic() + POLL_STOP_TIMEOUT + for thread in threads: + thread.join(max(0.0, deadline - time.monotonic())) + alive = [thread.name for thread in threads if thread.is_alive()] + if alive: + debugger("server: no clean stop in {} seconds for {}".format( + POLL_STOP_TIMEOUT, ", ".join(alive), + )) + + def handle_client( connection: socket.socket, ups_devices: dict[str, DeviceState], debug: int = 0, ) -> None: @@ -491,20 +516,25 @@ def main(): sock.listen(1) - stop_polling = threading.Event() - atexit.register(stop_polling.set) - start_polling(ups_devices, stop_polling, debug) + stop = threading.Event() + threads = start_polling(ups_devices, stop, debug) - while True: - if debug > 2: - debugger("socket: waiting for a connection") - try: - connection, _ = sock.accept() - except KeyboardInterrupt: - sys.exit(1) - threading.Thread( - target=handle_client, args=(connection, ups_devices, debug), name="client", daemon=True, - ).start() + try: + while True: + if debug > 2: + debugger("socket: waiting for a connection") + try: + connection, _ = sock.accept() + except KeyboardInterrupt: + sys.exit(1) + threading.Thread( + target=handle_client, args=(connection, ups_devices, debug), name="client", daemon=True, + ).start() + finally: + # SIGTERM arrives here as SystemExit. Refuse new clients at once, then + # let each device finish: a Bluetooth read must disconnect first. + sock.close() + stop_polling(threads, stop) if __name__ == "__main__": diff --git a/tests/test_ecoflow_bt.py b/tests/test_ecoflow_bt.py index d93e0f6..a9ef19e 100644 --- a/tests/test_ecoflow_bt.py +++ b/tests/test_ecoflow_bt.py @@ -9,6 +9,8 @@ import asyncio import importlib.util import struct import sys +import threading +import time from argparse import Namespace from dataclasses import asdict from itertools import permutations @@ -538,6 +540,78 @@ def test_successful_collection_logs_no_failure(monkeypatch, capsys): assert capsys.readouterr().out == "" +def _stoppable_device(monkeypatch, **steps): + """Stand-in device that records whether its disconnect ran in full.""" + events = [] + + async def disconnect(): + events.append("disconnect started") + await asyncio.sleep(0.05) + events.append("disconnect done") + + fake = Mock( + connect=AsyncMock(return_value=True), wait_authenticated=AsyncMock(return_value=True), + wait_for_measurements=AsyncMock(return_value=set()), disconnect=disconnect, + _pkt_count=1, last_error=None, + ) + fake.configure_mock(**steps) + monkeypatch.setattr("bmspy.ecoflow_bt.EcoFlowBTDevice", Mock(return_value=fake)) + monkeypatch.setattr("bmspy.ecoflow_bt._STOP_POLL_SECONDS", 0.01) + monkeypatch.setattr("bmspy.ecoflow_bt._last_missing", {}) + return fake, events + + +@pytest.mark.parametrize("step", ["connect", "wait_authenticated", "wait_for_measurements"]) +def test_stop_during_a_collection_disconnects_before_it_returns(monkeypatch, capsys, step): + """A server that stops in the middle of a read must not leave the link open.""" + stop = threading.Event() + + async def interrupted(*args, **kwargs): + stop.set() + await asyncio.sleep(30) + + fake, events = _stoppable_device(monkeypatch, **{step: interrupted}) + config = initialise_device(ADDRESS, SERIAL, "test-user", listen_duration=0) + started = time.monotonic() + assert collect_data(config, stop=stop) is False + assert time.monotonic() - started < 5 + assert events == ["disconnect started", "disconnect done"] + lines = capsys.readouterr().out.splitlines() + assert len(lines) == 1 + assert lines[0].endswith(f"ecoflow_bt [{ADDRESS}]: read stopped, device disconnected") + + +def test_stop_during_the_disconnect_lets_the_disconnect_finish(monkeypatch): + stop = threading.Event() + fake, events = _stoppable_device(monkeypatch) + slow_disconnect = fake.disconnect + + async def disconnect(): + stop.set() + await slow_disconnect() + + fake.disconnect = disconnect + config = initialise_device(ADDRESS, SERIAL, "test-user", listen_duration=0) + assert collect_data(config, stop=stop) is fake.snapshot + assert events == ["disconnect started", "disconnect done"] + + +def test_collection_does_not_connect_after_stop(monkeypatch): + stop = threading.Event() + stop.set() + device = Mock() + monkeypatch.setattr("bmspy.ecoflow_bt.EcoFlowBTDevice", device) + assert collect_data(initialise_device(ADDRESS, SERIAL, "test-user"), stop=stop) is False + device.assert_not_called() + + +def test_collection_with_an_unset_stop_event_returns_the_snapshot(monkeypatch): + fake, events = _stoppable_device(monkeypatch) + config = initialise_device(ADDRESS, SERIAL, "test-user", listen_duration=0) + assert collect_data(config, stop=threading.Event()) is fake.snapshot + assert events == ["disconnect started", "disconnect done"] + + def test_connect_records_the_reason_for_a_failure(monkeypatch): client = Mock(connect=AsyncMock(side_effect=TimeoutError("no answer")), disconnect=AsyncMock(), is_connected=False) monkeypatch.setitem(sys.modules, "bleak", SimpleNamespace(BleakClient=Mock(return_value=client))) diff --git a/tests/test_server.py b/tests/test_server.py index 7ddd762..b1aa216 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -25,6 +25,7 @@ def _poll_once(devices, stop, debug=0): for device in devices.values(): device.data = collect_device_data(device, debug) device.polled.set() + return [] # --------------------------------------------------------------------------- @@ -217,6 +218,39 @@ listener.accept() assert result.stderr.strip() == "terminating" assert not socket_path.exists() + def test_sigterm_during_a_bluetooth_read_waits_for_the_disconnect(self, tmp_path): + """The server process itself, with a stand-in for the Bluetooth read.""" + socket_path = tmp_path / "bms.sock" + script = """ +import os +import signal +import sys +import threading +from bmspy import server + +def bluetooth_read(config, debug=0, stop=None): + if stop.is_set(): + return False + print("read started", flush=True) + threading.Timer(0.1, os.kill, args=(os.getpid(), signal.SIGTERM)).start() + stop.wait(10) + print("disconnected", flush=True) + return False + +server.ecoflow_collect_data = bluetooth_read +sys.argv = ["bmspy-server", "--socket", sys.argv[1], + "-d", "river@02:00:00:00:00:01/R601TEST00000001/test-user"] +server.main() +""" + result = subprocess.run( + [sys.executable, "-c", script, str(socket_path)], + capture_output=True, text=True, timeout=30, + ) + assert result.stdout.splitlines() == ["read started", "disconnected"] + assert result.returncode == 1 + assert result.stderr.strip() == "terminating" + assert not socket_path.exists() + # --------------------------------------------------------------------------- # socket_cleanup @@ -765,7 +799,7 @@ class TestServerMain: patch("bmspy.server.signal.signal"), \ patch("bmspy.server.atexit.register"), \ patch("bmspy.server.initialise_serial"), \ - patch("bmspy.server.start_polling"), \ + patch("bmspy.server.start_polling", return_value=[]), \ patch("bmspy.server.os.path.isdir", return_value=True), \ patch.object(socket_module.socket, "accept", side_effect=KeyboardInterrupt): from bmspy.server import main as server_main @@ -773,6 +807,31 @@ class TestServerMain: server_main() assert exit_info.value.code == 1 + def test_exit_closes_the_socket_and_stops_polling(self, tmp_path): + sock_path = str(tmp_path / "server_stop.sock") + threads = [object()] + listeners = [] + original_listen = socket_module.socket.listen + + def listen(self, backlog=1): + listeners.append(self) + return original_listen(self, backlog) + + with patch("sys.argv", ["bmspy-server", "--socket", sock_path, "--device", "/dev/ttyUSB0"]), \ + patch("bmspy.server.signal.signal"), \ + patch("bmspy.server.atexit.register"), \ + patch("bmspy.server.initialise_serial"), \ + patch("bmspy.server.start_polling", return_value=threads) as start, \ + patch("bmspy.server.stop_polling") as stop_polling, \ + patch("bmspy.server.os.path.isdir", return_value=True), \ + patch.object(socket_module.socket, "listen", listen), \ + patch.object(socket_module.socket, "accept", side_effect=KeyboardInterrupt): + from bmspy.server import main as server_main + with pytest.raises(SystemExit): + server_main() + stop_polling.assert_called_once_with(threads, start.call_args.args[1]) + assert listeners[0].fileno() == -1 + def test_socket_read_error_logs_and_continues(self, tmp_path, capsys): """Test that read_request errors are caught and logged.""" sock_path = str(tmp_path / "server_err.sock") @@ -1098,7 +1157,7 @@ class TestServerMain: patch("bmspy.server.signal.signal"), \ patch("bmspy.server.atexit.register"), \ patch("bmspy.server.initialise_serial"), \ - patch("bmspy.server.start_polling"), \ + patch("bmspy.server.start_polling", return_value=[]), \ patch("bmspy.server.os.path.isdir", return_value=True), \ patch.object(socket_module.socket, "accept", side_effect=OSError("accept failed")) as accept: from bmspy.server import main as server_main diff --git a/tests/test_server_ecoflow.py b/tests/test_server_ecoflow.py index 00d1675..2b43a1e 100644 --- a/tests/test_server_ecoflow.py +++ b/tests/test_server_ecoflow.py @@ -1,6 +1,7 @@ """Server routing and caching tests with synthetic device identifiers.""" import threading +import time from unittest.mock import Mock import pytest @@ -71,7 +72,7 @@ def test_collection_dispatches_to_the_selected_backend(monkeypatch): assert server.collect_device_data(server.DeviceState(ser=port), 2) is serial_data serial_read.assert_called_once_with(port, 2) assert server.collect_device_data(server.DeviceState(ecoflow=config), 3) is bluetooth_data - bluetooth_read.assert_called_once_with(config, 3) + bluetooth_read.assert_called_once_with(config, 3, None) assert server.collect_device_data(server.DeviceState()) is None @@ -104,7 +105,7 @@ def test_poll_interval_depends_on_backend(monkeypatch, bluetooth, interval): assert state.timestamp == 100 assert state.polled.is_set() assert stop.waits == [interval] - read.assert_called_once_with(state, 2) + read.assert_called_once_with(state, 2, stop) def test_offline_bluetooth_does_not_retry_at_once_or_serve_stale_data(monkeypatch): @@ -151,7 +152,7 @@ def test_polling_threads_do_not_wait_for_each_other_and_stop(monkeypatch): offline_started, release = threading.Event(), threading.Event() working, offline = server.DeviceState(ser=Mock()), server.DeviceState(ser=Mock()) - def read(device, debug=0): + def read(device, debug=0, stop=None): if device is offline: offline_started.set() release.wait(5) @@ -172,3 +173,51 @@ def test_polling_threads_do_not_wait_for_each_other_and_stop(monkeypatch): for thread in threads: thread.join(5) assert not any(thread.is_alive() for thread in threads) + + +def test_stop_polling_waits_for_the_bluetooth_disconnect(monkeypatch): + events = [] + + def bluetooth_read(config, debug=0, stop=None): + if stop.is_set(): + return False + events.append("read started") + assert stop.wait(5) + events.append("disconnected") + return False + + monkeypatch.setattr(server, "ecoflow_collect_data", bluetooth_read) + # The second device waits for the first one. It must not start a read. + devices = { + name: server.DeviceState(ecoflow=initialise_device(ADDRESS, SERIAL, "test-user")) + for name in ("first", "second") + } + stop = threading.Event() + threads = server.start_polling(devices, stop) + while not events: + time.sleep(0.01) + server.stop_polling(threads, stop) + assert events == ["read started", "disconnected"] + assert not any(thread.is_alive() for thread in threads) + + +def test_stop_polling_reports_a_read_that_does_not_end(monkeypatch, capsys): + started, release = threading.Event(), threading.Event() + + def stuck_read(device, debug=0, stop=None): + started.set() + release.wait(5) + + monkeypatch.setattr(server, "collect_device_data", stuck_read) + monkeypatch.setattr(server, "POLL_STOP_TIMEOUT", 0.05) + stop = threading.Event() + threads = server.start_polling({"stuck": server.DeviceState(ser=Mock())}, stop) + try: + assert started.wait(5) + server.stop_polling(threads, stop) + assert "server: no clean stop in 0.05 seconds for poll-stuck" in capsys.readouterr().out + finally: + release.set() + for thread in threads: + thread.join(5) + assert not any(thread.is_alive() for thread in threads)