From f98cc2d6a3b67e90d35401ea6758a4b3b2d9ea19 Mon Sep 17 00:00:00 2001 From: Fedorov Dmitriy Date: Sat, 19 Sep 2026 15:48:44 +0300 Subject: [PATCH] =?UTF-8?q?work:=20=D0=90=D0=B2=D1=82=D0=BE=D0=BC=D0=B0?= =?UTF-8?q?=D1=82=D0=B8=D1=87=D0=B5=D1=81=D0=BA=D0=B8=D0=B9=20=D1=80=D0=B5?= =?UTF-8?q?=D0=B6=D0=B8=D0=BC=20=D1=83=D0=BF=D1=80=D0=B0=D0=B2=D0=BB=D0=B5?= =?UTF-8?q?=D0=BD=D0=B8=D1=8F=20=D1=81=D0=BA=D0=BE=D1=80=D0=BE=D1=81=D1=82?= =?UTF-8?q?=D1=8C=D1=8E=20=D0=BE=D1=82=D0=BB=D0=B0=D0=B6=D0=B5=D0=BD!?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/api.py | 5 +- app/my_dataclasses.py | 22 ++ app/qingping/service.py | 546 ++++++++++++++++++++++++-------- config/auto.yaml | 4 +- config/schedule.yaml | 20 +- tests/test_qingping_recovery.py | 415 ++++++++++++++++++++++++ 6 files changed, 867 insertions(+), 145 deletions(-) create mode 100644 tests/test_qingping_recovery.py diff --git a/app/api.py b/app/api.py index f7f3825..37705e8 100644 --- a/app/api.py +++ b/app/api.py @@ -219,6 +219,7 @@ def get_status() -> dict: if state is not None else None ), + "auto": get_auto_status(), "qingping": qingping_service.status(), @@ -430,9 +431,7 @@ def get_auto_status() -> dict: "target_speed": None, "auto_speed": None, "last_error": None, - "config_error": ( - auto_load_error - ), + "config_error": auto_load_error, } return { diff --git a/app/my_dataclasses.py b/app/my_dataclasses.py index 13dacff..72663e1 100644 --- a/app/my_dataclasses.py +++ b/app/my_dataclasses.py @@ -42,5 +42,27 @@ AUTO_CONFIG_FILE = ( / "auto.yaml" ) + + + +WATCHDOG_INTERVAL = 5.0 + +# Type 13 обычно приходит примерно раз в 60 секунд. +# Даём дополнительный запас. +HEARTBEAT_STALE_AFTER = 90.0 + +# После появления устройства ждём первый sensor sample +# не дольше этого времени. +FIRST_SAMPLE_TIMEOUT = 30.0 + +# Если новые sensor samples не приходят дольше этого времени, +# считаем sensor stream зависшим. +SAMPLE_STALE_AFTER = 45.0 + + + +QINGPING_SAMPLE_TIMEOUT = 60.0 +QINGPING_RECOVERY_TIMEOUT = 30.0 + #Классы diff --git a/app/qingping/service.py b/app/qingping/service.py index 2bd23d7..c06f2a6 100644 --- a/app/qingping/service.py +++ b/app/qingping/service.py @@ -202,7 +202,11 @@ from datetime import datetime, timezone import paho.mqtt.client as mqtt from .models import QingpingState - +from app.my_dataclasses import ( + WATCHDOG_INTERVAL, + QINGPING_SAMPLE_TIMEOUT, + QINGPING_RECOVERY_TIMEOUT, + ) logger = logging.getLogger(__name__) @@ -228,13 +232,12 @@ class QingpingService: self._watchdog_task: asyncio.Task | None = None self._last_sample_monotonic: float | None = None - self._connected_since: float | None = None - self._heartbeat_seen = False - self._recovery_sent = False + self._device_online = False + + self._recovery_waiting = False + self._recovery_started_monotonic: float | None = None - self._last_device_timestamp: int | None = None - self._reboot_detected = False @property def state(self) -> QingpingState: @@ -243,15 +246,11 @@ class QingpingService: @property def online(self) -> bool: - state = self.state - - if not state.mqtt_connected: - return False - - if self._last_sample_monotonic is None: - return False - - return time.monotonic() - self._last_sample_monotonic < 60 + with self._lock: + return ( + self._state.mqtt_connected + and self._device_online + ) async def start(self) -> None: client = mqtt.Client( @@ -298,128 +297,206 @@ class QingpingService: self._client.loop_stop() self._client = None + def status(self) -> dict: - result = self.state.to_dict() - result["online"] = self.online - return result + with self._lock: + state = self._state + + online = ( + state.mqtt_connected + and self._device_online + ) + + recovery_waiting = ( + self._recovery_waiting + ) + + return { + **state.to_dict(), + "online": online, + "recovery_waiting": recovery_waiting, + } def _on_connect( - self, - client, - userdata, - flags, - reason_code, - properties, - ): - if reason_code.is_failure: - logger.warning( + self, + client, + _userdata, + _flags, + reason_code, + _properties=None, + ) -> None: + if reason_code != 0: + logger.error( "Qingping MQTT connection failed: %s", reason_code, ) return + logger.info( + "Qingping MQTT connected" + ) + client.subscribe(self._up_topic) + now_monotonic = time.monotonic() + with self._lock: self._state = replace( self._state, mqtt_connected=True, + temperature=None, + humidity=None, + co2=None, + pm25=None, + pm10=None, + battery=None, + sample_timestamp=None, + sample_received_at=None, ) - self._connected_since = time.monotonic() - self._heartbeat_seen = False - self._recovery_sent = False + self._last_sample_monotonic = None + + self._device_online = False + + self._recovery_waiting = True + self._recovery_started_monotonic = ( + now_monotonic + ) logger.info( - "Qingping MQTT connected" + "Qingping startup initialization" ) + self._send_recovery() + def _on_disconnect( - self, - client, - userdata, - disconnect_flags, - reason_code, - properties, - ): + self, + _client, + _userdata, + _disconnect_flags, + reason_code, + _properties=None, + ) -> None: + logger.warning( + "Qingping MQTT disconnected: %s", + reason_code, + ) + with self._lock: self._state = replace( self._state, mqtt_connected=False, ) - logger.warning( - "Qingping MQTT disconnected: %s", - reason_code, - ) + self._device_online = False + self._recovery_waiting = False + self._recovery_started_monotonic = None def _on_message( - self, - client, - userdata, - message, - ): + self, + _client, + _userdata, + message, + ) -> None: try: payload = json.loads( message.payload.decode("utf-8") ) - except (UnicodeDecodeError, json.JSONDecodeError): - logger.warning( - "Invalid Qingping MQTT message" - ) - return - now = datetime.now(timezone.utc) + now = datetime.now(timezone.utc) + now_monotonic = time.monotonic() - with self._lock: - self._state = replace( - self._state, - last_message_at=now, + message_type = int( + payload.get("type") ) - message_type = str(payload.get("type")) - - if message_type == "13": - self._handle_heartbeat(payload) - return - - if message_type in {"12", "17"}: - self._handle_sensor_data( - payload, - now, + logger.debug( + "Qingping MQTT packet received: type=%s", + message_type, ) + # Любой пакет означает, что устройство + # физически присутствует в MQTT. + with self._lock: + self._state = replace( + self._state, + last_message_at=now, + ) + + start_recovery = False + + with self._lock: + if ( + not self._device_online + and + not self._recovery_waiting + ): + self._state = replace( + self._state, + temperature=None, + humidity=None, + co2=None, + pm25=None, + pm10=None, + battery=None, + sample_timestamp=None, + sample_received_at=None, + ) + + self._last_sample_monotonic = None + + self._device_online = True + + self._recovery_waiting = True + self._recovery_started_monotonic = ( + now_monotonic + ) + + start_recovery = True + + if start_recovery: + logger.info( + "Qingping packet received while offline: " + "type=%s -> starting recovery", + message_type, + ) + + self._send_recovery() + + # Единственный пакет, который реально + # обрабатываем как данные. + if message_type == 17: + self._handle_sensor_data( + payload, + now, + ) + return + + if message_type == 13: + self._handle_heartbeat( + payload, + ) + return + + logger.debug( + "Qingping service packet received: type=%s", + message_type, + ) + + except Exception: + logger.exception( + "Qingping MQTT packet processing failed" + ) + + def _handle_heartbeat( - self, - payload: dict, + self, + payload: dict, ) -> None: - self._heartbeat_seen = True - - device_timestamp = payload.get("timestamp") - - if isinstance(device_timestamp, int): - previous = self._last_device_timestamp - - if ( - device_timestamp < 300 - and ( - previous is None - or previous > 300 - ) - ): - self._reboot_detected = True - self._recovery_sent = False - - logger.info( - "Qingping reboot detected" - ) - - self._last_device_timestamp = device_timestamp + wifi_info = payload.get("wifi_info") wifi_rssi = None - wifi_info = payload.get("wifi_info") - if isinstance(wifi_info, str): parts = wifi_info.split(",") @@ -427,15 +504,26 @@ class QingpingService: try: wifi_rssi = int(parts[1]) except ValueError: - pass + logger.warning( + "Qingping invalid RSSI in wifi_info: %r", + wifi_info, + ) + + firmware = payload.get("sw_version") with self._lock: self._state = replace( self._state, wifi_rssi=wifi_rssi, - firmware=payload.get("sw_version"), + firmware=firmware, ) + logger.debug( + "Qingping heartbeat updated: " + "rssi=%s firmware=%s", + wifi_rssi, + firmware, + ) def _handle_sensor_data( self, payload: dict, @@ -498,7 +586,30 @@ class QingpingService: "battery", ) + now_monotonic = time.monotonic() + with self._lock: + current_timestamp = ( + self._state.sample_timestamp + ) + + # CGDN1 может повторять одну историческую точку. + # Такой пакет НЕ считается новым измерением. + if ( + current_timestamp is not None + and sample_timestamp <= current_timestamp + ): + logger.debug( + "Qingping duplicate sensor sample ignored: " + "timestamp=%s current=%s", + sample_timestamp, + current_timestamp, + ) + return + + was_recovering = self._recovery_waiting + was_offline = not self._device_online + self._state = replace( self._state, temperature=temperature, @@ -511,16 +622,123 @@ class QingpingService: sample_received_at=received_at, ) - self._last_sample_monotonic = ( - time.monotonic() - ) + self._last_sample_monotonic = ( + now_monotonic + ) - self._reboot_detected = False + self._device_online = True + + self._recovery_waiting = False + self._recovery_started_monotonic = None + + # if was_offline or was_recovering: + # logger.info( + # "Qingping sensor stream online: " + # "timestamp=%s co2=%s", + # sample_timestamp, + # co2, + # ) + # else: + # logger.debug( + # "Qingping sensor sample accepted: " + # "timestamp=%s co2=%s", + # sample_timestamp, + # co2, + # ) + + async def _watchdog_loop(self) -> None: + while True: + await asyncio.sleep( + WATCHDOG_INTERVAL + ) + + self._watchdog_tick( + time.monotonic() + ) + + def _watchdog_tick( + self, + now: float, + ) -> None: + recovery_timeout = None + sample_timeout = None + invalid_recovery_state = False + + with self._lock: + + # Ждём type 17 после recovery. + if self._recovery_waiting: + + if ( + self._recovery_started_monotonic + is None + ): + invalid_recovery_state = True + + self._recovery_waiting = False + self._device_online = False + + else: + elapsed = ( + now + - self._recovery_started_monotonic + ) + + if ( + elapsed + > QINGPING_RECOVERY_TIMEOUT + ): + recovery_timeout = elapsed + + self._recovery_waiting = False + self._recovery_started_monotonic = None + self._device_online = False + + # Нормальная работа: + # следим за последним НОВЫМ type 17. + elif self._last_sample_monotonic is not None: + + elapsed = ( + now + - self._last_sample_monotonic + ) + + if ( + elapsed + > QINGPING_SAMPLE_TIMEOUT + and + self._device_online + ): + sample_timeout = elapsed + + self._device_online = False + + if invalid_recovery_state: + logger.error( + "Qingping invalid recovery state: " + "recovery_waiting=True but " + "recovery_started_monotonic=None" + ) + + if recovery_timeout is not None: + logger.warning( + "Qingping recovery timeout: " + "no type 17 for %.1f sec " + "-> offline", + recovery_timeout, + ) + + if sample_timeout is not None: + logger.warning( + "Qingping sensor stream lost: " + "no type 17 for %.1f sec " + "-> offline", + sample_timeout, + ) @staticmethod - def _sample_timestamp( - sample: dict, - ) -> int: + def _sample_timestamp(sample: dict) -> int: + timestamp = sample.get("timestamp") if isinstance(timestamp, dict): @@ -532,10 +750,7 @@ class QingpingService: return 0 @staticmethod - def _value( - sample: dict, - key: str, - ): + def _value(sample: dict, key: str): value = sample.get(key) if isinstance(value, dict): @@ -545,35 +760,92 @@ class QingpingService: async def _watchdog_loop(self) -> None: while True: - await asyncio.sleep(5) + await asyncio.sleep( + WATCHDOG_INTERVAL + ) + + now = time.monotonic() + + # -------------------------------------------------- + # СЦЕНАРИЙ 1 + # + # Recovery уже отправлен. + # Ждём type 17 максимум 30 секунд. + # -------------------------------------------------- + + if self._recovery_waiting: + + if self._recovery_started_monotonic is None: + logger.error( + "Qingping invalid recovery state: " + "recovery_waiting=True but " + "recovery_started_monotonic=None" + ) + + self._recovery_waiting = False + self._device_online = False + continue + + elapsed = now - self._recovery_started_monotonic + + if elapsed > QINGPING_RECOVERY_TIMEOUT: + + logger.warning( + "Qingping recovery timeout: " + "no type 17 for %.1f sec " + "-> offline", + elapsed, + ) + + self._recovery_waiting = False + self._recovery_started_monotonic = ( + None + ) + + self._device_online = False - if self._client is None: continue - state = self.state + # -------------------------------------------------- + # СЦЕНАРИЙ 2 + # + # До сих пор не получили вообще ни одного + # измерения type 17. + # + # Ничего делать не надо. + # Первый любой MQTT-пакет запустит recovery. + # -------------------------------------------------- - if not state.mqtt_connected: + if self._last_sample_monotonic is None: continue - if self._recovery_sent: - continue + # -------------------------------------------------- + # СЦЕНАРИЙ 3 + # + # Нормально работали, но type 17 + # перестали приходить. + # -------------------------------------------------- - # Явно увидели reboot CGDN1. - if self._reboot_detected: - self._send_recovery() - continue + elapsed = ( + now + - self._last_sample_monotonic + ) - # Или сервис подключился, heartbeat есть, - # но свежих sensorData так и не появилось. if ( - self._connected_since is not None - and self._heartbeat_seen - and self._last_sample_monotonic is None - and time.monotonic() - - self._connected_since - > 30 + elapsed + > QINGPING_SAMPLE_TIMEOUT + and + self._device_online ): - self._send_recovery() + logger.warning( + "Qingping sensor stream lost: " + "no type 17 for %.1f sec " + "-> offline", + elapsed, + ) + + self._device_online = False + def _send_recovery(self) -> None: if self._client is None: @@ -598,8 +870,6 @@ class QingpingService: ) if result.rc == mqtt.MQTT_ERR_SUCCESS: - self._recovery_sent = True - logger.warning( "Qingping recovery command sent" ) @@ -607,4 +877,24 @@ class QingpingService: logger.warning( "Failed to send Qingping recovery: %s", result.rc, - ) \ No newline at end of file + ) + + def _reset_sample_session(self) -> None: + logger.debug( + "Qingping sensor session reset" + ) + + with self._lock: + self._state = replace( + self._state, + temperature=None, + humidity=None, + co2=None, + pm25=None, + pm10=None, + battery=None, + sample_timestamp=None, + sample_received_at=None, + ) + + self._last_sample_monotonic = None \ No newline at end of file diff --git a/config/auto.yaml b/config/auto.yaml index 1e4237b..bb936b2 100644 --- a/config/auto.yaml +++ b/config/auto.yaml @@ -7,10 +7,10 @@ co2: hysteresis: 100 thresholds: - - ppm: 800 + - ppm: 700 speed: 2 - - ppm: 1000 + - ppm: 900 speed: 3 - ppm: 1300 diff --git a/config/schedule.yaml b/config/schedule.yaml index d4cbb3e..29e24b3 100644 --- a/config/schedule.yaml +++ b/config/schedule.yaml @@ -8,30 +8,30 @@ templates: workday: - - time: "00:00" - action: - type: set - power: off - - time: "07:00" action: type: set power: on speed: 2 - heater: on + heater: off target_temp: 20 mode: outside - - time: "09:00" + - time: "15:45" action: type: set speed: 1 - - time: "13:00" + - time: "15:46" action: type: auto speed: 1 + - time: "15:47" + action: + type: set + speed: 6 + - time: "17:00" action: type: set @@ -73,10 +73,6 @@ templates: type: auto speed: 2 - - time: "00:05" - action: - type: set - speed: 1 weekend: diff --git a/tests/test_qingping_recovery.py b/tests/test_qingping_recovery.py new file mode 100644 index 0000000..38e1248 --- /dev/null +++ b/tests/test_qingping_recovery.py @@ -0,0 +1,415 @@ +import json +from types import SimpleNamespace +from unittest.mock import patch + +import paho.mqtt.client as mqtt + +from app.my_dataclasses import ( + QINGPING_SAMPLE_TIMEOUT, + QINGPING_RECOVERY_TIMEOUT, +) +from app.qingping.service import QingpingService + + +class FakeClock: + def __init__(self): + self.now = 0.0 + + def monotonic(self) -> float: + return self.now + + def advance(self, seconds: float) -> None: + self.now += seconds + + +class FakeMqttClient: + def __init__(self): + self.subscriptions = [] + self.published = [] + + def subscribe(self, topic): + self.subscriptions.append(topic) + + def publish( + self, + topic, + payload, + ): + self.published.append( + ( + topic, + json.loads(payload), + ) + ) + + return SimpleNamespace( + rc=mqtt.MQTT_ERR_SUCCESS + ) + + +class FakeMessage: + def __init__(self, payload: dict): + self.payload = json.dumps( + payload + ).encode("utf-8") + + +def sensor_message( + timestamp: int, + co2: int, +) -> FakeMessage: + return FakeMessage( + { + "type": 17, + "sensorData": [ + { + "timestamp": { + "value": timestamp + }, + "temperature": { + "value": 25.0 + }, + "humidity": { + "value": 50.0 + }, + "co2": { + "value": co2 + }, + "pm25": { + "value": 1 + }, + "pm10": { + "value": 1 + }, + "battery": { + "value": 100 + }, + } + ], + } + ) + + +def service_message() -> FakeMessage: + return FakeMessage( + { + "type": 13, + "wifi_info": ( + "TestWiFi,-50,1," + "00:00:00:00:00:00" + ), + "sw_version": "4.8.5", + } + ) + + +def assert_state( + service: QingpingService, + *, + online: bool, + recovery_waiting: bool, + co2, +): + status = service.status() + + assert status["online"] is online, status + assert ( + status["recovery_waiting"] + is recovery_waiting + ), status + + assert status["co2"] == co2, status + + +def test_double_recovery_cycle() -> None: + print() + print("TEST — DOUBLE RECOVERY CYCLE") + + clock = FakeClock() + + service = QingpingService( + host="127.0.0.1", + ) + + client = FakeMqttClient() + + # _send_recovery() использует self._client. + service._client = client + + with patch( + "app.qingping.service.time.monotonic", + clock.monotonic, + ): + + # -------------------------------------------------- + # START + # -------------------------------------------------- + + print("1. MQTT connect") + + service._on_connect( + client, + None, + None, + 0, + None, + ) + + assert len(client.published) == 1 + + assert_state( + service, + online=False, + recovery_waiting=True, + co2=None, + ) + + # -------------------------------------------------- + # Первый type 17. + # -------------------------------------------------- + + print("2. First sensor sample") + + service._on_message( + None, + None, + sensor_message( + timestamp=1000, + co2=900, + ), + ) + + assert_state( + service, + online=True, + recovery_waiting=False, + co2=900, + ) + + # -------------------------------------------------- + # Первый отказ. + # -------------------------------------------------- + + print("3. First sensor timeout") + + clock.advance( + QINGPING_SAMPLE_TIMEOUT + 1 + ) + + service._watchdog_tick( + clock.monotonic() + ) + + assert_state( + service, + online=False, + recovery_waiting=False, + co2=900, + ) + + # Recovery пока НЕ отправлялся. + assert len(client.published) == 1 + + # -------------------------------------------------- + # Устройство снова появляется. + # Любой пакет запускает recovery. + # -------------------------------------------------- + + print("4. First recovery") + + service._on_message( + None, + None, + service_message(), + ) + + assert len(client.published) == 2 + + assert_state( + service, + online=True, + recovery_waiting=True, + co2=None, + ) + + # -------------------------------------------------- + # Новый type 17. + # -------------------------------------------------- + + print("5. First recovery completed") + + service._on_message( + None, + None, + sensor_message( + timestamp=1015, + co2=1000, + ), + ) + + assert_state( + service, + online=True, + recovery_waiting=False, + co2=1000, + ) + + # -------------------------------------------------- + # Второй отказ. + # -------------------------------------------------- + + print("6. Second sensor timeout") + + clock.advance( + QINGPING_SAMPLE_TIMEOUT + 1 + ) + + service._watchdog_tick( + clock.monotonic() + ) + + assert_state( + service, + online=False, + recovery_waiting=False, + co2=1000, + ) + + assert len(client.published) == 2 + + # -------------------------------------------------- + # Второе восстановление. + # -------------------------------------------------- + + print("7. Second recovery") + + service._on_message( + None, + None, + service_message(), + ) + + assert len(client.published) == 3 + + assert_state( + service, + online=True, + recovery_waiting=True, + co2=None, + ) + + service._on_message( + None, + None, + sensor_message( + timestamp=1030, + co2=1100, + ), + ) + + assert_state( + service, + online=True, + recovery_waiting=False, + co2=1100, + ) + + print("8. Second recovery completed") + + +def test_recovery_timeout_and_retry() -> None: + print() + print("TEST — RECOVERY TIMEOUT AND RETRY") + + clock = FakeClock() + + service = QingpingService( + host="127.0.0.1", + ) + + client = FakeMqttClient() + + service._client = client + + with patch( + "app.qingping.service.time.monotonic", + clock.monotonic, + ): + + # Startup recovery. + service._on_connect( + client, + None, + None, + 0, + None, + ) + + assert len(client.published) == 1 + + assert_state( + service, + online=False, + recovery_waiting=True, + co2=None, + ) + + # Type 17 так и не пришёл. + clock.advance( + QINGPING_RECOVERY_TIMEOUT + 1 + ) + + service._watchdog_tick( + clock.monotonic() + ) + + assert_state( + service, + online=False, + recovery_waiting=False, + co2=None, + ) + + # Любой следующий пакет должен разрешить + # новую recovery-попытку. + service._on_message( + None, + None, + service_message(), + ) + + assert len(client.published) == 2 + + assert_state( + service, + online=True, + recovery_waiting=True, + co2=None, + ) + + # Теперь приходит type 17. + service._on_message( + None, + None, + sensor_message( + timestamp=2000, + co2=1200, + ), + ) + + assert_state( + service, + online=True, + recovery_waiting=False, + co2=1200, + ) + + +if __name__ == "__main__": + test_double_recovery_cycle() + test_recovery_timeout_and_retry() + + print() + print( + "ALL QINGPING RECOVERY TESTS PASSED" + ) \ No newline at end of file