diff --git a/esphome/components/ufm01/__init__.py b/esphome/components/ufm01/__init__.py index e1fcef107d5..1846958e591 100644 --- a/esphome/components/ufm01/__init__.py +++ b/esphome/components/ufm01/__init__.py @@ -1,7 +1,11 @@ +from dataclasses import dataclass + +from esphome import automation import esphome.codegen as cg from esphome.components import uart import esphome.config_validation as cv from esphome.const import CONF_ID +from esphome.core import CORE from esphome.types import ConfigType CODEOWNERS = ["@ljungqvist"] @@ -16,6 +20,23 @@ UFM01Component = ufm01_ns.class_("UFM01Component", uart.UARTDevice, cg.Component CONF_UFM01_ID = "ufm01_id" + +@dataclass +class UFM01Data: + clear_action_used: bool = False + + +def _get_data() -> UFM01Data: + if DOMAIN not in CORE.data: + CORE.data[DOMAIN] = UFM01Data() + return CORE.data[DOMAIN] + + +def _note_clear_action(config: ConfigType) -> ConfigType: + _get_data().clear_action_used = True + return config + + CONFIG_SCHEMA = ( cv.Schema( { @@ -26,6 +47,13 @@ CONFIG_SCHEMA = ( .extend(cv.COMPONENT_SCHEMA) ) +CLEAR_ACCUMULATED_FLOW_ACTION_SCHEMA = automation.maybe_simple_id( + { + cv.Required(CONF_ID): cv.use_id(UFM01Component), + }, + _note_clear_action, +) + FINAL_VALIDATE_SCHEMA = uart.final_validate_device_schema( "ufm01", require_tx=True, @@ -41,3 +69,12 @@ async def to_code(config: ConfigType) -> None: var = cg.new_Pvariable(config[CONF_ID]) await cg.register_component(var, config) await uart.register_uart_device(var, config) + if _get_data().clear_action_used: + cg.add_define("USE_UFM01_CLEAR_ACCUMULATED_FLOW_ACTION") + + +automation.register_apply_action( + "ufm01.clear_accumulated_flow", + CLEAR_ACCUMULATED_FLOW_ACTION_SCHEMA, + automation.ApplyCall("request_clear_accumulated_flow()"), +) diff --git a/esphome/components/ufm01/text_sensor.py b/esphome/components/ufm01/text_sensor.py new file mode 100644 index 00000000000..dcb5d3294c2 --- /dev/null +++ b/esphome/components/ufm01/text_sensor.py @@ -0,0 +1,34 @@ +import esphome.codegen as cg +from esphome.components import text_sensor +import esphome.config_validation as cv +from esphome.const import ENTITY_CATEGORY_DIAGNOSTIC, ICON_CHIP, ICON_FINGERPRINT +from esphome.types import ConfigType + +from . import CONF_UFM01_ID, UFM01Component + +DEPENDENCIES = ["ufm01"] + +CONF_METER_ID = "meter_id" +CONF_SOFTWARE_VERSION = "software_version" + +CONFIG_SCHEMA = { + cv.GenerateID(CONF_UFM01_ID): cv.use_id(UFM01Component), + cv.Optional(CONF_METER_ID): text_sensor.text_sensor_schema( + entity_category=ENTITY_CATEGORY_DIAGNOSTIC, icon=ICON_FINGERPRINT + ), + cv.Optional(CONF_SOFTWARE_VERSION): text_sensor.text_sensor_schema( + entity_category=ENTITY_CATEGORY_DIAGNOSTIC, icon=ICON_CHIP + ), +} + + +async def to_code(config: ConfigType) -> None: + hub = await cg.get_variable(config[CONF_UFM01_ID]) + sensors = text_sensor.sub_text_sensors(config) + if await sensors(CONF_METER_ID, hub.set_meter_id_text_sensor) is not None: + cg.add_define("USE_UFM01_METER_ID") + if ( + await sensors(CONF_SOFTWARE_VERSION, hub.set_software_version_text_sensor) + is not None + ): + cg.add_define("USE_UFM01_SOFTWARE_VERSION") diff --git a/esphome/components/ufm01/ufm01.cpp b/esphome/components/ufm01/ufm01.cpp index c68842ab6c1..47ff2165f8c 100644 --- a/esphome/components/ufm01/ufm01.cpp +++ b/esphome/components/ufm01/ufm01.cpp @@ -1,8 +1,9 @@ #include "ufm01.h" -#include "esphome/core/hal.h" +#include "esphome/core/application.h" #include "esphome/core/helpers.h" #include "esphome/core/log.h" +#include #include #include #include @@ -13,6 +14,7 @@ ESPHOME_LOG_TAG(TAG, "ufm01"); static constexpr uint8_t COMMAND_ACK = 0xE5; static constexpr uint32_t COMMAND_ACK_TIMEOUT_MS = 500; +static constexpr uint32_t CLEAR_QUEUE_TIMEOUT_MS = 15000; static constexpr uint32_t STARTUP_DELAY_MS = 2000; static constexpr uint32_t POST_RESET_DELAY_MS = 2000; static constexpr uint32_t RESET_RETRY_DELAY_MS = 800; @@ -20,23 +22,40 @@ static constexpr uint32_t STARTUP_RETRY_MS = 3000; static constexpr uint32_t PASSIVE_POLL_INTERVAL_MS = 1000; static constexpr uint32_t ACTIVE_STALE_MS = 5000; static constexpr uint32_t PASSIVE_READ_TIMEOUT_MS = 1000; +static constexpr uint32_t SOFTWARE_VERSION_READ_TIMEOUT_MS = 2000; +static constexpr uint32_t SOFTWARE_VERSION_RETRY_MS = 30000; +static constexpr uint8_t SOFTWARE_VERSION_MAX_ATTEMPTS = 3; static constexpr uint32_t ACTIVE_FRAME_TIMEOUT_MS = 3000; +// After this many consecutive passive poll failures, re-run the reset/startup sequence +static constexpr uint8_t PASSIVE_FAIL_ESCALATE_COUNT = 8; static constexpr float L_PER_M3 = 1000.0f; static constexpr float M3_PER_L = 1.0f / L_PER_M3; static constexpr std::array ACTIVE_MODE = {0xFE, 0xFE, 0x11, 0x5C, 0x00, 0x5C, 0x16}; +static constexpr std::array PASSIVE_MODE = {0xFE, 0xFE, 0x11, 0x5C, 0x01, 0x5D, 0x16}; static constexpr std::array CLEAR_ACCUMULATED_FLOW = {0xFE, 0xFE, 0x11, 0x5A, 0xFD, 0x57, 0x16}; static constexpr std::array RESET_DEVICE = {0xFE, 0xFE, 0x11, 0x5D, 0xCB, 0x28, 0x16}; static constexpr std::array READ_SENSOR_DATA_NO_ID = {0xFE, 0xFE, 0x11, 0x5B, 0x0F, 0x6A, 0x16}; +static constexpr std::array READ_SENSOR_DATA_WITH_ID = {0xFE, 0xFE, 0x11, 0x5B, 0xCB, 0x26, 0x16}; +static constexpr std::array GET_SOFTWARE_VERSION = {0xFE, 0xFE, 0x11, 0x5E, 0x62, 0xC0, 0x16}; // Active-mode frame layout (datasheet Table 7) +static constexpr size_t FRAME_DEVICE_ID_INDEX = 2; static constexpr size_t FRAME_CHECKSUM_INDEX = 30; static constexpr size_t FRAME_STOP_INDEX = 31; 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 size_t PASSIVE_WITH_ID_DEVICE_ID_INDEX = 2; +static constexpr uint8_t PASSIVE_WITH_ID_ACC_FLOW_FLAG_INDEX = 8; +static constexpr uint8_t PASSIVE_WITH_ID_INSTANT_FLOW_FLAG_INDEX = 22; +static constexpr uint8_t PASSIVE_WITH_ID_TEMP_FLAG_INDEX = 31; +static constexpr uint8_t PASSIVE_WITH_ID_ST1_INDEX = 35; +static constexpr size_t PASSIVE_WITH_ID_CHECKSUM_INDEX = 37; +static constexpr size_t PASSIVE_WITH_ID_STOP_INDEX = 38; static constexpr uint8_t FRAME_INDEX_INSTANT_FLOW_FLAG = 15; static constexpr uint8_t FRAME_INDEX_RESERVED_SECTION = 21; static constexpr uint8_t FRAME_INDEX_TEMP_FLAG = 24; @@ -58,8 +77,40 @@ static constexpr uint8_t ST2_UFC_ERROR_MASK = 0x20; static constexpr uint8_t ST2_FLOW_DIRECTION_WRONG_MASK = 0x08; static constexpr uint8_t ST2_FLOW_RATE_OUT_OF_RANGE_MASK = 0x04; +static_assert(SOFTWARE_VERSION_RESPONSE_SIZE <= PASSIVE_FRAME_MAX_SIZE, "software version read reuses passive_frame_"); +static constexpr size_t SOFTWARE_VERSION_CHECKSUM_INDEX = 5; +static constexpr size_t SOFTWARE_VERSION_STOP_INDEX = 6; + static float to_float(uint8_t data) { return (data >> 4) * 10 + (data & 0x0F); } +static bool bcd_to_digits(const uint8_t *data, size_t len, char *out) { + bool all_zero = true; + for (size_t i = 0; i < len; ++i) { + const uint8_t lo = data[i] & 0x0F; + const uint8_t hi = data[i] >> 4; + if (lo > 9 || hi > 9) + return false; + const size_t pos = (len - 1 - i) * 2; + if (hi != 0) + all_zero = false; + out[pos] = static_cast('0' + hi); + if (lo != 0) + all_zero = false; + out[pos + 1] = static_cast('0' + lo); + } + out[len * 2] = '\0'; + return !all_zero; +} + +static bool validate_software_version_response(const uint8_t data[SOFTWARE_VERSION_RESPONSE_SIZE]) { + if (data[0] != COMMAND_ACK || data[SOFTWARE_VERSION_STOP_INDEX] != FRAME_STOP_BYTE) + return false; + uint8_t sum = 0; + for (size_t i = 1; i < SOFTWARE_VERSION_CHECKSUM_INDEX; ++i) + sum += data[i]; + return data[SOFTWARE_VERSION_CHECKSUM_INDEX] == sum; +} + static bool check_byte(const uint8_t data[FRAME_SIZE], size_t index, uint8_t expected, const char *name) { if (data[index] == expected) return true; @@ -84,25 +135,42 @@ static bool validate_passive_frame(const uint8_t data[PASSIVE_FRAME_SIZE]) { return data[21] == sum_bytes(data, 21); } -static void passive_no_id_to_active_frame(const uint8_t passive[PASSIVE_FRAME_SIZE], uint8_t active[FRAME_SIZE]) { +#ifdef USE_UFM01_METER_ID +static bool validate_passive_with_id_frame(const uint8_t data[PASSIVE_FRAME_WITH_ID_SIZE]) { + if (data[0] != FRAME_START_BYTE_1 || data[1] != PASSIVE_START_BYTE_2_WITH_ID || + data[PASSIVE_WITH_ID_STOP_INDEX] != FRAME_STOP_BYTE) + return false; + uint8_t sum = 0; + for (size_t i = 0; i < PASSIVE_WITH_ID_CHECKSUM_INDEX; ++i) + sum += data[i]; + return data[PASSIVE_WITH_ID_CHECKSUM_INDEX] == sum; +} +#endif + +// Start index of each run in a passive reply; each run maps to the matching run of the active frame +struct PassiveLayout { + uint8_t acc_flow; // flag + 6 bytes, active 8..14 + uint8_t instant_flow; // flag + 5 bytes, active 15..20 + uint8_t temperature; // flag + 3 bytes, active 24..27 + uint8_t status; // ST1, ST2, checksum, active 28..30 +}; +static constexpr PassiveLayout PASSIVE_NO_ID_LAYOUT{2, 9, 15, 19}; +#ifdef USE_UFM01_METER_ID +static constexpr PassiveLayout PASSIVE_WITH_ID_LAYOUT{PASSIVE_WITH_ID_ACC_FLOW_FLAG_INDEX, + PASSIVE_WITH_ID_INSTANT_FLOW_FLAG_INDEX, + PASSIVE_WITH_ID_TEMP_FLAG_INDEX, PASSIVE_WITH_ID_ST1_INDEX}; +#endif + +static void passive_to_active_frame(const uint8_t *passive, const PassiveLayout &layout, uint8_t active[FRAME_SIZE]) { std::memset(active, 0, FRAME_SIZE); active[0] = FRAME_START_BYTE_1; active[1] = FRAME_START_BYTE_2; - active[7] = 0x01; - active[8] = passive[2]; - for (size_t i = 0; i < 6; ++i) - active[9 + i] = passive[3 + i]; - active[15] = passive[9]; - for (size_t i = 0; i < 5; ++i) - active[16 + i] = passive[10 + i]; - active[21] = FRAME_FLAG_RESERVED_SECTION; - active[24] = passive[15]; - for (size_t i = 0; i < 3; ++i) - active[25 + i] = passive[16 + i]; - active[28] = passive[19]; - active[29] = passive[20]; - active[30] = passive[21]; - active[31] = FRAME_STOP_BYTE; + std::copy_n(&passive[layout.acc_flow], 7, &active[FRAME_ACC_FLOW_FLAG_INDEX]); + std::copy_n(&passive[layout.instant_flow], 6, &active[FRAME_INDEX_INSTANT_FLOW_FLAG]); + active[FRAME_INDEX_RESERVED_SECTION] = FRAME_FLAG_RESERVED_SECTION; + std::copy_n(&passive[layout.temperature], 4, &active[FRAME_INDEX_TEMP_FLAG]); + std::copy_n(&passive[layout.status], 3, &active[FRAME_ST1_INDEX]); + active[FRAME_STOP_INDEX] = FRAME_STOP_BYTE; } static float read_accumulated_flow(const uint8_t data[FRAME_SIZE]) { @@ -118,8 +186,10 @@ static float read_flow(const uint8_t data[FRAME_SIZE]) { M3_PER_L; } +static constexpr size_t LOG_HEX_MAX_SIZE = std::max(FRAME_SIZE, PASSIVE_FRAME_MAX_SIZE); + static void log_hex(const uint8_t *data, size_t len) { - char hex_buf[format_hex_pretty_size(FRAME_SIZE)]; + char hex_buf[format_hex_pretty_size(LOG_HEX_MAX_SIZE)]; ESP_LOGD(TAG, "%s", format_hex_pretty_to(hex_buf, data, len, ' ')); } @@ -157,7 +227,6 @@ void UFM01Component::send_command_no_wait_(const std::array &command this->flush(); } -// Drains whatever is currently in the RX buffer, looking for a command ACK. bool UFM01Component::consume_ack_() { while (this->available()) { uint8_t byte; @@ -170,29 +239,65 @@ bool UFM01Component::consume_ack_() { return false; } -bool UFM01Component::send_command_(const std::array &command) { - this->send_command_no_wait_(command); - const uint32_t start = millis(); - while (millis() - start < COMMAND_ACK_TIMEOUT_MS) { - if (this->consume_ack_()) - return true; - delay(1); - } - return false; +#ifdef USE_UFM01_CLEAR_ACCUMULATED_FLOW_ACTION +bool UFM01Component::can_start_clear_action_() const { + return +#ifdef USE_UFM01_SOFTWARE_VERSION + !this->software_version_read_pending_ && +#endif + (this->operating_mode_ == OperatingMode::ACTIVE_STREAM || + (this->operating_mode_ == OperatingMode::PASSIVE_POLL && !this->passive_read_pending_)); } -bool UFM01Component::reset_device_() { return this->send_command_(RESET_DEVICE); } +void UFM01Component::request_clear_accumulated_flow() { + if (this->clear_pending_) { + ESP_LOGD(TAG, "Clear accumulated flow already pending"); + return; + } + this->clear_pending_ = true; + this->pending_clear_sent_ = false; + this->pending_clear_start_ms_ = App.get_loop_component_start_time(); +} -bool UFM01Component::clear_accumulated_flow_() { return this->send_command_(CLEAR_ACCUMULATED_FLOW); } +void UFM01Component::finish_pending_clear_() { + this->clear_pending_ = false; + this->pending_clear_sent_ = false; +} -bool UFM01Component::set_active_mode_() { return this->send_command_(ACTIVE_MODE); } +void UFM01Component::loop_pending_clear_action_(uint32_t now) { + if (!this->clear_pending_) + return; + if (!this->pending_clear_sent_) { + if (!this->can_start_clear_action_()) { + if (now - this->pending_clear_start_ms_ < CLEAR_QUEUE_TIMEOUT_MS) + return; + ESP_LOGW(TAG, "Clear accumulated flow timed out waiting for an idle UART"); + this->finish_pending_clear_(); + return; + } + this->pending_clear_start_ms_ = now; + this->pending_clear_sent_ = true; + this->send_command_no_wait_(CLEAR_ACCUMULATED_FLOW); + return; + } + if (this->consume_ack_()) { + ESP_LOGI(TAG, "Clear accumulated flow acknowledged"); + this->finish_pending_clear_(); + return; + } + if (now - this->pending_clear_start_ms_ < COMMAND_ACK_TIMEOUT_MS) + return; + ESP_LOGW(TAG, "Clear accumulated flow not acknowledged"); + this->finish_pending_clear_(); +} +#endif // USE_UFM01_CLEAR_ACCUMULATED_FLOW_ACTION float UFM01Component::get_setup_priority() const { return setup_priority::LATE; } void UFM01Component::setup() { ESP_LOGI(TAG, "Setting up UFM-01..."); this->startup_wait_ms_ = STARTUP_DELAY_MS; - this->set_startup_phase_(StartupPhase::WAIT); + this->set_startup_phase_(StartupPhase::WAIT, App.get_loop_component_start_time()); } void UFM01Component::dump_config() { @@ -208,9 +313,123 @@ void UFM01Component::dump_config() { LOG_BINARY_SENSOR(" ", "Empty Tube", this->empty_tube_binary_sensor_); LOG_BINARY_SENSOR(" ", "Flow Rate Out Of Range", this->flow_rate_out_of_range_binary_sensor_); #endif +#ifdef USE_UFM01_METER_ID + LOG_TEXT_SENSOR(" ", "Meter ID", this->meter_id_text_sensor_); +#endif +#ifdef USE_UFM01_SOFTWARE_VERSION + LOG_TEXT_SENSOR(" ", "Software Version", this->software_version_text_sensor_); +#endif } -void UFM01Component::on_active_frame_(uint8_t data[FRAME_SIZE]) { +// The binary sensors keep their last state while readings are stale; only flow and temperature go unknown +void UFM01Component::publish_stale_flow_and_temperature_() { +#ifdef USE_SENSOR + if (this->flow_sensor_ != nullptr) + this->flow_sensor_->publish_state(NAN); + if (this->temperature_sensor_ != nullptr) + this->temperature_sensor_->publish_state(NAN); +#endif +} + +#ifdef USE_UFM01_METER_ID +void UFM01Component::publish_meter_id_from_frame_(const uint8_t data[FRAME_SIZE]) { + if (this->meter_id_text_sensor_ == nullptr || this->meter_id_published_) + return; + + char meter_id_str[DEVICE_ID_STRING_LENGTH + 1]; + if (!bcd_to_digits(&data[FRAME_DEVICE_ID_INDEX], DEVICE_ID_LENGTH, meter_id_str)) { + if (!this->meter_id_invalid_logged_) { + ESP_LOGW(TAG, "Meter ID in frame is zero or not BCD, leaving the sensor unknown"); + this->meter_id_invalid_logged_ = true; + } + return; + } + + this->meter_id_text_sensor_->publish_state(meter_id_str); + this->meter_id_published_ = true; + ESP_LOGI(TAG, "UFM-01 meter ID: %s", meter_id_str); +} +#endif + +#ifdef USE_UFM01_SOFTWARE_VERSION +void UFM01Component::start_software_version_read_(uint32_t now) { + this->send_command_no_wait_(GET_SOFTWARE_VERSION); + this->passive_index_ = 0; + this->passive_start_ms_ = now; + this->software_version_read_pending_ = true; +} + +ReadResult UFM01Component::continue_software_version_read_(uint32_t now) { + while (this->available() && this->passive_index_ < SOFTWARE_VERSION_RESPONSE_SIZE) { + uint8_t byte; + if (!this->read_byte(&byte)) + break; + // The tail of an in-flight active-stream frame can arrive before the reply + if (this->passive_index_ == 0 && byte != COMMAND_ACK) + continue; + this->passive_frame_[this->passive_index_++] = byte; + } + + if (this->passive_index_ < SOFTWARE_VERSION_RESPONSE_SIZE) { + if (now - this->passive_start_ms_ < SOFTWARE_VERSION_READ_TIMEOUT_MS) + return ReadResult::READ_RESULT_PENDING; + ESP_LOGD(TAG, "software version read timeout (%u/%u bytes)", this->passive_index_, + static_cast(SOFTWARE_VERSION_RESPONSE_SIZE)); + return ReadResult::READ_RESULT_FAILURE; + } + + if (!validate_software_version_response(this->passive_frame_)) { + log_hex(this->passive_frame_, SOFTWARE_VERSION_RESPONSE_SIZE); + ESP_LOGW(TAG, "invalid software version response"); + return ReadResult::READ_RESULT_FAILURE; + } + + if (this->software_version_text_sensor_ != nullptr && !this->software_version_published_) { + char version_str[SOFTWARE_VERSION_STRING_LENGTH + 1]; + if (!bcd_to_digits(&this->passive_frame_[1], SOFTWARE_VERSION_LENGTH, version_str)) { + ESP_LOGW(TAG, "invalid BCD data in software version response"); + return ReadResult::READ_RESULT_FAILURE; + } + this->software_version_text_sensor_->publish_state(version_str); + this->software_version_published_ = true; + ESP_LOGI(TAG, "UFM-01 software version: %s", version_str); + } + + return ReadResult::READ_RESULT_SUCCESS; +} + +void UFM01Component::finish_software_version_read_(ReadResult result, uint32_t now) { + this->software_version_read_pending_ = false; + this->last_software_version_attempt_ms_ = now; + if (result == ReadResult::READ_RESULT_FAILURE && + ++this->software_version_failures_ == SOFTWARE_VERSION_MAX_ATTEMPTS) { + ESP_LOGW(TAG, "Software version not read after %u attempts, giving up", SOFTWARE_VERSION_MAX_ATTEMPTS); + } +} + +// Returns true while a software version read owns the UART, so the caller skips its own work +bool UFM01Component::service_software_version_(uint32_t now) { + if (this->software_version_read_pending_) { + const ReadResult result = this->continue_software_version_read_(now); + if (result != ReadResult::READ_RESULT_PENDING) + this->finish_software_version_read_(result, now); + return true; + } + if (this->software_version_text_sensor_ == nullptr || this->software_version_published_ || + this->software_version_failures_ >= SOFTWARE_VERSION_MAX_ATTEMPTS || this->passive_read_pending_) + return false; + if (this->last_software_version_attempt_ms_ != 0 && + now - this->last_software_version_attempt_ms_ < SOFTWARE_VERSION_RETRY_MS) + return false; + this->start_software_version_read_(now); + return true; +} +#endif // USE_UFM01_SOFTWARE_VERSION + +void UFM01Component::on_active_frame_(uint8_t data[FRAME_SIZE], uint32_t now) { +#ifdef USE_UFM01_METER_ID + this->publish_meter_id_from_frame_(data); +#endif bool empty_tube = read_empty_tube(data); #ifdef USE_BINARY_SENSOR if (this->ufc_chip_error_binary_sensor_ != nullptr) @@ -229,10 +448,7 @@ void UFM01Component::on_active_frame_(uint8_t data[FRAME_SIZE]) { this->accumulated_flow_sensor_->publish_state(read_accumulated_flow(data)); if (empty_tube) { - if (this->flow_sensor_ != nullptr) - this->flow_sensor_->publish_state(NAN); - if (this->temperature_sensor_ != nullptr) - this->temperature_sensor_->publish_state(NAN); + this->publish_stale_flow_and_temperature_(); } else { if (this->flow_sensor_ != nullptr) this->flow_sensor_->publish_state(read_flow(data)); @@ -240,12 +456,12 @@ void UFM01Component::on_active_frame_(uint8_t data[FRAME_SIZE]) { this->temperature_sensor_->publish_state(read_temperature(data)); } #endif - this->last_valid_frame_ms_ = millis(); + this->last_valid_frame_ms_ = now; this->status_clear_warning(); this->status_clear_error(); } -bool UFM01Component::process_active_stream_() { +bool UFM01Component::process_active_stream_(uint32_t now) { bool got_valid_frame = false; while (this->available()) { @@ -264,7 +480,7 @@ bool UFM01Component::process_active_stream_() { continue; if (validate_active_frame(this->data_)) { - this->on_active_frame_(this->data_); + this->on_active_frame_(this->data_, now); this->read_index_ = 0; got_valid_frame = true; continue; @@ -287,33 +503,89 @@ bool UFM01Component::process_active_stream_() { return got_valid_frame; } -void UFM01Component::set_startup_phase_(StartupPhase phase) { - this->startup_phase_ = phase; - this->phase_start_ms_ = millis(); -} - -void UFM01Component::enter_active_stream_(const char *reason) { - ESP_LOGI(TAG, "UFM-01 active stream %s", reason); - this->operating_mode_ = OperatingMode::ACTIVE_STREAM; +void UFM01Component::set_mode_(OperatingMode mode, uint32_t now) { + this->operating_mode_ = mode; + this->phase_start_ms_ = now; this->passive_read_pending_ = false; +#ifdef USE_UFM01_SOFTWARE_VERSION + this->software_version_read_pending_ = false; +#endif + this->consecutive_passive_failures_ = 0; } -void UFM01Component::start_passive_read_() { - this->send_command_no_wait_(READ_SENSOR_DATA_NO_ID); +void UFM01Component::set_startup_phase_(StartupPhase phase, uint32_t now) { + this->startup_phase_ = phase; + this->phase_start_ms_ = now; +} + +void UFM01Component::send_reset_(const LogString *status, uint32_t now) { + this->status_set_warning(status); + this->reset_retried_ = false; + this->send_command_no_wait_(RESET_DEVICE); + this->set_startup_phase_(StartupPhase::RESET_WAIT_ACK, now); +} + +void UFM01Component::enter_active_stream_(const LogString *reason, uint32_t now) { + ESP_LOGI(TAG, "UFM-01 active stream %s", LOG_STR_ARG(reason)); + this->set_mode_(OperatingMode::ACTIVE_STREAM, now); +} + +void UFM01Component::enter_passive_from_stale_(uint32_t now) { + ESP_LOGW(TAG, "Active stream stale, switching to passive polling"); + this->publish_stale_flow_and_temperature_(); + // Flushes leftover stream bytes first; the ACK is not awaited, the next command's flush discards it + this->send_command_no_wait_(PASSIVE_MODE); + this->set_mode_(OperatingMode::PASSIVE_POLL, now); + this->last_poll_ms_ = now; + this->status_set_warning(LOG_STR("UFM-01 passive poll")); +} + +void UFM01Component::restart_startup_(const LogString *reason, uint32_t now) { + ESP_LOGW(TAG, "%s, re-initializing UFM-01", LOG_STR_ARG(reason)); + this->publish_stale_flow_and_temperature_(); + this->set_mode_(OperatingMode::STARTUP, now); + this->send_reset_(LOG_STR("re-initializing UFM-01"), now); +} + +void UFM01Component::note_passive_poll_result_(bool success, uint32_t now) { + if (success) { + this->consecutive_passive_failures_ = 0; + return; + } + this->status_set_warning(LOG_STR("UFM-01 passive poll failed")); + if (++this->consecutive_passive_failures_ < PASSIVE_FAIL_ESCALATE_COUNT) + return; + this->restart_startup_(LOG_STR("Passive poll failed repeatedly"), now); +} + +void UFM01Component::start_passive_read_(uint32_t now) { +#ifdef USE_UFM01_METER_ID + if (this->meter_id_text_sensor_ != nullptr && !this->meter_id_published_) { + this->passive_expected_size_ = PASSIVE_FRAME_WITH_ID_SIZE; + this->passive_start_byte_2_ = PASSIVE_START_BYTE_2_WITH_ID; + this->send_command_no_wait_(READ_SENSOR_DATA_WITH_ID); + } else +#endif + { + this->passive_expected_size_ = PASSIVE_FRAME_SIZE; + this->passive_start_byte_2_ = PASSIVE_START_BYTE_2; + this->send_command_no_wait_(READ_SENSOR_DATA_NO_ID); + } this->passive_index_ = 0; - this->passive_start_ms_ = millis(); + this->passive_start_ms_ = now; } -// Accumulates the reply to a passive read request across loop iterations. -PassiveReadResult UFM01Component::continue_passive_read_() { - while (this->available() && this->passive_index_ < PASSIVE_FRAME_SIZE) { +ReadResult UFM01Component::continue_passive_read_(uint32_t now) { + const uint8_t expected_size = this->passive_expected_size_; + + while (this->available() && this->passive_index_ < expected_size) { uint8_t byte; if (!this->read_byte(&byte)) break; if (this->passive_index_ == 0 && byte != FRAME_START_BYTE_1) continue; - if (this->passive_index_ == 1 && byte != PASSIVE_START_BYTE_2) { + if (this->passive_index_ == 1 && byte != this->passive_start_byte_2_) { // The mismatched byte may itself be the start of the real frame this->passive_index_ = (byte == FRAME_START_BYTE_1) ? 1 : 0; continue; @@ -321,57 +593,69 @@ PassiveReadResult UFM01Component::continue_passive_read_() { this->passive_frame_[this->passive_index_++] = byte; } - if (this->passive_index_ < PASSIVE_FRAME_SIZE) { - if (millis() - this->passive_start_ms_ < PASSIVE_READ_TIMEOUT_MS) - return PassiveReadResult::PASSIVE_READ_RESULT_PENDING; - ESP_LOGD(TAG, "passive read timeout (%zu/%zu bytes)", this->passive_index_, PASSIVE_FRAME_SIZE); - return PassiveReadResult::PASSIVE_READ_RESULT_FAILURE; - } - - if (!validate_passive_frame(this->passive_frame_)) { - log_hex(this->passive_frame_, PASSIVE_FRAME_SIZE); - ESP_LOGW(TAG, "invalid passive frame"); - return PassiveReadResult::PASSIVE_READ_RESULT_FAILURE; + if (this->passive_index_ < expected_size) { + if (now - this->passive_start_ms_ < PASSIVE_READ_TIMEOUT_MS) + return ReadResult::READ_RESULT_PENDING; + ESP_LOGD(TAG, "passive read timeout (%u/%u bytes)", this->passive_index_, expected_size); + return ReadResult::READ_RESULT_FAILURE; } uint8_t active_frame[FRAME_SIZE]; - passive_no_id_to_active_frame(this->passive_frame_, active_frame); - this->on_active_frame_(active_frame); - return PassiveReadResult::PASSIVE_READ_RESULT_SUCCESS; +#ifdef USE_UFM01_METER_ID + if (expected_size == PASSIVE_FRAME_WITH_ID_SIZE) { + if (!validate_passive_with_id_frame(this->passive_frame_)) { + log_hex(this->passive_frame_, PASSIVE_FRAME_WITH_ID_SIZE); + ESP_LOGW(TAG, "invalid passive frame with ID"); + return ReadResult::READ_RESULT_FAILURE; + } + passive_to_active_frame(this->passive_frame_, PASSIVE_WITH_ID_LAYOUT, active_frame); + // Device ID and the byte after it + std::copy_n(&this->passive_frame_[PASSIVE_WITH_ID_DEVICE_ID_INDEX], DEVICE_ID_LENGTH + 1, + &active_frame[FRAME_DEVICE_ID_INDEX]); + } else +#endif + { + if (!validate_passive_frame(this->passive_frame_)) { + log_hex(this->passive_frame_, PASSIVE_FRAME_SIZE); + ESP_LOGW(TAG, "invalid passive frame"); + return ReadResult::READ_RESULT_FAILURE; + } + passive_to_active_frame(this->passive_frame_, PASSIVE_NO_ID_LAYOUT, active_frame); + active_frame[FRAME_DEVICE_ID_INDEX + DEVICE_ID_LENGTH] = 0x01; + } + this->on_active_frame_(active_frame, now); + return ReadResult::READ_RESULT_SUCCESS; } -void UFM01Component::loop_startup_() { - const uint32_t elapsed = millis() - this->phase_start_ms_; +void UFM01Component::loop_startup_(uint32_t now) { + const uint32_t elapsed = now - this->phase_start_ms_; switch (this->startup_phase_) { case StartupPhase::WAIT: // Pick up an already-streaming device without resetting it - if (this->process_active_stream_()) { - this->enter_active_stream_("started"); + if (this->process_active_stream_(now)) { + this->enter_active_stream_(LOG_STR("started"), now); return; } if (elapsed < this->startup_wait_ms_) return; ESP_LOGD(TAG, "Running startup sequence"); - this->status_set_warning("initializing UFM-01"); - this->reset_retried_ = false; - this->send_command_no_wait_(RESET_DEVICE); - this->set_startup_phase_(StartupPhase::RESET_WAIT_ACK); + this->send_reset_(LOG_STR("initializing UFM-01"), now); return; case StartupPhase::RESET_WAIT_ACK: if (this->consume_ack_()) { - this->set_startup_phase_(StartupPhase::POST_RESET_WAIT); + this->set_startup_phase_(StartupPhase::POST_RESET_WAIT, now); return; } if (elapsed < COMMAND_ACK_TIMEOUT_MS) return; if (!this->reset_retried_) { ESP_LOGW(TAG, "Reset not acknowledged, retrying in %" PRIu32 " ms", RESET_RETRY_DELAY_MS); - this->set_startup_phase_(StartupPhase::RESET_RETRY_WAIT); + this->set_startup_phase_(StartupPhase::RESET_RETRY_WAIT, now); } else { ESP_LOGW(TAG, "Reset failed during startup"); - this->set_startup_phase_(StartupPhase::POST_RESET_WAIT); + this->set_startup_phase_(StartupPhase::POST_RESET_WAIT, now); } return; @@ -380,91 +664,138 @@ void UFM01Component::loop_startup_() { return; this->reset_retried_ = true; this->send_command_no_wait_(RESET_DEVICE); - this->set_startup_phase_(StartupPhase::RESET_WAIT_ACK); + this->set_startup_phase_(StartupPhase::RESET_WAIT_ACK, now); return; case StartupPhase::POST_RESET_WAIT: if (elapsed < POST_RESET_DELAY_MS) return; +#ifdef USE_UFM01_SOFTWARE_VERSION + if (this->software_version_text_sensor_ != nullptr && !this->software_version_published_ && + this->software_version_failures_ < SOFTWARE_VERSION_MAX_ATTEMPTS) { + this->start_software_version_read_(now); + this->set_startup_phase_(StartupPhase::SOFTWARE_VERSION_WAIT_REPLY, now); + return; + } +#endif this->send_command_no_wait_(ACTIVE_MODE); - this->set_startup_phase_(StartupPhase::ACTIVE_WAIT_FRAME); + this->set_startup_phase_(StartupPhase::ACTIVE_WAIT_FRAME, now); return; +#ifdef USE_UFM01_SOFTWARE_VERSION + case StartupPhase::SOFTWARE_VERSION_WAIT_REPLY: { + const ReadResult result = this->continue_software_version_read_(now); + if (result == ReadResult::READ_RESULT_PENDING) + return; + if (result == ReadResult::READ_RESULT_SUCCESS) { + ESP_LOGD(TAG, "Software version read during startup"); + } else { + ESP_LOGW(TAG, "Software version read failed during startup, continuing"); + } + this->finish_software_version_read_(result, now); + this->send_command_no_wait_(ACTIVE_MODE); + this->set_startup_phase_(StartupPhase::ACTIVE_WAIT_FRAME, now); + return; + } +#endif + case StartupPhase::ACTIVE_WAIT_FRAME: // The command ACK (0xE5) is consumed by the frame parser as noise - if (this->process_active_stream_()) { - this->enter_active_stream_("started"); + if (this->process_active_stream_(now)) { + this->enter_active_stream_(LOG_STR("started"), now); return; } if (elapsed < ACTIVE_FRAME_TIMEOUT_MS) return; - this->start_passive_read_(); - this->set_startup_phase_(StartupPhase::PASSIVE_WAIT_REPLY); + this->send_command_no_wait_(PASSIVE_MODE); + this->set_startup_phase_(StartupPhase::SET_PASSIVE_WAIT_ACK, now); + return; + + case StartupPhase::SET_PASSIVE_WAIT_ACK: + if (this->consume_ack_()) { + ESP_LOGD(TAG, "Passive mode acknowledged during startup"); + } else if (elapsed < COMMAND_ACK_TIMEOUT_MS) { + return; + } else { + ESP_LOGW(TAG, "SET_PASSIVE_MODE not acknowledged during startup, continuing"); + } + this->start_passive_read_(now); + this->set_startup_phase_(StartupPhase::PASSIVE_WAIT_REPLY, now); return; case StartupPhase::PASSIVE_WAIT_REPLY: - switch (this->continue_passive_read_()) { - case PassiveReadResult::PASSIVE_READ_RESULT_PENDING: + switch (this->continue_passive_read_(now)) { + case ReadResult::READ_RESULT_PENDING: return; - case PassiveReadResult::PASSIVE_READ_RESULT_SUCCESS: + case ReadResult::READ_RESULT_SUCCESS: ESP_LOGI(TAG, "UFM-01 using passive polling"); - this->operating_mode_ = OperatingMode::PASSIVE_POLL; - this->passive_read_pending_ = false; - this->last_poll_ms_ = millis(); + this->set_mode_(OperatingMode::PASSIVE_POLL, now); + this->last_poll_ms_ = now; return; - case PassiveReadResult::PASSIVE_READ_RESULT_FAILURE: + case ReadResult::READ_RESULT_FAILURE: ESP_LOGW(TAG, "Startup failed, retrying in %" PRIu32 " ms", STARTUP_RETRY_MS); this->startup_wait_ms_ = STARTUP_RETRY_MS; - this->set_startup_phase_(StartupPhase::WAIT); + this->set_startup_phase_(StartupPhase::WAIT, now); return; } } } -void UFM01Component::loop_active_stream_() { - this->process_active_stream_(); - if (this->last_valid_frame_ms_ != 0 && millis() - this->last_valid_frame_ms_ > ACTIVE_STALE_MS) { - ESP_LOGW(TAG, "Active stream stale, switching to passive polling"); - this->operating_mode_ = OperatingMode::PASSIVE_POLL; - this->passive_read_pending_ = false; - this->last_poll_ms_ = 0; - this->status_set_warning("UFM-01 passive poll"); - } +void UFM01Component::loop_active_stream_(uint32_t now) { +#ifdef USE_UFM01_SOFTWARE_VERSION + if (this->service_software_version_(now)) + return; +#endif + + this->process_active_stream_(now); + + if (this->last_valid_frame_ms_ != 0 && now - this->last_valid_frame_ms_ > ACTIVE_STALE_MS) + this->enter_passive_from_stale_(now); } -void UFM01Component::loop_passive_poll_() { +void UFM01Component::loop_passive_poll_(uint32_t now) { +#ifdef USE_UFM01_SOFTWARE_VERSION + if (this->service_software_version_(now)) + return; +#endif + if (this->passive_read_pending_) { - const PassiveReadResult result = this->continue_passive_read_(); - if (result == PassiveReadResult::PASSIVE_READ_RESULT_PENDING) + const ReadResult result = this->continue_passive_read_(now); + if (result == ReadResult::READ_RESULT_PENDING) return; this->passive_read_pending_ = false; - if (result == PassiveReadResult::PASSIVE_READ_RESULT_FAILURE) - this->status_set_warning("UFM-01 passive poll failed"); + this->note_passive_poll_result_(result == ReadResult::READ_RESULT_SUCCESS, now); return; } - if (this->process_active_stream_()) { - this->enter_active_stream_("resumed"); + if (this->process_active_stream_(now)) { + this->enter_active_stream_(LOG_STR("resumed"), now); return; } - if (millis() - this->last_poll_ms_ >= PASSIVE_POLL_INTERVAL_MS) { - this->last_poll_ms_ = millis(); - this->start_passive_read_(); + if (now - this->last_poll_ms_ >= PASSIVE_POLL_INTERVAL_MS) { + this->last_poll_ms_ = now; + this->start_passive_read_(now); this->passive_read_pending_ = true; } } void UFM01Component::loop() { + const uint32_t now = App.get_loop_component_start_time(); +#ifdef USE_UFM01_CLEAR_ACCUMULATED_FLOW_ACTION + this->loop_pending_clear_action_(now); + if (this->clear_pending_ && this->pending_clear_sent_) + return; +#endif switch (this->operating_mode_) { case OperatingMode::STARTUP: - this->loop_startup_(); + this->loop_startup_(now); return; case OperatingMode::ACTIVE_STREAM: - this->loop_active_stream_(); + this->loop_active_stream_(now); return; case OperatingMode::PASSIVE_POLL: - this->loop_passive_poll_(); + this->loop_passive_poll_(now); return; } } diff --git a/esphome/components/ufm01/ufm01.h b/esphome/components/ufm01/ufm01.h index 0a39dcc9afd..cf73daa5fa7 100644 --- a/esphome/components/ufm01/ufm01.h +++ b/esphome/components/ufm01/ufm01.h @@ -8,6 +8,9 @@ #ifdef USE_SENSOR #include "esphome/components/sensor/sensor.h" #endif +#ifdef USE_TEXT_SENSOR +#include "esphome/components/text_sensor/text_sensor.h" +#endif #include "esphome/components/uart/uart.h" #include @@ -23,6 +26,17 @@ class TestableUFM01; static constexpr size_t FRAME_SIZE = 32; static constexpr size_t PASSIVE_FRAME_SIZE = 23; +static constexpr size_t PASSIVE_FRAME_WITH_ID_SIZE = 39; +#ifdef USE_UFM01_METER_ID +static constexpr size_t PASSIVE_FRAME_MAX_SIZE = PASSIVE_FRAME_WITH_ID_SIZE; +#else +static constexpr size_t PASSIVE_FRAME_MAX_SIZE = PASSIVE_FRAME_SIZE; +#endif +static constexpr size_t SOFTWARE_VERSION_RESPONSE_SIZE = 7; +static constexpr size_t DEVICE_ID_LENGTH = 5; +static constexpr size_t DEVICE_ID_STRING_LENGTH = 10; +static constexpr size_t SOFTWARE_VERSION_LENGTH = 4; +static constexpr size_t SOFTWARE_VERSION_STRING_LENGTH = 8; enum class OperatingMode : uint8_t { STARTUP = 0, @@ -31,18 +45,22 @@ enum class OperatingMode : uint8_t { }; enum class StartupPhase : uint8_t { - WAIT = 0, - RESET_WAIT_ACK = 1, - RESET_RETRY_WAIT = 2, - POST_RESET_WAIT = 3, - ACTIVE_WAIT_FRAME = 4, - PASSIVE_WAIT_REPLY = 5, + WAIT, + RESET_WAIT_ACK, + RESET_RETRY_WAIT, + POST_RESET_WAIT, +#ifdef USE_UFM01_SOFTWARE_VERSION + SOFTWARE_VERSION_WAIT_REPLY, +#endif + ACTIVE_WAIT_FRAME, + SET_PASSIVE_WAIT_ACK, + PASSIVE_WAIT_REPLY, }; -enum class PassiveReadResult : uint8_t { - PASSIVE_READ_RESULT_PENDING = 0, - PASSIVE_READ_RESULT_SUCCESS = 1, - PASSIVE_READ_RESULT_FAILURE = 2, +enum class ReadResult : uint8_t { + READ_RESULT_PENDING = 0, + READ_RESULT_SUCCESS = 1, + READ_RESULT_FAILURE = 2, }; class UFM01Component : public uart::UARTDevice, public Component { @@ -59,6 +77,13 @@ class UFM01Component : public uart::UARTDevice, public Component { SUB_BINARY_SENSOR(flow_rate_out_of_range) #endif +#ifdef USE_UFM01_METER_ID + SUB_TEXT_SENSOR(meter_id) +#endif +#ifdef USE_UFM01_SOFTWARE_VERSION + SUB_TEXT_SENSOR(software_version) +#endif + public: void setup() override; @@ -66,43 +91,85 @@ class UFM01Component : public uart::UARTDevice, public Component { void loop() override; +#ifdef USE_UFM01_CLEAR_ACCUMULATED_FLOW_ACTION + /// Queue a clear of the accumulated flow; sent once the UART is idle, a request while one is pending joins it + void request_clear_accumulated_flow(); +#endif + float get_setup_priority() const override; - protected: - bool clear_accumulated_flow_(); - bool set_active_mode_(); - bool reset_device_(); - private: - bool send_command_(const std::array &command); +#ifdef USE_UFM01_CLEAR_ACCUMULATED_FLOW_ACTION + bool can_start_clear_action_() const; + void loop_pending_clear_action_(uint32_t now); + void finish_pending_clear_(); +#endif void send_command_no_wait_(const std::array &command); bool consume_ack_(); void flush_rx_(); - bool process_active_stream_(); - void on_active_frame_(uint8_t data[FRAME_SIZE]); + bool process_active_stream_(uint32_t now); + void on_active_frame_(uint8_t data[FRAME_SIZE], uint32_t now); +#ifdef USE_UFM01_METER_ID + void publish_meter_id_from_frame_(const uint8_t data[FRAME_SIZE]); +#endif + void publish_stale_flow_and_temperature_(); +#ifdef USE_UFM01_SOFTWARE_VERSION + void start_software_version_read_(uint32_t now); + ReadResult continue_software_version_read_(uint32_t now); + void finish_software_version_read_(ReadResult result, uint32_t now); + bool service_software_version_(uint32_t now); +#endif - void loop_startup_(); - void loop_active_stream_(); - void loop_passive_poll_(); - void set_startup_phase_(StartupPhase phase); - void enter_active_stream_(const char *reason); - void start_passive_read_(); - PassiveReadResult continue_passive_read_(); + void loop_startup_(uint32_t now); + void loop_active_stream_(uint32_t now); + void loop_passive_poll_(uint32_t now); + void set_mode_(OperatingMode mode, uint32_t now); + void set_startup_phase_(StartupPhase phase, uint32_t now); + void send_reset_(const LogString *status, uint32_t now); + void enter_active_stream_(const LogString *reason, uint32_t now); + void enter_passive_from_stale_(uint32_t now); + void restart_startup_(const LogString *reason, uint32_t now); + void start_passive_read_(uint32_t now); + ReadResult continue_passive_read_(uint32_t now); + void note_passive_poll_result_(bool success, uint32_t now); + + uint32_t phase_start_ms_{0}; + uint32_t startup_wait_ms_{0}; + uint32_t last_valid_frame_ms_{0}; + uint32_t last_poll_ms_{0}; + // Shared by the passive read and the software version read, which never overlap + uint32_t passive_start_ms_{0}; +#ifdef USE_UFM01_SOFTWARE_VERSION + uint32_t last_software_version_attempt_ms_{0}; +#endif +#ifdef USE_UFM01_CLEAR_ACCUMULATED_FLOW_ACTION + uint32_t pending_clear_start_ms_{0}; +#endif + int32_t read_index_{0}; OperatingMode operating_mode_{OperatingMode::STARTUP}; StartupPhase startup_phase_{StartupPhase::WAIT}; - uint32_t phase_start_ms_{0}; - uint32_t startup_wait_ms_{0}; bool reset_retried_{false}; - uint32_t last_valid_frame_ms_{0}; - uint32_t last_poll_ms_{0}; - bool passive_read_pending_{false}; - uint32_t passive_start_ms_{0}; - size_t passive_index_{0}; - uint8_t passive_frame_[PASSIVE_FRAME_SIZE]; + uint8_t consecutive_passive_failures_{0}; + uint8_t passive_index_{0}; + uint8_t passive_expected_size_{PASSIVE_FRAME_SIZE}; + uint8_t passive_start_byte_2_{0}; +#ifdef USE_UFM01_SOFTWARE_VERSION + bool software_version_published_{false}; + bool software_version_read_pending_{false}; + uint8_t software_version_failures_{0}; +#endif +#ifdef USE_UFM01_METER_ID + bool meter_id_published_{false}; + bool meter_id_invalid_logged_{false}; +#endif +#ifdef USE_UFM01_CLEAR_ACCUMULATED_FLOW_ACTION + bool clear_pending_{false}; + bool pending_clear_sent_{false}; +#endif - int32_t read_index_ = 0; + uint8_t passive_frame_[PASSIVE_FRAME_MAX_SIZE]; uint8_t data_[FRAME_SIZE]; friend class testing::TestableUFM01; diff --git a/esphome/core/defines.h b/esphome/core/defines.h index 552221dd5eb..730e17fb53a 100644 --- a/esphome/core/defines.h +++ b/esphome/core/defines.h @@ -248,6 +248,9 @@ #define USE_TOUCHSCREEN #define USE_UART_DEBUGGER #define USE_UART_WAKE_LOOP_ON_RX +#define USE_UFM01_CLEAR_ACCUMULATED_FLOW_ACTION +#define USE_UFM01_METER_ID +#define USE_UFM01_SOFTWARE_VERSION #define USE_UPDATE #define USE_UPTIME_TIMESTAMP #define USE_VALVE diff --git a/tests/components/ufm01/__init__.py b/tests/components/ufm01/__init__.py new file mode 100644 index 00000000000..19febacc61a --- /dev/null +++ b/tests/components/ufm01/__init__.py @@ -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 diff --git a/tests/components/ufm01/benchmark.yaml b/tests/components/ufm01/benchmark.yaml new file mode 100644 index 00000000000..f78cdfe0334 --- /dev/null +++ b/tests/components/ufm01/benchmark.yaml @@ -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 diff --git a/tests/components/ufm01/common.h b/tests/components/ufm01/common.h index 1582358700c..e04e09b765c 100644 --- a/tests/components/ufm01/common.h +++ b/tests/components/ufm01/common.h @@ -1,6 +1,7 @@ #pragma once #include +#include #include #include #include @@ -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; +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 void enqueue(const std::array &data) { + this->rx_queue.insert(this->rx_queue.end(), data.begin(), data.end()); + } + void enqueue(std::initializer_list 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 inline uint8_t checksum(const std::array &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 make_active_frame() { std::array 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 make_software_version_response() { + // Datasheet example version 23380315 + std::array 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 make_passive_frame() { std::array frame{}; frame[0] = FRAME_START_BYTE_1; @@ -133,22 +246,93 @@ inline std::array 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 make_passive_with_id_frame() { + std::array 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(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_; }; diff --git a/tests/components/ufm01/common.yaml b/tests/components/ufm01/common.yaml index c818dc29651..29bc05dfabd 100644 --- a/tests/components/ufm01/common.yaml +++ b/tests/components/ufm01/common.yaml @@ -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 diff --git a/tests/components/ufm01/test.esp32-idf.yaml b/tests/components/ufm01/test.esp32-idf.yaml index 34041cc2237..a2d4469798c 100644 --- a/tests/components/ufm01/test.esp32-idf.yaml +++ b/tests/components/ufm01/test.esp32-idf.yaml @@ -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 diff --git a/tests/components/ufm01/test.esp8266-ard.yaml b/tests/components/ufm01/test.esp8266-ard.yaml index 195f4b41b59..7826cdc44a0 100644 --- a/tests/components/ufm01/test.esp8266-ard.yaml +++ b/tests/components/ufm01/test.esp8266-ard.yaml @@ -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 diff --git a/tests/components/ufm01/test.rp2040-ard.yaml b/tests/components/ufm01/test.rp2040-ard.yaml index 13b3284fe30..1483674ddcf 100644 --- a/tests/components/ufm01/test.rp2040-ard.yaml +++ b/tests/components/ufm01/test.rp2040-ard.yaml @@ -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 diff --git a/tests/components/ufm01/ufm01_clear_action_test.cpp b/tests/components/ufm01/ufm01_clear_action_test.cpp new file mode 100644 index 00000000000..a1d10b1926d --- /dev/null +++ b/tests/components/ufm01/ufm01_clear_action_test.cpp @@ -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 diff --git a/tests/components/ufm01/ufm01_diagnostics_test.cpp b/tests/components/ufm01/ufm01_diagnostics_test.cpp new file mode 100644 index 00000000000..5898b565633 --- /dev/null +++ b/tests/components/ufm01/ufm01_diagnostics_test.cpp @@ -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(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 diff --git a/tests/components/ufm01/ufm01_frame_test.cpp b/tests/components/ufm01/ufm01_frame_test.cpp index 3b1b148d504..8e659ce40df 100644 --- a/tests/components/ufm01/ufm01_frame_test.cpp +++ b/tests/components/ufm01/ufm01_frame_test.cpp @@ -4,7 +4,7 @@ namespace esphome::ufm01::testing { TEST_F(UFM01Test, ValidActiveFrameAccepted) { auto frame = make_active_frame(); - this->mock_uart_.enqueue(std::vector(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(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(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(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(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(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(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(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(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 diff --git a/tests/components/ufm01/ufm01_startup_test.cpp b/tests/components/ufm01/ufm01_startup_test.cpp index 6b7feb0f1b0..c6d5a7f4ebf 100644 --- a/tests/components/ufm01/ufm01_startup_test.cpp +++ b/tests/components/ufm01/ufm01_startup_test.cpp @@ -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(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(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 diff --git a/tests/components/ufm01/validate.host.yaml b/tests/components/ufm01/validate.host.yaml new file mode 100644 index 00000000000..19c0739f0bd --- /dev/null +++ b/tests/components/ufm01/validate.host.yaml @@ -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 diff --git a/tests/integration/fixtures/uart_mock_ufm01.yaml b/tests/integration/fixtures/uart_mock_ufm01.yaml new file mode 100644 index 00000000000..0f72225c089 --- /dev/null +++ b/tests/integration/fixtures/uart_mock_ufm01.yaml @@ -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 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{ + 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{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 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;" diff --git a/tests/integration/test_uart_mock_ufm01.py b/tests/integration/test_uart_mock_ufm01.py new file mode 100644 index 00000000000..c2c2f08f523 --- /dev/null +++ b/tests/integration/test_uart_mock_ufm01.py @@ -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]