github/esphome
2
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2026-10-14 10:12:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

[ufm01] Improve stale recovery, add diagnostics and clear flow action (#18899)

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: pre-commit-ci-lite[bot] <117423508+pre-commit-ci-lite[bot]@users.noreply.github.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: J. Nick Koston <nick@home-assistant.io>
This commit is contained in:
Petter Ljungqvist
2026-10-11 09:29:56 -10:00
committed by GitHub
co-authored by Cursor pre-commit-ci-lite[bot] Claude Opus 5.5 J. Nick Koston
parent c130e2808b
commit 86192d89f4
19 changed files with 1858 additions and 199 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/components/ufm01/__init__.py
+12
View File
@@ -0,0 +1,12 @@
import esphome.codegen as cg
from tests.testing_helpers import ComponentManifestOverride
def override_manifest(manifest: ComponentManifestOverride) -> None:
# The optional features are define-gated; emit them so the tests cover those paths
async def to_code_testing(config):
cg.add_define("USE_UFM01_METER_ID")
cg.add_define("USE_UFM01_SOFTWARE_VERSION")
cg.add_define("USE_UFM01_CLEAR_ACCUMULATED_FLOW_ACTION")
manifest.to_code = to_code_testing
tests/components/ufm01/benchmark.yaml
+29
View File
@@ -0,0 +1,29 @@
# Defines USE_SENSOR; the tests attach their own instances
sensor:
- platform: template
id: bench_flow
name: "Flow"
- platform: template
id: bench_temperature
name: "Temperature"
# Defines USE_TEXT_SENSOR
text_sensor:
- platform: template
id: bench_meter_id
name: "Meter ID"
- platform: template
id: bench_software_version
name: "Software version"
# One instance so __init__.py's to_code runs and emits the feature defines
uart:
- id: uart_bus
baud_rate: 2400
parity: EVEN
stop_bits: 1
port: /dev/null
ufm01:
- id: ufm01_bench
uart_id: uart_bus
tests/components/ufm01/common.h
+199 -15
View File
@@ -1,6 +1,7 @@
#pragma once
#include <array>
#include <cmath>
#include <cstdint>
#include <deque>
#include <vector>
@@ -9,17 +10,36 @@
#include "esphome/components/uart/uart_component.h"
#include "esphome/components/ufm01/ufm01.h"
#include "esphome/core/application.h"
#ifdef USE_SENSOR
#include "esphome/components/sensor/sensor.h"
#endif
#ifdef USE_TEXT_SENSOR
#include "esphome/components/text_sensor/text_sensor.h"
#endif
namespace esphome::ufm01::testing {
static constexpr uint8_t FRAME_START_BYTE_1 = 0x3C;
static constexpr uint8_t FRAME_START_BYTE_2 = 0x32;
static constexpr uint8_t PASSIVE_START_BYTE_2 = 0x64;
static constexpr uint8_t PASSIVE_START_BYTE_2_WITH_ID = 0x96;
static constexpr uint8_t FRAME_STOP_BYTE = 0x16;
static constexpr uint8_t FRAME_FLAG_INSTANT_FLOW = 0x0B;
static constexpr uint8_t FRAME_FLAG_RESERVED_SECTION = 0x0C;
static constexpr uint8_t FRAME_FLAG_TEMP = 0x0D;
static constexpr uint8_t COMMAND_ACK = 0xE5;
// Matches PASSIVE_FAIL_ESCALATE_COUNT in ufm01.cpp
static constexpr uint8_t ESCALATE_COUNT = 8;
using Command = std::array<uint8_t, 7>;
static constexpr Command ACTIVE_MODE = {0xFE, 0xFE, 0x11, 0x5C, 0x00, 0x5C, 0x16};
static constexpr Command PASSIVE_MODE = {0xFE, 0xFE, 0x11, 0x5C, 0x01, 0x5D, 0x16};
static constexpr Command CLEAR_ACCUMULATED_FLOW = {0xFE, 0xFE, 0x11, 0x5A, 0xFD, 0x57, 0x16};
static constexpr Command RESET_DEVICE = {0xFE, 0xFE, 0x11, 0x5D, 0xCB, 0x28, 0x16};
static constexpr Command READ_SENSOR_DATA_NO_ID = {0xFE, 0xFE, 0x11, 0x5B, 0x0F, 0x6A, 0x16};
static constexpr Command READ_SENSOR_DATA_WITH_ID = {0xFE, 0xFE, 0x11, 0x5B, 0xCB, 0x26, 0x16};
static constexpr Command GET_SOFTWARE_VERSION = {0xFE, 0xFE, 0x11, 0x5E, 0x62, 0xC0, 0x16};
// UART mock with a byte queue for read-side simulation.
class QueuedMockUART : public uart::UARTComponent {
@@ -31,6 +51,10 @@ class QueuedMockUART : public uart::UARTComponent {
this->rx_queue.insert(this->rx_queue.end(), data.begin(), data.end());
}
template<size_t N> void enqueue(const std::array<uint8_t, N> &data) {
this->rx_queue.insert(this->rx_queue.end(), data.begin(), data.end());
}
void enqueue(std::initializer_list<uint8_t> data) {
for (uint8_t byte : data)
this->rx_queue.push_back(byte);
@@ -71,15 +95,37 @@ class TestableUFM01 : public UFM01Component {
public:
void set_mock_uart(QueuedMockUART *uart) { this->set_uart_parent(uart); }
bool process_active_stream() { return this->process_active_stream_(); }
static uint32_t now() { return App.get_loop_component_start_time(); }
PassiveReadResult continue_passive_read() { return this->continue_passive_read_(); }
bool process_active_stream() { return this->process_active_stream_(now()); }
ReadResult continue_passive_read() { return this->continue_passive_read_(now()); }
bool consume_ack() { return this->consume_ack_(); }
void start_passive_read() { this->start_passive_read_(); }
void start_passive_read() { this->start_passive_read_(now()); }
void loop_startup() { this->loop_startup_(); }
void loop_startup() { this->loop_startup_(now()); }
void loop_active_stream() { this->loop_active_stream_(now()); }
void loop_passive_poll() { this->loop_passive_poll_(now()); }
void loop_pending_clear_action() { this->loop_pending_clear_action_(now()); }
bool clear_pending() const { return this->clear_pending_; }
bool pending_clear_sent() const { return this->pending_clear_sent_; }
bool meter_id_published() const { return this->meter_id_published_; }
void start_software_version_read() { this->start_software_version_read_(now()); }
ReadResult continue_software_version_read() { return this->continue_software_version_read_(now()); }
bool software_version_published() const { return this->software_version_published_; }
void set_last_software_version_attempt_ms(uint32_t ms) { this->last_software_version_attempt_ms_ = ms; }
OperatingMode operating_mode() const { return this->operating_mode_; }
@@ -87,20 +133,52 @@ class TestableUFM01 : public UFM01Component {
int32_t read_index() const { return this->read_index_; }
size_t passive_index() const { return this->passive_index_; }
uint8_t passive_index() const { return this->passive_index_; }
uint32_t last_valid_frame_ms() const { return this->last_valid_frame_ms_; }
uint8_t consecutive_passive_failures() const { return this->consecutive_passive_failures_; }
void set_operating_mode(OperatingMode mode) { this->operating_mode_ = mode; }
// Expects a reply without the meter ID, as start_passive_read_() picks when no meter ID sensor is set
void prepare_passive_read() {
this->passive_index_ = 0;
this->passive_start_ms_ = millis();
this->passive_expected_size_ = PASSIVE_FRAME_SIZE;
this->passive_start_byte_2_ = PASSIVE_START_BYTE_2;
this->passive_start_ms_ = now();
}
void begin_pending_passive_read() {
this->prepare_passive_read();
this->passive_read_pending_ = true;
}
void init_wait_phase() {
this->operating_mode_ = OperatingMode::STARTUP;
this->startup_phase_ = StartupPhase::WAIT;
this->startup_wait_ms_ = 60000;
this->phase_start_ms_ = millis();
this->phase_start_ms_ = now();
}
void begin_active_stream() {
this->operating_mode_ = OperatingMode::ACTIVE_STREAM;
this->last_valid_frame_ms_ = now();
this->passive_read_pending_ = false;
this->read_index_ = 0;
}
void begin_active_wait_frame() {
this->operating_mode_ = OperatingMode::STARTUP;
this->startup_phase_ = StartupPhase::ACTIVE_WAIT_FRAME;
this->phase_start_ms_ = now();
}
void begin_post_reset_wait() {
this->operating_mode_ = OperatingMode::STARTUP;
this->startup_phase_ = StartupPhase::POST_RESET_WAIT;
this->phase_start_ms_ = now();
this->software_version_published_ = false;
}
void reset_state() {
@@ -108,24 +186,59 @@ class TestableUFM01 : public UFM01Component {
this->last_valid_frame_ms_ = 0;
this->passive_index_ = 0;
this->passive_read_pending_ = false;
this->operating_mode_ = OperatingMode::STARTUP;
this->consecutive_passive_failures_ = 0;
this->meter_id_published_ = false;
this->software_version_published_ = false;
this->software_version_read_pending_ = false;
this->software_version_failures_ = 0;
this->meter_id_invalid_logged_ = false;
this->last_software_version_attempt_ms_ = 0;
this->startup_phase_ = StartupPhase::WAIT;
this->clear_pending_ = false;
this->pending_clear_sent_ = false;
}
};
template<size_t N> inline uint8_t checksum(const std::array<uint8_t, N> &data, size_t begin, size_t end) {
uint8_t sum = 0;
for (size_t i = begin; i < end; ++i)
sum += data[i];
return sum;
}
inline std::array<uint8_t, FRAME_SIZE> make_active_frame() {
std::array<uint8_t, FRAME_SIZE> frame{};
frame[0] = FRAME_START_BYTE_1;
frame[1] = FRAME_START_BYTE_2;
// Datasheet example device ID 2307140001
frame[2] = 0x01;
frame[3] = 0x00;
frame[4] = 0x14;
frame[5] = 0x07;
frame[6] = 0x23;
frame[7] = 0x01;
frame[15] = FRAME_FLAG_INSTANT_FLOW;
frame[21] = FRAME_FLAG_RESERVED_SECTION;
frame[24] = FRAME_FLAG_TEMP;
frame[31] = FRAME_STOP_BYTE;
uint8_t sum = 0;
for (size_t i = 0; i < 30; ++i)
sum += frame[i];
frame[30] = sum;
frame[30] = checksum(frame, 0, 30);
return frame;
}
inline std::array<uint8_t, SOFTWARE_VERSION_RESPONSE_SIZE> make_software_version_response() {
// Datasheet example version 23380315
std::array<uint8_t, SOFTWARE_VERSION_RESPONSE_SIZE> response{};
response[0] = COMMAND_ACK;
response[1] = 0x15;
response[2] = 0x03;
response[3] = 0x38;
response[4] = 0x23;
response[5] = checksum(response, 1, 5);
response[6] = FRAME_STOP_BYTE;
return response;
}
inline std::array<uint8_t, PASSIVE_FRAME_SIZE> make_passive_frame() {
std::array<uint8_t, PASSIVE_FRAME_SIZE> frame{};
frame[0] = FRAME_START_BYTE_1;
@@ -133,22 +246,93 @@ inline std::array<uint8_t, PASSIVE_FRAME_SIZE> make_passive_frame() {
frame[9] = FRAME_FLAG_INSTANT_FLOW;
frame[15] = FRAME_FLAG_TEMP;
frame[22] = FRAME_STOP_BYTE;
uint8_t sum = 0;
for (size_t i = 0; i < 21; ++i)
sum += frame[i];
frame[21] = sum;
frame[21] = checksum(frame, 0, 21);
return frame;
}
inline std::array<uint8_t, PASSIVE_FRAME_WITH_ID_SIZE> make_passive_with_id_frame() {
std::array<uint8_t, PASSIVE_FRAME_WITH_ID_SIZE> frame{};
frame[0] = FRAME_START_BYTE_1;
frame[1] = PASSIVE_START_BYTE_2_WITH_ID;
frame[2] = 0x01;
frame[3] = 0x00;
frame[4] = 0x14;
frame[5] = 0x07;
frame[6] = 0x23;
frame[7] = 0x01;
frame[8] = 0x0A;
// Non-measurement bytes ensure a shifted instant-flow offset is detected.
for (size_t i = 17; i < 22; ++i)
frame[i] = 0x99;
frame[22] = FRAME_FLAG_INSTANT_FLOW;
frame[23] = 0x34;
frame[24] = 0x12;
frame[27] = 0x00;
frame[31] = FRAME_FLAG_TEMP;
frame[32] = 0x50;
frame[33] = 0x23;
frame[38] = FRAME_STOP_BYTE;
frame[37] = checksum(frame, 0, 37);
return frame;
}
class UFM01Test : public ::testing::Test {
protected:
// Non-zero so a frame received at the start time still counts as a valid timestamp
static constexpr uint32_t START_MS = 1000;
void SetUp() override {
this->mock_uart_.clear_rx();
this->mock_uart_.written_data.clear();
this->ufm01_.set_mock_uart(&this->mock_uart_);
this->ufm01_.reset_state();
this->set_clock(START_MS);
}
// Publishes the time the component reads as the loop clock
void set_clock(uint32_t ms) {
this->now_ = ms;
LoopBlockingGuard clock{nullptr, nullptr, ms};
}
void advance(uint32_t ms) { this->set_clock(this->now_ + ms); }
void expect_written(const Command &command) {
EXPECT_EQ(this->mock_uart_.written_data, std::vector<uint8_t>(command.begin(), command.end()));
}
void fail_passive_poll() {
this->ufm01_.begin_pending_passive_read();
this->advance(2000);
this->ufm01_.loop_passive_poll();
}
void make_active_stream_stale() {
this->ufm01_.begin_active_stream();
this->advance(6000);
}
#ifdef USE_SENSOR
void attach_flow_and_temperature_sensors() {
this->ufm01_.set_flow_sensor(&this->flow_sensor_);
this->ufm01_.set_temperature_sensor(&this->temperature_sensor_);
this->flow_sensor_.publish_state(1.23f);
this->temperature_sensor_.publish_state(20.5f);
}
sensor::Sensor flow_sensor_;
sensor::Sensor temperature_sensor_;
#endif
void attach_diagnostic_text_sensors() {
this->ufm01_.set_meter_id_text_sensor(&this->meter_id_text_sensor_);
this->ufm01_.set_software_version_text_sensor(&this->software_version_text_sensor_);
}
text_sensor::TextSensor meter_id_text_sensor_;
text_sensor::TextSensor software_version_text_sensor_;
uint32_t now_{0};
QueuedMockUART mock_uart_;
TestableUFM01 ufm01_;
};
tests/components/ufm01/common.yaml
+14
View File
@@ -28,3 +28,17 @@ binary_sensor:
flow_rate_out_of_range:
id: flow_rate_out_of_range
name: "Flow rate out of range"
text_sensor:
- platform: ufm01
meter_id:
id: ufm01_meter_id
name: "Meter ID"
software_version:
id: ufm01_software_version
name: "Software version"
script:
- id: ufm01_clear_accumulated_flow
then:
- ufm01.clear_accumulated_flow: ufm01_component
tests/components/ufm01/test.esp32-idf.yaml
+1 -2
View File
@@ -1,4 +1,3 @@
packages:
uart_2400_even: !include ../../test_build_components/common/uart_2400_even/esp32-idf.yaml
<<: !include common.yaml
ufm01: !include common.yaml
tests/components/ufm01/test.esp8266-ard.yaml
+1 -2
View File
@@ -1,4 +1,3 @@
packages:
uart_2400_even: !include ../../test_build_components/common/uart_2400_even/esp8266-ard.yaml
<<: !include common.yaml
ufm01: !include common.yaml
tests/components/ufm01/test.rp2040-ard.yaml
+1 -2
View File
@@ -1,4 +1,3 @@
packages:
uart_2400_even: !include ../../test_build_components/common/uart_2400_even/rp2040-ard.yaml
<<: !include common.yaml
ufm01: !include common.yaml
tests/components/ufm01/ufm01_clear_action_test.cpp
+113
View File
@@ -0,0 +1,113 @@
#include "common.h"
namespace esphome::ufm01::testing {
TEST_F(UFM01Test, ClearAccumulatedFlowSendsCommand) {
this->ufm01_.set_operating_mode(OperatingMode::ACTIVE_STREAM);
this->ufm01_.request_clear_accumulated_flow();
EXPECT_TRUE(this->mock_uart_.written_data.empty());
this->ufm01_.loop_pending_clear_action();
this->expect_written(CLEAR_ACCUMULATED_FLOW);
EXPECT_TRUE(this->ufm01_.clear_pending());
}
TEST_F(UFM01Test, ClearAccumulatedFlowFinishesOnAck) {
this->ufm01_.set_operating_mode(OperatingMode::ACTIVE_STREAM);
this->ufm01_.request_clear_accumulated_flow();
this->ufm01_.loop_pending_clear_action();
this->mock_uart_.enqueue({COMMAND_ACK});
this->ufm01_.loop_pending_clear_action();
EXPECT_FALSE(this->ufm01_.clear_pending());
}
TEST_F(UFM01Test, ClearAccumulatedFlowFinishesOnTimeout) {
this->ufm01_.set_operating_mode(OperatingMode::ACTIVE_STREAM);
this->ufm01_.request_clear_accumulated_flow();
this->ufm01_.loop_pending_clear_action();
this->advance(400);
this->ufm01_.loop_pending_clear_action();
ASSERT_TRUE(this->ufm01_.clear_pending());
this->advance(200);
this->ufm01_.loop_pending_clear_action();
EXPECT_FALSE(this->ufm01_.clear_pending());
}
TEST_F(UFM01Test, DuplicateClearRequestJoinsThePendingOne) {
this->ufm01_.set_operating_mode(OperatingMode::ACTIVE_STREAM);
this->ufm01_.request_clear_accumulated_flow();
this->ufm01_.loop_pending_clear_action();
this->mock_uart_.written_data.clear();
this->ufm01_.request_clear_accumulated_flow();
this->ufm01_.loop_pending_clear_action();
EXPECT_TRUE(this->mock_uart_.written_data.empty());
this->mock_uart_.enqueue({COMMAND_ACK});
this->ufm01_.loop_pending_clear_action();
EXPECT_FALSE(this->ufm01_.clear_pending());
this->ufm01_.request_clear_accumulated_flow();
this->ufm01_.loop_pending_clear_action();
this->expect_written(CLEAR_ACCUMULATED_FLOW);
}
TEST_F(UFM01Test, ClearWaitsForPendingPassiveRead) {
this->ufm01_.set_operating_mode(OperatingMode::PASSIVE_POLL);
this->ufm01_.begin_pending_passive_read();
this->ufm01_.request_clear_accumulated_flow();
this->ufm01_.loop_pending_clear_action();
EXPECT_TRUE(this->mock_uart_.written_data.empty());
this->mock_uart_.enqueue(make_passive_frame());
this->ufm01_.loop_passive_poll();
this->ufm01_.loop_pending_clear_action();
this->expect_written(CLEAR_ACCUMULATED_FLOW);
}
TEST_F(UFM01Test, ClearWaitsForPendingSoftwareVersionRead) {
this->ufm01_.set_operating_mode(OperatingMode::ACTIVE_STREAM);
this->ufm01_.start_software_version_read();
this->mock_uart_.written_data.clear();
this->ufm01_.request_clear_accumulated_flow();
this->ufm01_.loop_pending_clear_action();
EXPECT_TRUE(this->mock_uart_.written_data.empty());
this->mock_uart_.enqueue(make_software_version_response());
this->ufm01_.loop_active_stream();
this->ufm01_.loop_pending_clear_action();
this->expect_written(CLEAR_ACCUMULATED_FLOW);
}
TEST_F(UFM01Test, ClearWaitsDuringStartup) {
this->ufm01_.set_operating_mode(OperatingMode::STARTUP);
this->ufm01_.request_clear_accumulated_flow();
this->ufm01_.loop_pending_clear_action();
EXPECT_TRUE(this->mock_uart_.written_data.empty());
EXPECT_TRUE(this->ufm01_.clear_pending());
EXPECT_FALSE(this->ufm01_.pending_clear_sent());
}
TEST_F(UFM01Test, ClearTimesOutWhileWaitingForIdle) {
this->ufm01_.set_operating_mode(OperatingMode::STARTUP);
this->ufm01_.request_clear_accumulated_flow();
this->advance(20000);
this->ufm01_.loop_pending_clear_action();
EXPECT_FALSE(this->ufm01_.clear_pending());
EXPECT_TRUE(this->mock_uart_.written_data.empty());
}
} // namespace esphome::ufm01::testing
tests/components/ufm01/ufm01_diagnostics_test.cpp
+173
View File
@@ -0,0 +1,173 @@
#include "common.h"
namespace esphome::ufm01::testing {
TEST_F(UFM01Test, ActiveFramePublishesMeterIdOnce) {
this->attach_diagnostic_text_sensors();
auto frame = make_active_frame();
this->mock_uart_.enqueue(frame);
EXPECT_TRUE(this->ufm01_.process_active_stream());
EXPECT_TRUE(this->ufm01_.meter_id_published());
EXPECT_STREQ(this->meter_id_text_sensor_.state.c_str(), "2307140001");
this->mock_uart_.enqueue(frame);
EXPECT_TRUE(this->ufm01_.process_active_stream());
EXPECT_STREQ(this->meter_id_text_sensor_.state.c_str(), "2307140001");
}
#ifdef USE_SENSOR
TEST_F(UFM01Test, PassiveFrameWithIdUsesDocumentedMeasurementOffsets) {
this->attach_diagnostic_text_sensors();
this->attach_flow_and_temperature_sensors();
this->ufm01_.start_passive_read();
this->expect_written(READ_SENSOR_DATA_WITH_ID);
this->mock_uart_.enqueue(make_passive_with_id_frame());
EXPECT_EQ(this->ufm01_.continue_passive_read(), ReadResult::READ_RESULT_SUCCESS);
EXPECT_STREQ(this->meter_id_text_sensor_.state.c_str(), "2307140001");
EXPECT_NEAR(this->flow_sensor_.get_state(), 0.01234f, 0.000001f);
EXPECT_NEAR(this->temperature_sensor_.get_state(), 23.5f, 0.001f);
}
#endif
TEST_F(UFM01Test, InvalidPassiveFrameWithIdIsRejected) {
this->attach_diagnostic_text_sensors();
this->ufm01_.start_passive_read();
auto frame = make_passive_with_id_frame();
frame[37] ^= 0xFF;
this->mock_uart_.enqueue(frame);
EXPECT_EQ(this->ufm01_.continue_passive_read(), ReadResult::READ_RESULT_FAILURE);
EXPECT_FALSE(this->ufm01_.meter_id_published());
}
TEST_F(UFM01Test, SoftwareVersionResponseDecoded) {
this->attach_diagnostic_text_sensors();
this->ufm01_.start_software_version_read();
this->mock_uart_.enqueue(make_software_version_response());
EXPECT_EQ(this->ufm01_.continue_software_version_read(), ReadResult::READ_RESULT_SUCCESS);
EXPECT_TRUE(this->ufm01_.software_version_published());
EXPECT_STREQ(this->software_version_text_sensor_.state.c_str(), "23380315");
}
TEST_F(UFM01Test, SoftwareVersionReadSkipsLeftoverStreamBytes) {
this->attach_diagnostic_text_sensors();
this->ufm01_.start_software_version_read();
// Tail of an active frame that was still on the wire when the command was sent
auto frame = make_active_frame();
this->mock_uart_.enqueue(std::vector<uint8_t>(frame.begin() + 20, frame.end()));
this->mock_uart_.enqueue(make_software_version_response());
EXPECT_EQ(this->ufm01_.continue_software_version_read(), ReadResult::READ_RESULT_SUCCESS);
EXPECT_STREQ(this->software_version_text_sensor_.state.c_str(), "23380315");
}
TEST_F(UFM01Test, InvalidSoftwareVersionBcdFails) {
this->attach_diagnostic_text_sensors();
this->ufm01_.start_software_version_read();
auto response = make_software_version_response();
response[2] = 0xFA;
response[5] = checksum(response, 1, 5);
this->mock_uart_.enqueue(response);
EXPECT_EQ(this->ufm01_.continue_software_version_read(), ReadResult::READ_RESULT_FAILURE);
EXPECT_FALSE(this->ufm01_.software_version_published());
}
TEST_F(UFM01Test, SoftwareVersionReadTimeoutFails) {
this->attach_diagnostic_text_sensors();
this->ufm01_.start_software_version_read();
this->advance(1900);
EXPECT_EQ(this->ufm01_.continue_software_version_read(), ReadResult::READ_RESULT_PENDING);
this->advance(100);
EXPECT_EQ(this->ufm01_.continue_software_version_read(), ReadResult::READ_RESULT_FAILURE);
EXPECT_FALSE(this->ufm01_.software_version_published());
}
TEST_F(UFM01Test, StartupReadsSoftwareVersionBeforeActiveMode) {
this->attach_diagnostic_text_sensors();
this->ufm01_.begin_post_reset_wait();
this->advance(3000);
this->ufm01_.loop_startup();
EXPECT_EQ(this->ufm01_.startup_phase(), StartupPhase::SOFTWARE_VERSION_WAIT_REPLY);
this->expect_written(GET_SOFTWARE_VERSION);
this->mock_uart_.enqueue(make_software_version_response());
this->mock_uart_.written_data.clear();
this->ufm01_.loop_startup();
EXPECT_EQ(this->ufm01_.startup_phase(), StartupPhase::ACTIVE_WAIT_FRAME);
this->expect_written(ACTIVE_MODE);
EXPECT_TRUE(this->ufm01_.software_version_published());
EXPECT_STREQ(this->software_version_text_sensor_.state.c_str(), "23380315");
}
TEST_F(UFM01Test, AlreadyActiveStartupRequestsSoftwareVersion) {
this->attach_diagnostic_text_sensors();
this->mock_uart_.enqueue(make_active_frame());
this->ufm01_.init_wait_phase();
this->ufm01_.loop_startup();
EXPECT_EQ(this->ufm01_.operating_mode(), OperatingMode::ACTIVE_STREAM);
EXPECT_TRUE(this->ufm01_.meter_id_published());
EXPECT_TRUE(this->mock_uart_.written_data.empty());
this->ufm01_.loop_active_stream();
this->expect_written(GET_SOFTWARE_VERSION);
}
TEST_F(UFM01Test, ActiveStreamRetriesSoftwareVersionRead) {
this->attach_diagnostic_text_sensors();
this->ufm01_.set_operating_mode(OperatingMode::ACTIVE_STREAM);
this->ufm01_.set_last_software_version_attempt_ms(this->now_);
this->advance(29000);
this->ufm01_.loop_active_stream();
EXPECT_TRUE(this->mock_uart_.written_data.empty());
this->advance(2000);
this->ufm01_.loop_active_stream();
this->expect_written(GET_SOFTWARE_VERSION);
}
TEST_F(UFM01Test, ActiveStreamGivesUpSoftwareVersionAfterRepeatedFailures) {
this->attach_diagnostic_text_sensors();
this->ufm01_.set_operating_mode(OperatingMode::ACTIVE_STREAM);
for (int attempt = 0; attempt < 3; ++attempt) {
this->ufm01_.loop_active_stream();
this->expect_written(GET_SOFTWARE_VERSION);
this->mock_uart_.written_data.clear();
this->advance(2000);
this->ufm01_.loop_active_stream();
this->advance(30000);
}
this->ufm01_.loop_active_stream();
EXPECT_TRUE(this->mock_uart_.written_data.empty());
EXPECT_FALSE(this->ufm01_.software_version_published());
}
TEST_F(UFM01Test, InvalidMeterIdLeavesSensorUnknown) {
this->attach_diagnostic_text_sensors();
auto frame = make_active_frame();
frame[2] = 0xAB;
frame[30] = checksum(frame, 0, 30);
this->mock_uart_.enqueue(frame);
EXPECT_TRUE(this->ufm01_.process_active_stream());
EXPECT_FALSE(this->ufm01_.meter_id_published());
EXPECT_FALSE(this->meter_id_text_sensor_.has_state());
}
} // namespace esphome::ufm01::testing
tests/components/ufm01/ufm01_frame_test.cpp
+13 -13
View File
@@ -4,7 +4,7 @@ namespace esphome::ufm01::testing {
TEST_F(UFM01Test, ValidActiveFrameAccepted) {
auto frame = make_active_frame();
this->mock_uart_.enqueue(std::vector<uint8_t>(frame.begin(), frame.end()));
this->mock_uart_.enqueue(frame);
EXPECT_TRUE(this->ufm01_.process_active_stream());
EXPECT_EQ(this->ufm01_.read_index(), 0);
@@ -14,7 +14,7 @@ TEST_F(UFM01Test, ValidActiveFrameAccepted) {
TEST_F(UFM01Test, GarbagePrefixThenValidActiveFrame) {
this->mock_uart_.enqueue({0x00, 0xFF, 0xAA});
auto frame = make_active_frame();
this->mock_uart_.enqueue(std::vector<uint8_t>(frame.begin(), frame.end()));
this->mock_uart_.enqueue(frame);
EXPECT_TRUE(this->ufm01_.process_active_stream());
EXPECT_EQ(this->ufm01_.read_index(), 0);
@@ -23,7 +23,7 @@ TEST_F(UFM01Test, GarbagePrefixThenValidActiveFrame) {
TEST_F(UFM01Test, InvalidActiveFrameChecksumRejected) {
auto frame = make_active_frame();
frame[30] ^= 0xFF;
this->mock_uart_.enqueue(std::vector<uint8_t>(frame.begin(), frame.end()));
this->mock_uart_.enqueue(frame);
EXPECT_FALSE(this->ufm01_.process_active_stream());
EXPECT_EQ(this->ufm01_.read_index(), 0);
@@ -32,10 +32,10 @@ TEST_F(UFM01Test, InvalidActiveFrameChecksumRejected) {
TEST_F(UFM01Test, ValidPassiveFrameReadSuccess) {
auto frame = make_passive_frame();
this->mock_uart_.enqueue(std::vector<uint8_t>(frame.begin(), frame.end()));
this->mock_uart_.enqueue(frame);
this->ufm01_.prepare_passive_read();
EXPECT_EQ(this->ufm01_.continue_passive_read(), PassiveReadResult::PASSIVE_READ_RESULT_SUCCESS);
EXPECT_EQ(this->ufm01_.continue_passive_read(), ReadResult::READ_RESULT_SUCCESS);
EXPECT_EQ(this->ufm01_.passive_index(), PASSIVE_FRAME_SIZE);
EXPECT_NE(this->ufm01_.last_valid_frame_ms(), 0u);
}
@@ -43,29 +43,29 @@ TEST_F(UFM01Test, ValidPassiveFrameReadSuccess) {
TEST_F(UFM01Test, InvalidPassiveChecksumFails) {
auto frame = make_passive_frame();
frame[21] ^= 0xFF;
this->mock_uart_.enqueue(std::vector<uint8_t>(frame.begin(), frame.end()));
this->mock_uart_.enqueue(frame);
this->ufm01_.prepare_passive_read();
EXPECT_EQ(this->ufm01_.continue_passive_read(), PassiveReadResult::PASSIVE_READ_RESULT_FAILURE);
EXPECT_EQ(this->ufm01_.continue_passive_read(), ReadResult::READ_RESULT_FAILURE);
EXPECT_EQ(this->ufm01_.last_valid_frame_ms(), 0u);
}
TEST_F(UFM01Test, PassiveReadResyncsAfterGarbagePrefix) {
auto frame = make_passive_frame();
this->mock_uart_.enqueue({0x00, 0x01, 0x02});
this->mock_uart_.enqueue(std::vector<uint8_t>(frame.begin(), frame.end()));
this->mock_uart_.enqueue(frame);
this->ufm01_.prepare_passive_read();
EXPECT_EQ(this->ufm01_.continue_passive_read(), PassiveReadResult::PASSIVE_READ_RESULT_SUCCESS);
EXPECT_EQ(this->ufm01_.continue_passive_read(), ReadResult::READ_RESULT_SUCCESS);
}
TEST_F(UFM01Test, PassiveReadResyncsOnSecondStartByte) {
auto frame = make_passive_frame();
this->mock_uart_.enqueue({FRAME_START_BYTE_1, 0x99});
this->mock_uart_.enqueue(std::vector<uint8_t>(frame.begin(), frame.end()));
this->mock_uart_.enqueue(frame);
this->ufm01_.prepare_passive_read();
EXPECT_EQ(this->ufm01_.continue_passive_read(), PassiveReadResult::PASSIVE_READ_RESULT_SUCCESS);
EXPECT_EQ(this->ufm01_.continue_passive_read(), ReadResult::READ_RESULT_SUCCESS);
}
TEST_F(UFM01Test, PassiveReadPendingWhenPartial) {
@@ -73,11 +73,11 @@ TEST_F(UFM01Test, PassiveReadPendingWhenPartial) {
this->mock_uart_.enqueue(std::vector<uint8_t>(frame.begin(), frame.begin() + 10));
this->ufm01_.prepare_passive_read();
EXPECT_EQ(this->ufm01_.continue_passive_read(), PassiveReadResult::PASSIVE_READ_RESULT_PENDING);
EXPECT_EQ(this->ufm01_.continue_passive_read(), ReadResult::READ_RESULT_PENDING);
EXPECT_LT(this->ufm01_.passive_index(), PASSIVE_FRAME_SIZE);
this->mock_uart_.enqueue(std::vector<uint8_t>(frame.begin() + 10, frame.end()));
EXPECT_EQ(this->ufm01_.continue_passive_read(), PassiveReadResult::PASSIVE_READ_RESULT_SUCCESS);
EXPECT_EQ(this->ufm01_.continue_passive_read(), ReadResult::READ_RESULT_SUCCESS);
}
} // namespace esphome::ufm01::testing
tests/components/ufm01/ufm01_startup_test.cpp
+112 -15
View File
@@ -1,14 +1,7 @@
#include "common.h"
#include "esphome/core/component.h"
namespace esphome::ufm01::testing {
TEST(UFM01SetupPriority, IsLate) {
TestableUFM01 ufm01;
EXPECT_EQ(ufm01.get_setup_priority(), setup_priority::LATE);
}
TEST_F(UFM01Test, ConsumeAckFindsByteAmongGarbage) {
this->mock_uart_.enqueue({0x00, 0x01, COMMAND_ACK, 0x02});
@@ -19,8 +12,7 @@ TEST_F(UFM01Test, ConsumeAckFindsByteAmongGarbage) {
TEST_F(UFM01Test, ConsumeAckReturnsFalseWhenEmpty) { EXPECT_FALSE(this->ufm01_.consume_ack()); }
TEST_F(UFM01Test, StartupWaitDetectsActiveStream) {
auto frame = make_active_frame();
this->mock_uart_.enqueue(std::vector<uint8_t>(frame.begin(), frame.end()));
this->mock_uart_.enqueue(make_active_frame());
this->ufm01_.init_wait_phase();
this->ufm01_.loop_startup();
@@ -32,12 +24,117 @@ TEST_F(UFM01Test, StartupWaitDetectsActiveStream) {
TEST_F(UFM01Test, StartPassiveReadSendsCommand) {
this->ufm01_.start_passive_read();
ASSERT_EQ(this->mock_uart_.written_data.size(), 7u);
EXPECT_EQ(this->mock_uart_.written_data[0], 0xFE);
EXPECT_EQ(this->mock_uart_.written_data[1], 0xFE);
EXPECT_EQ(this->mock_uart_.written_data[2], 0x11);
EXPECT_EQ(this->mock_uart_.written_data[3], 0x5B);
EXPECT_EQ(this->mock_uart_.written_data[6], FRAME_STOP_BYTE);
this->expect_written(READ_SENSOR_DATA_NO_ID);
}
TEST_F(UFM01Test, StartupActiveWaitTimeoutSendsSetPassiveMode) {
this->ufm01_.begin_active_wait_frame();
this->advance(4000);
this->ufm01_.loop_startup();
EXPECT_EQ(this->ufm01_.startup_phase(), StartupPhase::SET_PASSIVE_WAIT_ACK);
this->expect_written(PASSIVE_MODE);
}
TEST_F(UFM01Test, StartupSetPassiveWaitAckStartsPassiveRead) {
this->ufm01_.begin_active_wait_frame();
this->advance(4000);
this->ufm01_.loop_startup();
this->mock_uart_.enqueue({COMMAND_ACK});
this->mock_uart_.written_data.clear();
this->ufm01_.loop_startup();
EXPECT_EQ(this->ufm01_.startup_phase(), StartupPhase::PASSIVE_WAIT_REPLY);
this->expect_written(READ_SENSOR_DATA_NO_ID);
}
TEST_F(UFM01Test, StaleActiveStreamSendsSetPassiveMode) {
// Leftover stream noise should be flushed before SET_PASSIVE_MODE
this->mock_uart_.enqueue({0x3C, 0x32, 0x00});
this->make_active_stream_stale();
this->ufm01_.loop_active_stream();
EXPECT_EQ(this->ufm01_.operating_mode(), OperatingMode::PASSIVE_POLL);
EXPECT_EQ(this->ufm01_.read_index(), 0);
this->expect_written(PASSIVE_MODE);
EXPECT_EQ(this->mock_uart_.available(), 0u); // RX flushed
}
TEST_F(UFM01Test, StalePassivePollWaitsOneIntervalAndDiscardsLateAck) {
this->make_active_stream_stale();
this->ufm01_.loop_active_stream();
ASSERT_EQ(this->ufm01_.operating_mode(), OperatingMode::PASSIVE_POLL);
this->mock_uart_.written_data.clear();
this->mock_uart_.enqueue({COMMAND_ACK});
this->advance(500);
this->ufm01_.loop_passive_poll();
EXPECT_TRUE(this->mock_uart_.written_data.empty());
EXPECT_EQ(this->ufm01_.operating_mode(), OperatingMode::PASSIVE_POLL);
this->advance(500);
this->ufm01_.loop_passive_poll();
this->expect_written(READ_SENSOR_DATA_NO_ID);
EXPECT_EQ(this->mock_uart_.available(), 0u);
}
#ifdef USE_SENSOR
TEST_F(UFM01Test, StaleActiveStreamPublishesNanToFlowAndTemperature) {
this->attach_flow_and_temperature_sensors();
this->make_active_stream_stale();
this->ufm01_.loop_active_stream();
EXPECT_TRUE(std::isnan(this->flow_sensor_.get_state()));
EXPECT_TRUE(std::isnan(this->temperature_sensor_.get_state()));
}
TEST_F(UFM01Test, RepeatedPassiveFailuresPublishNanBeforeReset) {
this->attach_flow_and_temperature_sensors();
this->ufm01_.set_operating_mode(OperatingMode::PASSIVE_POLL);
for (uint8_t i = 0; i < ESCALATE_COUNT; ++i)
this->fail_passive_poll();
EXPECT_TRUE(std::isnan(this->flow_sensor_.get_state()));
EXPECT_TRUE(std::isnan(this->temperature_sensor_.get_state()));
}
#endif
TEST_F(UFM01Test, PassivePollSuccessClearsFailureCount) {
this->ufm01_.set_operating_mode(OperatingMode::PASSIVE_POLL);
for (int i = 0; i < 3; ++i)
this->fail_passive_poll();
ASSERT_EQ(this->ufm01_.consecutive_passive_failures(), 3u);
this->mock_uart_.enqueue(make_passive_frame());
this->ufm01_.begin_pending_passive_read();
this->ufm01_.loop_passive_poll();
EXPECT_EQ(this->ufm01_.consecutive_passive_failures(), 0u);
EXPECT_EQ(this->ufm01_.operating_mode(), OperatingMode::PASSIVE_POLL);
}
TEST_F(UFM01Test, RepeatedPassiveFailuresEscalateToReset) {
this->ufm01_.set_operating_mode(OperatingMode::PASSIVE_POLL);
for (uint8_t i = 0; i < ESCALATE_COUNT - 1; ++i) {
this->fail_passive_poll();
ASSERT_EQ(this->ufm01_.operating_mode(), OperatingMode::PASSIVE_POLL) << "i=" << static_cast<int>(i);
}
EXPECT_EQ(this->ufm01_.consecutive_passive_failures(), ESCALATE_COUNT - 1);
this->mock_uart_.written_data.clear();
this->fail_passive_poll();
EXPECT_EQ(this->ufm01_.operating_mode(), OperatingMode::STARTUP);
EXPECT_EQ(this->ufm01_.startup_phase(), StartupPhase::RESET_WAIT_ACK);
EXPECT_EQ(this->ufm01_.consecutive_passive_failures(), 0u);
this->expect_written(RESET_DEVICE);
}
} // namespace esphome::ufm01::testing
tests/components/ufm01/validate.host.yaml
+16
View File
@@ -0,0 +1,16 @@
esphome:
name: ufm01-validate
host:
logger:
packages:
ufm01: !include common.yaml
uart:
id: uart_bus
baud_rate: 2400
parity: EVEN
stop_bits: 1
port: /dev/null
tests/integration/fixtures/uart_mock_ufm01.yaml
+153
View File
@@ -0,0 +1,153 @@
esphome:
name: uart-mock-ufm01
host:
api:
batch_delay: 0ms # Disable batching to receive all state updates
actions:
- action: clear_accumulated_flow
then:
- ufm01.clear_accumulated_flow: ufm01_dev
- lambda: |-
id(clear_count) += 1;
id(clear_completions).publish_state(id(clear_count));
logger:
level: DEBUG
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
# Dummy uart entry to satisfy ufm01's DEPENDENCIES = ["uart"]
# The actual UART bus used is the uart_mock component below
uart:
baud_rate: 2400
port: /dev/null
globals:
# The simulated meter streams active frames only after SET_ACTIVE_MODE
- id: streaming
type: bool
initial_value: "false"
# Set by CLEAR_ACCUMULATED_FLOW; later frames report zero accumulated flow
- id: cleared
type: bool
initial_value: "false"
- id: clear_count
type: int
initial_value: "0"
# Simulated UFM-01: every command is one 7 byte write
uart_mock:
id: mock_uart
baud_rate: 2400
parity: EVEN
on_tx:
- then:
- lambda: |-
if (id(silent).state || data.size() != 7) {
return;
}
static const std::vector<uint8_t> ACK = {0xE5};
switch (data[3]) {
case 0x5D: // RESET_DEVICE
id(streaming) = false;
id(mock_uart).inject_to_rx_buffer(ACK);
break;
case 0x5C: // SET_ACTIVE_MODE (00) or SET_PASSIVE_MODE (01)
id(streaming) = data[4] == 0x00;
id(mock_uart).inject_to_rx_buffer(ACK);
break;
case 0x5A: // CLEAR_ACCUMULATED_FLOW
id(cleared) = true;
id(mock_uart).inject_to_rx_buffer(ACK);
break;
case 0x5B: // READ_SENSOR_DATA without meter ID
if (data[4] != 0x0F) {
break;
}
id(mock_uart).inject_to_rx_buffer(std::vector<uint8_t>{
0x3C, 0x64, // start
0x0A, 0x78, 0x56, 0x02, 0x00, 0x00, 0x00, // accumulated flow 25.678 L
0x0B, 0x00, 0x67, 0x05, 0x00, 0x00, // flow 0.567 m3/h
0x0D, 0x25, 0x19, 0x00, // temperature 19.25 C
0x00, 0x00, // ST1, ST2
0x3C, 0x16, // checksum, stop
});
break;
case 0x5E: // GET_SOFTWARE_VERSION: ACK, version 23380315 (BCD, low byte first), checksum, stop
id(mock_uart).inject_to_rx_buffer(std::vector<uint8_t>{0xE5, 0x15, 0x03, 0x38, 0x23, 0x73, 0x16});
break;
}
interval:
- interval: 1s
then:
- if:
condition:
lambda: "return id(streaming) && !id(silent).state;"
then:
- lambda: |-
// Meter ID 2307140001 (BCD, low byte first), flow 1.2345 m3/h, temperature 23.5 C
std::vector<uint8_t> frame{
0x3C, 0x32, // start
0x01, 0x00, 0x14, 0x07, 0x23, 0x01, // meter ID, reserved
0x0A, 0x45, 0x23, 0x01, 0x00, 0x00, 0x00, // accumulated flow 12.345 L
0x0B, 0x50, 0x34, 0x12, 0x00, 0x00, // flow
0x0C, 0x00, 0x00, // reserved section
0x0D, 0x50, 0x23, 0x00, // temperature
0x00, 0x00, // ST1, ST2
0x4E, 0x16, // checksum, stop
};
if (id(cleared)) {
// Accumulated flow 0 m3: unit flag 0x1A, zero digits, checksum follows
frame[8] = 0x1A;
std::fill(frame.begin() + 9, frame.begin() + 15, 0x00);
frame[30] = 0xF5;
}
id(mock_uart).inject_to_rx_buffer(frame);
ufm01:
id: ufm01_dev
uart_id: mock_uart
sensor:
- platform: ufm01
accumulated_flow:
name: "Accumulated Flow"
flow:
name: "Flow"
temperature:
name: "Temperature"
- platform: template
id: clear_completions
name: "Clear Completions"
update_interval: never
text_sensor:
- platform: ufm01
meter_id:
name: "Meter ID"
software_version:
name: "Software Version"
switch:
# The meter stops streaming and answers nothing
- platform: template
id: silent
name: "Silent"
optimistic: true
turn_on_action:
- lambda: "id(streaming) = false;"
button:
# The stream stops while the meter still answers commands
- platform: template
name: "Pause Stream"
on_press:
- lambda: "id(streaming) = false;"
- platform: template
name: "Resume Stream"
on_press:
- lambda: "id(streaming) = true;"
tests/integration/test_uart_mock_ufm01.py
+399
View File
@@ -0,0 +1,399 @@
"""Integration tests for the UFM-01 flow meter with a simulated meter on the mock UART.
All tests share uart_mock_ufm01.yaml, whose on_tx lambda answers like a UFM-01: it ACKs commands,
answers sensor data and software version reads, and streams active frames each second after
SET_ACTIVE_MODE. The "Silent" switch stops it answering, "Pause Stream" stops the stream while it still
answers, and "Resume Stream" starts the stream again.
test_uart_mock_ufm01_startup:
Reset, software version read and active mode go out in order, the active frames publish flow,
temperature and accumulated flow, and the meter ID and software version publish once.
test_uart_mock_ufm01_stale_recovery:
A stopped stream marks flow and temperature unknown after 5 s and switches to passive polling,
whose answers publish; a resumed stream switches back to the active stream and polling stops.
test_uart_mock_ufm01_escalation:
Eight unanswered passive polls send a reset and run the startup sequence again; values publish
once the meter answers, and the diagnostics are not published again.
test_uart_mock_ufm01_clear_action:
The clear command is sent once the UART is idle while the automation continues at once; the
meter's ACK, or its absence, is only logged, and a second request joins the pending one.
"""
from __future__ import annotations
import asyncio
from collections.abc import Awaitable, Callable
import math
import re
from aioesphomeapi import (
APIClient,
ButtonInfo,
EntityInfo,
EntityState,
SensorState,
SwitchInfo,
TextSensorState,
UserService,
)
import pytest
from .state_utils import InitialStateHelper, StateWaiter
from .types import APIClientConnectedFactory, RunCompiledFunction
# Commands as the uart mock logs them: FE FE 11, command, argument, checksum, 16
RESET_DEVICE = "FE:FE:11:5D:CB:28:16"
SET_ACTIVE_MODE = "FE:FE:11:5C:00:5C:16"
SET_PASSIVE_MODE = "FE:FE:11:5C:01:5D:16"
READ_SENSOR_DATA = "FE:FE:11:5B:0F:6A:16"
GET_SOFTWARE_VERSION = "FE:FE:11:5E:62:C0:16"
CLEAR_ACCUMULATED_FLOW = "FE:FE:11:5A:FD:57:16"
# Values the fixture's frames encode
ACTIVE_FLOW = 1.2345
ACTIVE_TEMPERATURE = 23.5
ACTIVE_ACCUMULATED = 12.345
PASSIVE_FLOW = 0.567
PASSIVE_TEMPERATURE = 19.25
PASSIVE_ACCUMULATED = 25.678
METER_ID = "2307140001"
SOFTWARE_VERSION = "23380315"
ACTIVE_STARTED = "UFM-01 active stream started"
ACTIVE_RESUMED = "UFM-01 active stream resumed"
STALE = "Active stream stale, switching to passive polling"
ESCALATED = "Passive poll failed repeatedly, re-initializing UFM-01"
CLEAR_ACKED = "Clear accumulated flow acknowledged"
CLEAR_NOT_ACKED = "Clear accumulated flow not acknowledged"
# Startup wait, reset ACK, post-reset wait, then the first frame within the 1 s stream interval
STARTUP_TIMEOUT = 8.0
ACTIVE_STALE_S = 5.0
ANSI_ESCAPE = re.compile(r"\x1b\[[0-9;]*m")
class _Log:
"""Record TX frames and ufm01 log lines with their arrival time, and wait for lines."""
def __init__(self) -> None:
self.events: list[tuple[float, str]] = []
self._awaited: list[tuple[str, asyncio.Future[float]]] = []
def on_line(self, line: str) -> None:
line = ANSI_ESCAPE.sub("", line)
if "uart_mock" in line and "TX " in line and "bytes: " in line:
text = line.split("bytes: ", 1)[1].strip()
elif "[ufm01" in line:
text = line
else:
return
now = asyncio.get_running_loop().time()
self.events.append((now, text))
for wanted, future in self._awaited:
if wanted in text and not future.done():
future.set_result(now)
def next(self, text: str) -> asyncio.Future[float]:
"""Return a future for the time of the next event containing ``text``."""
future: asyncio.Future[float] = asyncio.get_running_loop().create_future()
self._awaited.append((text, future))
return future
@property
def tx(self) -> list[str]:
return [text for _, text in self.events if text.startswith("FE:")]
def index(self, text: str, start: int = 0) -> int:
return next(
i
for i, (_, event) in enumerate(self.events)
if i >= start and text in event
)
def tx_between(self, start: int, end: int) -> list[str]:
return [text for _, text in self.events[start:end] if text.startswith("FE:")]
class _Meter:
"""Entities and state history of the running fixture."""
def __init__(
self, client: APIClient, entities: list[EntityInfo], services: list[UserService]
) -> None:
self.client = client
self.services = services
self.by_id = {entity.object_id: entity for entity in entities}
self.waiter = StateWaiter()
self.states: dict[str, list[float | str]] = {name: [] for name in self.by_id}
self._names = {entity.key: entity.object_id for entity in entities}
self.helper = InitialStateHelper(entities)
def on_state(self, state: EntityState) -> None:
if (
isinstance(state, (SensorState, TextSensorState))
and not state.missing_state
):
self.states[self._names[state.key]].append(state.state)
self.waiter.on_state(state)
def expect(
self, name: str, check: Callable[[float | str], bool], timeout: float
) -> Awaitable[EntityState]:
key = self.by_id[name].key
return self.waiter.expect(
lambda s: s.key == key and not s.missing_state and check(s.state),
timeout=timeout,
label=name,
)
def press(self, name: str) -> None:
entity = self.by_id[name]
assert isinstance(entity, ButtonInfo)
self.client.button_command(entity.key)
def set_silent(self, silent: bool) -> None:
entity = self.by_id["silent"]
assert isinstance(entity, SwitchInfo)
self.client.switch_command(entity.key, silent)
def all_values(self, name: str) -> list[float | str]:
"""Return every value ``name`` held, including one already set at connect time."""
initial = self.helper.initial_states[self.by_id[name].key]
return ([] if initial.missing_state else [initial.state]) + self.states[name]
async def _connect(client: APIClient) -> _Meter:
entities, services = await client.list_entities_services()
meter = _Meter(client, entities, services)
client.subscribe_states(meter.helper.on_state_wrapper(meter.on_state))
try:
await meter.helper.wait_for_initial_states()
except TimeoutError:
pytest.fail("Timeout waiting for initial states")
return meter
def _is_nan(value: float | str) -> bool:
return isinstance(value, float) and math.isnan(value)
def _approx(expected: float) -> Callable[[float | str], bool]:
return lambda value: value == pytest.approx(expected, abs=1e-4)
def _active_values(meter: _Meter, timeout: float) -> list[Awaitable[EntityState]]:
"""Arm waits for the next flow, temperature and accumulated flow from an active frame."""
return [
meter.expect("flow", _approx(ACTIVE_FLOW), timeout),
meter.expect("temperature", _approx(ACTIVE_TEMPERATURE), timeout),
meter.expect("accumulated_flow", _approx(ACTIVE_ACCUMULATED), timeout),
]
async def _gather(meter: _Meter, waits: list[Awaitable[EntityState]]) -> None:
try:
await asyncio.gather(*waits)
except TimeoutError as err:
pytest.fail(f"{err}; states: {meter.states}")
async def _wait_for(log: _Log, future: asyncio.Future[float], timeout: float) -> float:
try:
return await asyncio.wait_for(future, timeout)
except TimeoutError:
pytest.fail(f"Timed out; log: {log.events}")
@pytest.mark.shared_yaml("uart_mock_ufm01")
@pytest.mark.asyncio
async def test_uart_mock_ufm01_startup(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Startup reaches the active stream, publishes the decoded values and the diagnostics once."""
log = _Log()
started = log.next(ACTIVE_STARTED)
async with (
run_compiled(yaml_config, line_callback=log.on_line),
api_client_connected() as client,
):
meter = await _connect(client)
values = _active_values(meter, STARTUP_TIMEOUT)
await _wait_for(log, started, STARTUP_TIMEOUT)
await _gather(meter, values)
# Two more stream frames
for _ in range(2):
await meter.expect("flow", _approx(ACTIVE_FLOW), 2.0)
assert log.tx == [RESET_DEVICE, GET_SOFTWARE_VERSION, SET_ACTIVE_MODE]
assert meter.all_values("meter_id") == [METER_ID]
assert meter.all_values("software_version") == [SOFTWARE_VERSION]
@pytest.mark.shared_yaml("uart_mock_ufm01")
@pytest.mark.asyncio
async def test_uart_mock_ufm01_stale_recovery(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""A stopped stream falls back to passive polling and a resumed stream ends it."""
log = _Log()
started = log.next(ACTIVE_STARTED)
async with (
run_compiled(yaml_config, line_callback=log.on_line),
api_client_connected() as client,
):
meter = await _connect(client)
await _wait_for(log, started, STARTUP_TIMEOUT)
await meter.expect("flow", _approx(ACTIVE_FLOW), 2.0)
stale = log.next(STALE)
flow_unknown = meter.expect("flow", _is_nan, ACTIVE_STALE_S + 2)
temperature_unknown = meter.expect("temperature", _is_nan, ACTIVE_STALE_S + 2)
polled = [
meter.expect("flow", _approx(PASSIVE_FLOW), ACTIVE_STALE_S + 4),
meter.expect(
"temperature", _approx(PASSIVE_TEMPERATURE), ACTIVE_STALE_S + 4
),
meter.expect(
"accumulated_flow", _approx(PASSIVE_ACCUMULATED), ACTIVE_STALE_S + 4
),
]
meter.press("pause_stream")
paused_at = asyncio.get_running_loop().time()
stale_at = await _wait_for(log, stale, ACTIVE_STALE_S + 2)
# The last frame came up to 1 s before the pause
assert stale_at - paused_at >= ACTIVE_STALE_S - 1.5
await _gather(meter, [flow_unknown, temperature_unknown])
await _gather(meter, polled)
stale_index = log.index(STALE)
assert log.tx_between(stale_index, len(log.events))[:2] == [
SET_PASSIVE_MODE,
READ_SENSOR_DATA,
]
# Unknown first, then the passive answers
assert _is_nan(meter.states["flow"][-2])
assert _is_nan(meter.states["temperature"][-2])
resumed = log.next(ACTIVE_RESUMED)
meter.press("resume_stream")
values = _active_values(meter, 3.0)
await _wait_for(log, resumed, 3.0)
await _gather(meter, values)
# Two more stream frames without any poll in between
for _ in range(2):
await meter.expect("flow", _approx(ACTIVE_FLOW), 2.0)
assert READ_SENSOR_DATA not in log.tx_between(
log.index(ACTIVE_RESUMED), len(log.events)
)
@pytest.mark.shared_yaml("uart_mock_ufm01")
@pytest.mark.asyncio
async def test_uart_mock_ufm01_escalation(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Eight unanswered passive polls reset the meter and rerun the startup sequence."""
log = _Log()
started = log.next(ACTIVE_STARTED)
async with (
run_compiled(yaml_config, line_callback=log.on_line),
api_client_connected() as client,
):
meter = await _connect(client)
await _wait_for(log, started, STARTUP_TIMEOUT)
await meter.expect("flow", _approx(ACTIVE_FLOW), 2.0)
escalated = log.next(ESCALATED)
meter.set_silent(True)
# Stale after 5 s, then eight polls that each time out after 1 s
flow_count = len(meter.states["flow"])
await _wait_for(log, escalated, ACTIVE_STALE_S + 8 + 4)
restarted = log.next(ACTIVE_STARTED)
values = _active_values(meter, STARTUP_TIMEOUT)
meter.set_silent(False)
await _wait_for(log, restarted, STARTUP_TIMEOUT)
await _gather(meter, values)
stale_index = log.index(STALE)
escalated_index = log.index(ESCALATED, stale_index)
assert (
log.tx_between(stale_index, escalated_index)
== [SET_PASSIVE_MODE] + [READ_SENSOR_DATA] * 8
)
# A reset (resent once if the meter was still silent), then active mode; the software
# version is already known, so it is not read again
after = log.tx_between(
escalated_index, log.index(ACTIVE_STARTED, escalated_index)
)
assert after in (
[RESET_DEVICE, SET_ACTIVE_MODE],
[RESET_DEVICE, RESET_DEVICE, SET_ACTIVE_MODE],
), after
# Unknown when the stream went stale and again on the re-init, nothing in between;
# a frame already in flight when the meter went silent may come first
flow = meter.states["flow"][flow_count:]
first_unknown = next(i for i, value in enumerate(flow) if _is_nan(value))
assert _is_nan(flow[first_unknown + 1]), flow
assert flow[first_unknown + 2] == pytest.approx(ACTIVE_FLOW, abs=1e-4), flow
assert meter.all_values("meter_id") == [METER_ID]
assert meter.all_values("software_version") == [SOFTWARE_VERSION]
@pytest.mark.shared_yaml("uart_mock_ufm01")
@pytest.mark.asyncio
async def test_uart_mock_ufm01_clear_action(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""The clear command is sent once the UART is idle; the automation continues at once."""
log = _Log()
started = log.next(ACTIVE_STARTED)
acked = log.next(CLEAR_ACKED)
async with (
run_compiled(yaml_config, line_callback=log.on_line),
api_client_connected() as client,
):
meter = await _connect(client)
# A second entity listing would re-send the initial states and duplicate the first completion
(clear,) = meter.services
# Normally still inside the startup wait; a slow runner may already be streaming
queued_during_startup = not started.done()
completed = meter.expect("clear_completions", _approx(1), STARTUP_TIMEOUT)
zeroed = meter.expect("accumulated_flow", _approx(0.0), STARTUP_TIMEOUT + 2)
await client.execute_service(clear, {})
await _wait_for(log, acked, STARTUP_TIMEOUT)
await _gather(meter, [completed, zeroed])
# Queued during startup, sent only once the active stream started
if queued_during_startup:
assert CLEAR_ACCUMULATED_FLOW not in log.tx_between(
0, log.index(ACTIVE_STARTED)
)
assert log.tx.count(CLEAR_ACCUMULATED_FLOW) == 1
assert log.index(CLEAR_ACCUMULATED_FLOW) < log.index(CLEAR_ACKED)
# Two requests in a row share one command; without an answer it is reported after the 500 ms ACK timeout
sent = log.next(CLEAR_ACCUMULATED_FLOW)
not_acked = log.next(CLEAR_NOT_ACKED)
completed = meter.expect("clear_completions", _approx(3), 3.0)
meter.set_silent(True)
await asyncio.sleep(0.2)
await client.execute_service(clear, {})
await client.execute_service(clear, {})
sent_at = await _wait_for(log, sent, 2.0)
not_acked_at = await _wait_for(log, not_acked, 2.0)
await _gather(meter, [completed])
assert not_acked_at - sent_at >= 0.4
assert log.tx.count(CLEAR_ACCUMULATED_FLOW) == 2
assert meter.states["clear_completions"] == [1.0, 2.0, 3.0]