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

Merge branch 'dev' into mqtt-subscription-topic-builders

Browse Source
This commit is contained in:
J. Nick Koston
2026-10-14 10:10:23 -10:00
committed by GitHub
parent d8d4c2f121 3c4cf50cb6
commit 56702436c6
37 changed files with 2445 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
esphome/components/ld2450/__init__.py
+11
View File
@@ -1,6 +1,7 @@
from esphome import automation
import esphome.codegen as cg
from esphome.components import uart
from esphome.config_helpers import filter_source_files_from_defines
import esphome.config_validation as cv
from esphome.const import CONF_ID, CONF_ON_DATA, CONF_THROTTLE
from esphome.types import ConfigType
@@ -16,6 +17,8 @@ LD2450Component = ld2450_ns.class_("LD2450Component", cg.Component, uart.UARTDev
CONF_LD2450_ID = "ld2450_id"
CONF_POLYGON_ZONE_ID = "polygon_zone_id"
CONF_POLYGON_ZONES = "polygon_zones"
CONFIG_SCHEMA = cv.All(
cv.Schema(
@@ -50,6 +53,14 @@ _CALLBACK_AUTOMATIONS = (
automation.CallbackAutomation(CONF_ON_DATA, "add_on_data_callback"),
)
# Polygon zone sources build only when zones are configured
FILTER_SOURCE_FILES = filter_source_files_from_defines(
{
"polygon.cpp": "LD2450_POLYGON_ZONE_COUNT",
"polygon_zone.cpp": "LD2450_POLYGON_ZONE_COUNT",
}
)
async def to_code(config: ConfigType) -> None:
var = cg.new_Pvariable(config[CONF_ID])
esphome/components/ld2450/binary_sensor.py
+35 -1
View File
@@ -2,6 +2,7 @@ import esphome.codegen as cg
from esphome.components import binary_sensor
import esphome.config_validation as cv
from esphome.const import (
CONF_FILTERS,
CONF_HAS_MOVING_TARGET,
CONF_HAS_STILL_TARGET,
CONF_HAS_TARGET,
@@ -11,7 +12,14 @@ from esphome.const import (
)
from esphome.types import ConfigType
from . import CONF_LD2450_ID, LD2450Component
from . import ( # noqa: F401 pylint: disable=unused-import
CONF_LD2450_ID,
CONF_POLYGON_ZONE_ID,
CONF_POLYGON_ZONES,
FILTER_SOURCE_FILES,
LD2450Component,
)
from .text import PolygonZone
DEPENDENCIES = ["ld2450"]
@@ -19,6 +27,18 @@ ICON_MEDITATION = "mdi:meditation"
ICON_SHIELD_ACCOUNT = "mdi:shield-account"
ICON_TARGET_ACCOUNT = "mdi:target-account"
def _reject_filters(config: ConfigType) -> ConfigType:
# Clearing a polygon makes presence unknown without going through filters, so they would get out of step
if CONF_FILTERS in config:
raise cv.Invalid(
"Polygon zone presence binary sensors can't have filters; use the ld2450 number platform's "
"'presence_timeout' to keep presence on after targets leave",
path=[CONF_FILTERS],
)
return config
CONFIG_SCHEMA = {
cv.GenerateID(CONF_ID): cv.declare_id(cg.EntityBase),
cv.GenerateID(CONF_LD2450_ID): cv.use_id(LD2450Component),
@@ -37,6 +57,16 @@ CONFIG_SCHEMA = {
filters=[{"settle": cv.TimePeriod(milliseconds=1000)}],
icon=ICON_MEDITATION,
),
cv.Optional(CONF_POLYGON_ZONES): cv.ensure_list(
binary_sensor.binary_sensor_schema(
device_class=DEVICE_CLASS_OCCUPANCY,
).extend(
{
cv.Required(CONF_POLYGON_ZONE_ID): cv.use_id(PolygonZone),
}
),
_reject_filters,
),
}
@@ -46,3 +76,7 @@ async def to_code(config: ConfigType) -> None:
await binary_sensors(CONF_HAS_TARGET, hub.set_target_binary_sensor)
await binary_sensors(CONF_HAS_MOVING_TARGET, hub.set_moving_target_binary_sensor)
await binary_sensors(CONF_HAS_STILL_TARGET, hub.set_still_target_binary_sensor)
for presence_conf in config.get(CONF_POLYGON_ZONES, []):
presence = await binary_sensor.new_binary_sensor(presence_conf)
zone = await cg.get_variable(presence_conf[CONF_POLYGON_ZONE_ID])
cg.add(zone.set_presence_binary_sensor(presence))
esphome/components/ld2450/ld2450.cpp
+31 -3
View File
@@ -179,6 +179,11 @@ void LD2450Component::setup() {
this->pref_ = this->presence_timeout_number_->make_entity_preference<float>();
this->set_presence_timeout();
}
#endif
#ifdef LD2450_POLYGON_ZONE_COUNT
for (PolygonZone *zone : this->polygon_zones_) {
zone->setup();
}
#endif
this->restart_and_read_all_info();
}
@@ -265,6 +270,12 @@ void LD2450Component::dump_config() {
LOG_BUTTON(" ", "FactoryReset", this->factory_reset_button_);
LOG_BUTTON(" ", "Restart", this->restart_button_);
#endif
#ifdef LD2450_POLYGON_ZONE_COUNT
ESP_LOGCONFIG(TAG, "Polygon Zones:");
for (PolygonZone *zone : this->polygon_zones_) {
zone->dump_config();
}
#endif
}
void LD2450Component::loop() {
@@ -465,15 +476,15 @@ void LD2450Component::handle_periodic_data_() {
#if defined(USE_BINARY_SENSOR) || defined(USE_SENSOR) || defined(USE_TEXT_SENSOR)
// Loop thru targets
for (index = 0; index < MAX_TARGETS; index++) {
#ifdef USE_SENSOR
// X
start = TARGET_X + index * 8;
is_moving = false;
// tx is used for further calculations, so always needs to be populated
// tx and ty are used for further calculations, so always need to be populated
tx = ld2450::decode_coordinate(this->buffer_data_[start], this->buffer_data_[start + 1]);
// Y
start = TARGET_Y + index * 8;
ty = ld2450::decode_coordinate(this->buffer_data_[start], this->buffer_data_[start + 1]);
#ifdef USE_SENSOR
// RESOLUTION
start = TARGET_RESOLUTION + index * 8;
res = (this->buffer_data_[start + 1] << 8) | this->buffer_data_[start];
@@ -542,6 +553,23 @@ void LD2450Component::handle_periodic_data_() {
still_target_count = target_count - moving_target_count;
#endif
#ifdef LD2450_POLYGON_ZONE_COUNT
const uint32_t now = App.get_loop_component_start_time();
for (PolygonZone *zone : this->polygon_zones_) {
if (!zone->has_polygon())
continue;
bool target_inside = false;
for (const Target &target : this->target_info_) {
// Untracked targets are stored as (0, 0)
if ((target.x != 0 || target.y != 0) && zone->contains(target.x, target.y)) {
target_inside = true;
break;
}
}
zone->update(target_inside, now, this->get_timeout_status_(zone->get_last_inside_ms()));
}
#endif
#ifdef USE_SENSOR
// Loop thru zones
uint8_t zone_still_targets = 0;
@@ -602,7 +630,7 @@ void LD2450Component::handle_periodic_data_() {
}
}
#endif
#ifdef USE_SENSOR
#ifdef USE_BINARY_SENSOR
// For presence timeout check
if (target_count > 0) {
this->presence_millis_ = App.get_loop_component_start_time();
esphome/components/ld2450/ld2450.h
+12 -1
View File
@@ -24,6 +24,10 @@
#include "esphome/components/binary_sensor/binary_sensor.h"
#endif
#ifdef LD2450_POLYGON_ZONE_COUNT
#include "polygon_zone.h"
#endif
#include "esphome/components/ld24xx/ld24xx.h"
#include "esphome/components/uart/uart.h"
#include "esphome/core/helpers.h"
@@ -137,6 +141,9 @@ class LD2450Component : public Component, public uart::UARTDevice {
void set_zone_target_count_sensor(uint8_t zone, sensor::Sensor *s);
void set_zone_still_target_count_sensor(uint8_t zone, sensor::Sensor *s);
void set_zone_moving_target_count_sensor(uint8_t zone, sensor::Sensor *s);
#endif
#ifdef LD2450_POLYGON_ZONE_COUNT
void register_polygon_zone(PolygonZone *zone) { this->polygon_zones_.push_back(zone); }
#endif
void reset_radar_zone();
void set_radar_zone(int32_t zone_type, int32_t zone1_x1, int32_t zone1_y1, int32_t zone1_x2, int32_t zone1_y2,
@@ -167,7 +174,7 @@ class LD2450Component : public Component, public uart::UARTDevice {
uint32_t presence_millis_ = 0;
uint32_t still_presence_millis_ = 0;
uint32_t moving_presence_millis_ = 0;
uint32_t timeout_ = 5;
uint32_t timeout_ = DEFAULT_PRESENCE_TIMEOUT * 1000; // presence_timeout, in ms
uint8_t buffer_data_[MAX_LINE_LENGTH];
uint8_t mac_address_[MAC_ADDRESS_SIZE] = {0, 0, 0, 0, 0, 0};
uint8_t version_[6] = {0, 0, 0, 0, 0, 0};
@@ -200,6 +207,10 @@ class LD2450Component : public Component, public uart::UARTDevice {
std::array<text_sensor::TextSensor *, MAX_TARGETS> direction_text_sensors_{};
#endif
#ifdef LD2450_POLYGON_ZONE_COUNT
StaticVector<PolygonZone *, LD2450_POLYGON_ZONE_COUNT> polygon_zones_;
#endif
LazyCallbackManager<void()> data_callback_;
};
esphome/components/ld2450/polygon.cpp
+123
View File
@@ -0,0 +1,123 @@
#include "polygon.h"
#include "esphome/core/helpers.h"
#include <cstdio>
namespace esphome::ld2450 {
namespace {
class Scanner {
public:
Scanner(const char *str, size_t len) : pos_(str), end_(str + len) {}
bool at_end() const { return this->pos_ == this->end_; }
void skip_spaces() {
while (this->pos_ != this->end_ && *this->pos_ == ' ')
this->pos_++;
}
bool consume(char c) {
this->skip_spaces();
if (this->pos_ != this->end_ && *this->pos_ == c) {
this->pos_++;
return true;
}
return false;
}
/// Read an integer in [min, max]; returns false if there is none or it is out of range.
bool read_int(int16_t min, int16_t max, int16_t &out) {
bool negative = this->consume('-');
int32_t value = 0;
uint8_t digits = 0;
while (this->pos_ != this->end_ && *this->pos_ >= '0' && *this->pos_ <= '9') {
// Six digits is already beyond any valid coordinate; stop before value can overflow
if (++digits > 6)
return false;
value = value * 10 + (*this->pos_ - '0');
this->pos_++;
}
if (digits == 0)
return false;
if (negative)
value = -value;
if (value < min || value > max)
return false;
out = static_cast<int16_t>(value);
return true;
}
protected:
const char *pos_;
const char *end_;
};
} // namespace
bool Polygon::is_valid() const {
if (this->count == 0)
return true;
if (this->count < MIN_POLYGON_POINTS || this->count > MAX_POLYGON_POINTS)
return false;
for (uint8_t i = 0; i < this->count; i++) {
const Point &p = this->points[i];
if (p.x < POLYGON_MIN_X || p.x > POLYGON_MAX_X || p.y < POLYGON_MIN_Y || p.y > POLYGON_MAX_Y)
return false;
}
return true;
}
bool Polygon::contains(int16_t x, int16_t y) const {
bool inside = false;
for (uint8_t i = 0, j = this->count - 1; i < this->count; j = i++) {
const Point &a = this->points[i];
const Point &b = this->points[j];
if ((a.y > y) == (b.y > y))
continue;
// Does the edge a-b cross the ray going from (x, y) towards +x? Compare
// x - a.x < (y - a.y) * (b.x - a.x) / (b.y - a.y) without division; values fit in 32 bits.
const int32_t dy = static_cast<int32_t>(b.y) - a.y;
const int32_t lhs = (static_cast<int32_t>(x) - a.x) * dy;
const int32_t rhs = (static_cast<int32_t>(y) - a.y) * (static_cast<int32_t>(b.x) - a.x);
if (dy > 0 ? lhs < rhs : lhs > rhs)
inside = !inside;
}
return inside;
}
bool Polygon::parse(const char *str, size_t len) {
Polygon result;
Scanner scanner(str, len);
scanner.skip_spaces();
while (!scanner.at_end()) {
if (result.count == MAX_POLYGON_POINTS)
return false;
Point &p = result.points[result.count];
if (!scanner.read_int(POLYGON_MIN_X, POLYGON_MAX_X, p.x) || !scanner.consume(',') ||
!scanner.read_int(POLYGON_MIN_Y, POLYGON_MAX_Y, p.y))
return false;
result.count++;
bool separator = scanner.consume(';');
scanner.skip_spaces();
if (!separator && !scanner.at_end())
return false;
}
if (!result.is_valid())
return false;
*this = result;
return true;
}
void Polygon::format(char *buf) const {
size_t pos = 0;
buf[0] = '\0';
for (uint8_t i = 0; i < this->count; i++) {
pos = buf_append_printf(buf, POLYGON_STR_SIZE, pos, "%s%d,%d", i == 0 ? "" : ";", this->points[i].x,
this->points[i].y);
}
}
} // namespace esphome::ld2450
esphome/components/ld2450/polygon.h
+53
View File
@@ -0,0 +1,53 @@
#pragma once
#include <array>
#include <cstddef>
#include <cstdint>
namespace esphome::ld2450 {
static constexpr uint8_t MIN_POLYGON_POINTS = 3;
// The most points whose text always fits Home Assistant's 255-character limit: 23 x "-4860,7560" + 22 x ';' = 252
static constexpr uint8_t MAX_POLYGON_POINTS = 23;
// Coordinate limits of the radar detection area, in mm
static constexpr int16_t POLYGON_MIN_X = -4860;
static constexpr int16_t POLYGON_MAX_X = 4860;
static constexpr int16_t POLYGON_MIN_Y = 0;
static constexpr int16_t POLYGON_MAX_Y = 7560;
// Longest point is "-4860,7560;" (11 characters), plus the null terminator
static constexpr size_t POLYGON_STR_SIZE = MAX_POLYGON_POINTS * 11 + 1;
struct Point {
int16_t x;
int16_t y;
};
/// A polygon in radar coordinates. Trivially copyable so it can be saved to preferences as-is.
struct Polygon {
std::array<Point, MAX_POLYGON_POINTS> points{};
uint8_t count{0};
/// True when the polygon has no points, which disables the zone.
bool empty() const { return this->count == 0; }
/// True when the polygon is empty, or has a valid number of points that are all inside the radar area.
bool is_valid() const;
/// Return whether the point is inside the polygon (even-odd rule). Always false for an empty polygon.
bool contains(int16_t x, int16_t y) const;
/** Parse "x,y;x,y;..." (mm) into this polygon.
*
* Spaces around numbers and a trailing ';' are allowed. An empty or all-space string gives an empty polygon.
* @return false if the text is invalid; the polygon is then left unchanged.
*/
bool parse(const char *str, size_t len);
/// Write the polygon as "x,y;x,y;..." into buf, which must hold at least POLYGON_STR_SIZE bytes.
void format(char *buf) const;
};
// text.py counts on this size
static_assert(sizeof(Polygon) == 94, "Polygon layout changed; update POLYGON_SAVED_BYTES in text.py");
} // namespace esphome::ld2450
esphome/components/ld2450/polygon_zone.cpp
+60
View File
@@ -0,0 +1,60 @@
#include "polygon_zone.h"
#ifdef LD2450_POLYGON_ZONE_COUNT
#include "esphome/core/log.h"
namespace esphome::ld2450 {
ESPHOME_LOG_TAG(TAG, "ld2450.polygon_zone");
void PolygonZone::setup() {
// Bump the version when Polygon's layout changes
this->pref_ = this->make_entity_preference<Polygon>(1);
Polygon saved;
if (this->pref_.load(&saved)) {
if (saved.is_valid()) {
this->polygon_ = saved;
} else {
ESP_LOGW(TAG, "'%s': saved polygon is invalid; zone disabled", this->get_name().c_str());
}
}
// Presence stays unknown until the first radar frame
this->publish_polygon_();
}
void PolygonZone::dump_config() {
char polygon_s[POLYGON_STR_SIZE];
this->polygon_.format(polygon_s);
LOG_TEXT(" ", "PolygonZone", this);
ESP_LOGCONFIG(TAG, " Polygon: '%s'", polygon_s);
LOG_BINARY_SENSOR(" ", "Presence", this->presence_);
}
void PolygonZone::control(const std::string &value) {
Polygon polygon;
if (!polygon.parse(value.c_str(), value.size())) {
ESP_LOGW(TAG, "'%s': invalid polygon '%s'; expected 'x,y;x,y;...' with %u to %u points", this->get_name().c_str(),
value.c_str(), MIN_POLYGON_POINTS, MAX_POLYGON_POINTS);
} else if (!this->pref_.save(&polygon)) {
// On ESP8266, the space for saved settings is small and can run out with several zones
ESP_LOGW(TAG, "'%s': could not save the polygon; keeping the previous one", this->get_name().c_str());
} else {
this->polygon_ = polygon;
// A target seen inside the previous polygon must not keep the new one on
this->last_inside_ms_ = 0;
if (this->polygon_.empty())
this->presence_->invalidate_state();
}
// On error this publishes the previous polygon again, so the frontend shows the value in use
this->publish_polygon_();
}
void PolygonZone::publish_polygon_() {
char polygon_s[POLYGON_STR_SIZE];
this->polygon_.format(polygon_s);
this->publish_state(polygon_s);
}
} // namespace esphome::ld2450
#endif // LD2450_POLYGON_ZONE_COUNT
esphome/components/ld2450/polygon_zone.h
+63
View File
@@ -0,0 +1,63 @@
#pragma once
#include "esphome/core/defines.h"
#ifdef LD2450_POLYGON_ZONE_COUNT
#include "esphome/components/binary_sensor/binary_sensor.h"
#include "esphome/components/text/text.h"
#include "esphome/core/preferences.h"
#include "polygon.h"
namespace esphome::ld2450 {
/** A zone with any polygon shape. The text state holds the polygon, so it can be edited and read back from HA.
*
* The polygon is in the radar's raw coordinates (mm), and targets are checked against it as decoded from each
* radar frame, before any sensor filter. Filters on the target X/Y sensors change only what they publish, so a
* filter that changes their values (multiply, offset, lambda, ...) puts them in a different frame from the polygon.
*
* Like the hub's has_target binary sensors, presence stays on until presence_timeout has passed since a target was
* last inside the zone.
*/
class PolygonZone : public text::Text {
public:
/// Config validation guarantees every zone gets exactly one presence binary sensor.
void set_presence_binary_sensor(binary_sensor::BinarySensor *presence) { this->presence_ = presence; }
/// Load the saved polygon and publish it. Called from the hub's setup().
void setup();
void dump_config();
/** Publish presence for one radar frame. A zone without a polygon keeps presence unknown.
*
* @param target_inside whether any target is inside the zone in this frame.
* @param now the frame's time, in ms.
* @param timed_out whether presence_timeout has passed since get_last_inside_ms().
*/
void update(bool target_inside, uint32_t now, bool timed_out) {
if (this->polygon_.empty())
return;
if (target_inside) {
this->last_inside_ms_ = now;
this->presence_->publish_state(true);
} else if (timed_out) {
this->presence_->publish_state(false);
}
}
/// When a target was last inside the zone, in ms; 0 if never since the polygon was set.
uint32_t get_last_inside_ms() const { return this->last_inside_ms_; }
bool contains(int16_t x, int16_t y) const { return this->polygon_.contains(x, y); }
bool has_polygon() const { return !this->polygon_.empty(); }
protected:
void control(const std::string &value) override;
void publish_polygon_();
binary_sensor::BinarySensor *presence_{nullptr};
Polygon polygon_;
ESPPreferenceObject pref_;
uint32_t last_inside_ms_{0};
};
} // namespace esphome::ld2450
#endif // LD2450_POLYGON_ZONE_COUNT
esphome/components/ld2450/sensor.py
+1 -1
View File
@@ -17,7 +17,7 @@ from esphome.const import (
)
from esphome.types import ConfigType
from . import CONF_LD2450_ID, LD2450Component
from . import CONF_LD2450_ID, FILTER_SOURCE_FILES, LD2450Component # noqa: F401 pylint: disable=unused-import
DEPENDENCIES = ["ld2450"]
esphome/components/ld2450/text.py
+120
View File
@@ -0,0 +1,120 @@
from esphome import preferences
import esphome.codegen as cg
from esphome.components import text
import esphome.config_validation as cv
from esphome.const import CONF_ID, CONF_PLATFORM, ENTITY_CATEGORY_CONFIG
import esphome.final_validate as fv
from esphome.types import ConfigType
from . import ( # noqa: F401 pylint: disable=unused-import
CONF_LD2450_ID,
CONF_POLYGON_ZONE_ID,
CONF_POLYGON_ZONES,
DOMAIN,
FILTER_SOURCE_FILES,
LD2450Component,
ld2450_ns,
)
DEPENDENCIES = ["ld2450"]
ICON_VECTOR_POLYGON = "mdi:vector-polygon"
# Must match MIN_POLYGON_POINTS and MAX_POLYGON_POINTS in polygon.h
MIN_POLYGON_POINTS = 3
MAX_POLYGON_POINTS = 23
# Matches what Polygon::parse() accepts, apart from coordinate ranges, which only the device checks:
# empty, or MIN_POLYGON_POINTS to MAX_POLYGON_POINTS "x,y" points separated by ';'. HA validates input with it.
# No character classes: the HA frontend compiles it with the JS "v" flag, HA core with Python re.
POLYGON_PATTERN = (
r"^ *$|^ *-?\d{1,4} *, *\d{1,4} *"
rf"(; *-?\d{{1,4}} *, *\d{{1,4}} *){{{MIN_POLYGON_POINTS - 1},{MAX_POLYGON_POINTS - 1}}};? *$"
)
# Home Assistant's limit for a state; the longest polygon the device publishes is 252 characters
POLYGON_MAX_LENGTH = 255
# sizeof(Polygon), pinned by a static_assert in polygon.h
POLYGON_SAVED_BYTES = 94
PolygonZone = ld2450_ns.class_("PolygonZone", text.Text)
_request_polygon_zone_slot = cg.slot_counter("LD2450_POLYGON_ZONE_COUNT")
def _require_manual_id(config: ConfigType) -> ConfigType:
if not config[CONF_ID].is_manual:
raise cv.Invalid(
f"'{CONF_ID}' is required, so the zone's presence binary sensor can refer to it",
path=[CONF_ID],
)
return config
CONFIG_SCHEMA = cv.Schema(
{
cv.GenerateID(CONF_LD2450_ID): cv.use_id(LD2450Component),
# Polygons are in raw radar mm, checked before sensor filters (polygon_zone.h)
cv.Required(CONF_POLYGON_ZONES): cv.ensure_list(
text.text_schema(
PolygonZone,
entity_category=ENTITY_CATEGORY_CONFIG,
icon=ICON_VECTOR_POLYGON,
mode="TEXT",
),
_require_manual_id,
),
}
)
def _final_validate(config: ConfigType) -> None:
"""Each polygon zone must have exactly one presence binary sensor, on the same LD2450."""
full_config = fv.full_config.get()
# Polygon zone id -> the LD2450 id of each presence binary sensor that uses it
references: dict[str, list[str]] = {}
for binary_sensor_conf in full_config.get("binary_sensor", []):
if binary_sensor_conf.get(CONF_PLATFORM) != "ld2450":
continue
hub_id = binary_sensor_conf[CONF_LD2450_ID].id
for presence_conf in binary_sensor_conf.get(CONF_POLYGON_ZONES, []):
zone_id = presence_conf[CONF_POLYGON_ZONE_ID].id
references.setdefault(zone_id, []).append(hub_id)
zone_hub_id = config[CONF_LD2450_ID].id
for index, zone_conf in enumerate(config[CONF_POLYGON_ZONES]):
zone_id = zone_conf[CONF_ID].id
hub_ids = references.get(zone_id, [])
if len(hub_ids) != 1:
problem = (
"has no presence binary sensor"
if not hub_ids
else "is used by more than one presence binary sensor"
)
raise cv.Invalid(
f"Polygon zone '{zone_id}' {problem}; each zone needs exactly one, under "
f"'binary_sensor: - platform: ld2450' -> '{CONF_POLYGON_ZONES}'",
path=[CONF_POLYGON_ZONES, index],
)
if hub_ids[0] != zone_hub_id:
raise cv.Invalid(
f"Polygon zone '{zone_id}' belongs to LD2450 '{zone_hub_id}', but its presence "
f"binary sensor is under LD2450 '{hub_ids[0]}'; use the same '{CONF_LD2450_ID}'",
path=[CONF_POLYGON_ZONES, index],
)
FINAL_VALIDATE_SCHEMA = _final_validate
async def to_code(config: ConfigType) -> None:
hub = await cg.get_variable(config[CONF_LD2450_ID])
# Each zone saves its polygon; the ESP8266 checks that every component's saved values fit
preferences.reserve_preferences(
len(config[CONF_POLYGON_ZONES]), POLYGON_SAVED_BYTES
)
for zone_conf in config[CONF_POLYGON_ZONES]:
zone = await text.new_text(
zone_conf, max_length=POLYGON_MAX_LENGTH, pattern=POLYGON_PATTERN
)
_request_polygon_zone_slot(str(hub))
cg.add(hub.register_polygon_zone(zone))
esphome/components/ld2450/text_sensor.py
+1 -1
View File
@@ -14,7 +14,7 @@ from esphome.const import (
)
from esphome.types import ConfigType
from . import CONF_LD2450_ID, LD2450Component
from . import CONF_LD2450_ID, FILTER_SOURCE_FILES, LD2450Component # noqa: F401 pylint: disable=unused-import
DEPENDENCIES = ["ld2450"]
esphome/core/__init__.py
+11 -2
View File
@@ -442,6 +442,16 @@ class ID:
)
# Replaced by write_cpp with a #line back into main.cpp
LINE_DIRECTIVE_RESET = "// ESPHOME_LINE_DIRECTIVE_RESET"
def line_directive(document: str, line: int) -> str:
"""A #line directive naming the 1-based ``line`` of ``document``."""
escaped = document.replace("\\", "\\\\")
return f'#line {line} "{escaped}"'
class DocumentLocation:
def __init__(self, document: str, line: int, column: int):
self.document: str = document
@@ -457,8 +467,7 @@ class DocumentLocation:
@property
def as_line_directive(self):
document_path = str(self.document).replace("\\", "\\\\")
return f'#line {self.line + 1} "{document_path}"'
return line_directive(str(self.document), self.line + 1)
class DocumentRange:
esphome/core/defines.h
+1
View File
@@ -157,6 +157,7 @@
#define MDNS_SERVICE_COUNT 3
#define USE_MDNS_DYNAMIC_TXT
#define MDNS_DYNAMIC_TXT_COUNT 2
#define LD2450_POLYGON_ZONE_COUNT 2
#define MICRONOVA_LISTENER_COUNT 1
#define USE_MICRONOVA_WRITER
#define MK2PVROUTER_LISTENER_COUNT 1
esphome/cpp_generator.py
+8 -4
View File
@@ -8,6 +8,7 @@ from typing import Any
from esphome.core import (
CORE,
ID,
LINE_DIRECTIVE_RESET,
Define,
EnumValue,
EsphomeError,
@@ -214,9 +215,13 @@ class LambdaExpression(Expression):
if self.return_type is not None:
cpp += f" -> {self.return_type}"
cpp += " {\n"
body = f"{self.content}\n"
if self.source is not None:
cpp += f"{self.source.as_line_directive}\n"
cpp += f"{self.content}\n}}"
body = f"{self.source.as_line_directive}\n{body}"
# A #line in the body, the source's or lvgl's own marks, needs the reset after it
if "#line " in body:
body += f"{LINE_DIRECTIVE_RESET}\n"
cpp += f"{body}}}"
return indent_all_but_first_and_last(cpp)
@property
@@ -946,8 +951,7 @@ async def process_lambda(
parts[i * 3 + 2] = ""
if isinstance(value, ESPHomeDataBase) and value.esp_range is not None:
location = value.esp_range.start_mark
location.line += value.content_offset
location = value.content_start
else:
location = None
return LambdaExpression(parts, parameters, capture, return_type, location)
esphome/writer.py
+21 -2
View File
@@ -15,7 +15,7 @@ from esphome.const import (
SOURCE_FILE_EXTENSIONS,
__version__,
)
from esphome.core import CORE, EsphomeError
from esphome.core import CORE, LINE_DIRECTIVE_RESET, EsphomeError, line_directive
from esphome.helpers import (
copy_file_if_changed,
cpp_string_escape,
@@ -514,6 +514,25 @@ const char ESPHOME_COMMENT_STR[] = "{escaped_comment}";
"""
def _reset_line_directives(text: str, path: Path) -> str:
"""Point the code after each lambda back at its own line of *path*.
A lambda's #line directive points the compiler at the YAML; without a reset
every later line of main.cpp would be reported against the YAML too.
"""
if LINE_DIRECTIVE_RESET not in text:
return text
document = str(path.absolute())
lines = text.split("\n")
for index, line in enumerate(lines):
if line.endswith(LINE_DIRECTIVE_RESET):
# The directive numbers the line after it: index + 2, counted from 1
lines[index] = line.replace(
LINE_DIRECTIVE_RESET, line_directive(document, index + 2)
)
return "\n".join(lines)
def write_cpp(code_s):
path = CORE.relative_src_path("main.cpp")
if path.is_file():
@@ -540,7 +559,7 @@ def write_cpp(code_s):
f"{code_format[1] + CPP_AUTO_GENERATE_BEGIN}\n{code_s}{CPP_AUTO_GENERATE_END}"
)
full_file += code_format[2]
write_file_if_changed(path, full_file)
write_file_if_changed(path, _reset_line_directives(full_file, path))
def clean_cmake_cache():
esphome/yaml_util.py
+12 -5
View File
@@ -29,6 +29,7 @@ from esphome.config_helpers import Extend, Remove
from esphome.const import CONF_DEFAULTS
from esphome.core import (
CORE,
DocumentLocation,
DocumentRange,
EsphomeError,
Lambda,
@@ -142,14 +143,20 @@ class ESPHomeDataBase:
def content_offset(self):
return getattr(self, "_content_offset", 0)
@property
def content_start(self) -> DocumentLocation:
"""Where the value's own text starts: the line after the marker for a block scalar."""
start = self._esp_range.start_mark
return DocumentLocation(
start.document, start.line + self.content_offset, start.column
)
def from_node(self, node):
# pylint: disable=attribute-defined-outside-init
self._esp_range = DocumentRange.from_marks(node.start_mark, node.end_mark)
if (
isinstance(node, yaml.ScalarNode)
and node.style is not None
and node.style in "|>"
):
# Only block scalars (| and >) start their content on the next line. A plain
# scalar's style is "", which `in "|>"` would also match.
if isinstance(node, yaml.ScalarNode) and node.style in ("|", ">"):
self._content_offset = 1
def from_database(self, database):
platformio.ini
+2
View File
@@ -599,6 +599,8 @@ build_flags =
${flags:clangtidy.build_flags}
-DUSE_BK72XX
-DUSE_LIBRETINY_VARIANT_BK7231N
; The libretiny component adds this at code generation so FreeRTOSQueue can use xQueueCreateStatic
-DconfigSUPPORT_STATIC_ALLOCATION=1
build_unflags =
${common.build_unflags}
tests/component_tests/ld2450/__init__.py
View File
tests/component_tests/ld2450/config/esp8266_zones.yaml
+31
View File
@@ -0,0 +1,31 @@
esphome:
name: test
esp8266:
board: d1_mini
uart:
tx_pin: GPIO0
rx_pin: GPIO2
baud_rate: 256000
ld2450:
id: radar
text:
- platform: ld2450
ld2450_id: radar
polygon_zones:
- id: couch
name: Couch
- id: desk
name: Desk
binary_sensor:
- platform: ld2450
ld2450_id: radar
polygon_zones:
- polygon_zone_id: couch
name: Couch Occupied
- polygon_zone_id: desk
name: Desk Occupied
tests/component_tests/ld2450/test_polygon_pattern.py
+64
View File
@@ -0,0 +1,64 @@
"""Tests for the polygon zone text pattern and maximum length.
Home Assistant validates polygon input with the pattern before it reaches the device, so the pattern must accept
what Polygon::parse() accepts. Coordinate ranges are the one exception: only the device checks those.
"""
from __future__ import annotations
import re
import pytest
from esphome.components.ld2450.text import (
MAX_POLYGON_POINTS,
POLYGON_MAX_LENGTH,
POLYGON_PATTERN,
)
# The same inputs as the C++ parser tests (ld2450_polygon.cpp), without out-of-range values
ACCEPTED = [
"",
" ",
"-1000,500;1000,500;0,3000",
" -1000 , 500 ; 1000,500; 0,3000; ",
"-4860,0;4860,0;4860,7560",
]
REJECTED = [
"0,0",
"0,0;100,100",
"junk",
"1,2;3",
"1,2;3,4;5,",
"1;2;3",
"1,2 3,4 5,6",
"1,2;;3,4;5,6",
"--1,2;3,4;5,6",
"1,2;3,4;5,6x",
"1,2;\t3,4;5,6",
]
@pytest.mark.parametrize("value", ACCEPTED)
def test_pattern_accepts_valid_polygons(value: str) -> None:
assert re.fullmatch(POLYGON_PATTERN, value)
@pytest.mark.parametrize("value", REJECTED)
def test_pattern_rejects_invalid_polygons(value: str) -> None:
assert not re.fullmatch(POLYGON_PATTERN, value)
def test_point_count_limits() -> None:
assert re.fullmatch(POLYGON_PATTERN, ";".join(["0,0"] * MAX_POLYGON_POINTS))
assert not re.fullmatch(
POLYGON_PATTERN, ";".join(["0,0"] * (MAX_POLYGON_POINTS + 1))
)
def test_longest_polygon_fits_max_length() -> None:
longest = ";".join(["-4860,7560"] * MAX_POLYGON_POINTS)
assert re.fullmatch(POLYGON_PATTERN, longest)
assert len(longest) <= POLYGON_MAX_LENGTH
# MAX_POLYGON_POINTS is the most that always fits
assert len(longest + ";-4860,7560") > POLYGON_MAX_LENGTH
tests/component_tests/ld2450/test_polygon_zone_presence.py
+45
View File
@@ -0,0 +1,45 @@
"""Tests for the ld2450 polygon zone presence binary sensor schema."""
from __future__ import annotations
import pytest
from esphome.components.ld2450 import CONF_POLYGON_ZONE_ID, CONF_POLYGON_ZONES
from esphome.components.ld2450.binary_sensor import (
CONFIG_SCHEMA as BINARY_SENSOR_CONFIG_SCHEMA,
)
import esphome.config_validation as cv
from esphome.const import CONF_FILTERS, CONF_NAME, PlatformFramework
from esphome.types import ConfigType
from tests.component_tests.types import SetCoreConfigCallable
def _presence(**extra: object) -> ConfigType:
return {
CONF_POLYGON_ZONES: [
{CONF_POLYGON_ZONE_ID: "couch_zone", CONF_NAME: "Couch Occupied", **extra}
]
}
def test_presence_without_filters_passes(
set_core_config: SetCoreConfigCallable,
) -> None:
set_core_config(PlatformFramework.ESP32_IDF)
cv.Schema(BINARY_SENSOR_CONFIG_SCHEMA)(_presence())
def test_presence_with_filters_is_rejected(
set_core_config: SetCoreConfigCallable,
) -> None:
"""Clearing a polygon bypasses filters, so they would get out of step with the zone."""
set_core_config(PlatformFramework.ESP32_IDF)
with pytest.raises(
cv.Invalid,
match=r"^Polygon zone presence binary sensors can't have filters; .*presence_timeout",
):
cv.Schema(BINARY_SENSOR_CONFIG_SCHEMA)(
_presence(**{CONF_FILTERS: [{"delayed_off": "1s"}]})
)
tests/component_tests/ld2450/test_polygon_zones.py
+163
View File
@@ -0,0 +1,163 @@
"""Tests for the ld2450 polygon zone validation.
Polygon zones are declared in the text platform and their presence sensors in the
binary sensor platform, so the pairing between the two is checked at final validation.
"""
from __future__ import annotations
from collections.abc import Callable
from pathlib import Path
import pytest
from esphome.components.esp8266.const import (
KEY_ESP8266,
KEY_PREFERENCES,
preference_words,
)
from esphome.components.ld2450 import CONF_POLYGON_ZONE_ID, CONF_POLYGON_ZONES
from esphome.components.ld2450.text import (
CONFIG_SCHEMA as TEXT_CONFIG_SCHEMA,
FINAL_VALIDATE_SCHEMA as TEXT_FINAL_VALIDATE_SCHEMA,
POLYGON_SAVED_BYTES,
)
import esphome.config_validation as cv
from esphome.const import CONF_ID, CONF_NAME, CONF_PLATFORM, PlatformFramework
from esphome.core import CORE, ID
from esphome.types import ConfigType
from tests.component_tests.types import SetCoreConfigCallable
HUB_ID = "ld2450_hub"
OTHER_HUB_ID = "ld2450_other"
def _text_config(*zone_ids: str) -> ConfigType:
"""A text platform config declaring the given polygon zones."""
return {
"ld2450_id": ID(HUB_ID, is_declaration=False, type="ld2450"),
CONF_POLYGON_ZONES: [
{CONF_ID: ID(zone_id, is_declaration=True), CONF_NAME: zone_id}
for zone_id in zone_ids
],
}
def _binary_sensors(*zone_ids: str, hub_id: str = HUB_ID) -> list[ConfigType]:
"""A binary sensor platform list with one presence sensor per given zone reference."""
return [
{
CONF_PLATFORM: "ld2450",
"ld2450_id": ID(hub_id, is_declaration=False, type="ld2450"),
CONF_POLYGON_ZONES: [
{CONF_POLYGON_ZONE_ID: ID(zone_id, is_declaration=False)}
for zone_id in zone_ids
],
}
]
def test_zones_paired_with_one_presence_pass(
set_core_config: SetCoreConfigCallable,
) -> None:
set_core_config(
PlatformFramework.ESP32_IDF,
full_config={"binary_sensor": _binary_sensors("couch", "desk")},
)
TEXT_FINAL_VALIDATE_SCHEMA(_text_config("couch", "desk"))
def test_zone_without_presence_is_rejected(
set_core_config: SetCoreConfigCallable,
) -> None:
set_core_config(
PlatformFramework.ESP32_IDF,
full_config={"binary_sensor": _binary_sensors("couch")},
)
with pytest.raises(
cv.Invalid,
match=r"^Polygon zone 'desk' has no presence binary sensor; .* @ data\['polygon_zones'\]\[1\]$",
):
TEXT_FINAL_VALIDATE_SCHEMA(_text_config("couch", "desk"))
def test_zone_without_any_binary_sensor_platform_is_rejected(
set_core_config: SetCoreConfigCallable,
) -> None:
set_core_config(PlatformFramework.ESP32_IDF, full_config={})
with pytest.raises(cv.Invalid, match="'couch' has no presence binary sensor"):
TEXT_FINAL_VALIDATE_SCHEMA(_text_config("couch"))
def test_zone_with_two_presences_is_rejected(
set_core_config: SetCoreConfigCallable,
) -> None:
"""Two references to one zone are rejected, even across binary sensor platform entries."""
set_core_config(
PlatformFramework.ESP32_IDF,
full_config={
"binary_sensor": _binary_sensors("couch") + _binary_sensors("couch")
},
)
with pytest.raises(
cv.Invalid,
match=r"^Polygon zone 'couch' is used by more than one presence binary sensor; ",
):
TEXT_FINAL_VALIDATE_SCHEMA(_text_config("couch"))
def test_other_platforms_are_ignored(
set_core_config: SetCoreConfigCallable,
) -> None:
"""Binary sensors of other platforms never count as presence for a zone."""
other = {CONF_PLATFORM: "template", CONF_POLYGON_ZONES: []}
set_core_config(
PlatformFramework.ESP32_IDF,
full_config={"binary_sensor": [other, *_binary_sensors("couch")]},
)
TEXT_FINAL_VALIDATE_SCHEMA(_text_config("couch"))
def test_zone_without_id_is_rejected(
set_core_config: SetCoreConfigCallable,
) -> None:
"""The presence sensor refers to the zone by id, so an automatic id is no use."""
set_core_config(PlatformFramework.ESP32_IDF)
with pytest.raises(cv.Invalid, match=r"^'id' is required, so the zone's presence"):
TEXT_CONFIG_SCHEMA({CONF_POLYGON_ZONES: [{CONF_NAME: "Couch Zone Polygon"}]})
def test_presence_on_another_ld2450_is_rejected(
set_core_config: SetCoreConfigCallable,
) -> None:
"""The zone's own LD2450 drives its presence, so the sensor must name the same one."""
set_core_config(
PlatformFramework.ESP32_IDF,
full_config={"binary_sensor": _binary_sensors("couch", hub_id=OTHER_HUB_ID)},
)
with pytest.raises(
cv.Invalid,
match=(
r"^Polygon zone 'couch' belongs to LD2450 'ld2450_hub', but its presence binary "
r"sensor is under LD2450 'ld2450_other'; .* @ data\['polygon_zones'\]\[0\]$"
),
):
TEXT_FINAL_VALIDATE_SCHEMA(_text_config("couch"))
def test_esp8266_zones_reserve_their_saved_space(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
"""Every zone reserves its polygon, so the ESP8266 can check that all saved values fit together."""
generate_main(component_config_path("esp8266_zones.yaml"))
words = preference_words(POLYGON_SAVED_BYTES)
assert CORE.data[KEY_ESP8266][KEY_PREFERENCES] == [(words, False)] * 2
tests/components/ld2450/__init__.py
+9
View File
@@ -0,0 +1,9 @@
from esphome.loader import FileResource
from tests.testing_helpers import ComponentManifestOverride
def override_manifest(manifest: ComponentManifestOverride) -> None:
# ld2450 is MULTI_CONF, so no codegen runs here to keep polygon.cpp; the tests need it
manifest.resources = manifest.resources + [
FileResource("esphome.components.ld2450", "polygon.cpp")
]
tests/components/ld2450/common.yaml
+14
View File
@@ -96,6 +96,20 @@ binary_sensor:
name: Moving Target
has_still_target:
name: Still Target
polygon_zones:
- polygon_zone_id: ld2450_couch_zone
name: Couch Occupied
- polygon_zone_id: ld2450_desk_zone
name: Desk Occupied
text:
- platform: ld2450
ld2450_id: ld2450_radar
polygon_zones:
- id: ld2450_couch_zone
name: Couch Zone Polygon
- id: ld2450_desk_zone
name: Desk Zone Polygon
switch:
- platform: ld2450
tests/components/ld2450/ld2450_polygon.cpp
+162
View File
@@ -0,0 +1,162 @@
#include <cstring>
#include <gtest/gtest.h>
#include "esphome/components/ld2450/polygon.h"
namespace esphome::ld2450::testing {
static bool parse(Polygon &polygon, const char *str) { return polygon.parse(str, strlen(str)); }
static std::string format(const Polygon &polygon) {
char buf[POLYGON_STR_SIZE];
polygon.format(buf);
return buf;
}
// --- Parsing ---
TEST(LD2450PolygonTest, ParseValid) {
Polygon polygon;
ASSERT_TRUE(parse(polygon, "-1000,500;1000,500;0,3000"));
ASSERT_EQ(polygon.count, 3);
EXPECT_EQ(polygon.points[0].x, -1000);
EXPECT_EQ(polygon.points[0].y, 500);
EXPECT_EQ(polygon.points[2].x, 0);
EXPECT_EQ(polygon.points[2].y, 3000);
EXPECT_EQ(format(polygon), "-1000,500;1000,500;0,3000");
}
TEST(LD2450PolygonTest, ParseAllowsSpacesAndTrailingSeparator) {
Polygon polygon;
ASSERT_TRUE(parse(polygon, " -1000 , 500 ; 1000,500; 0,3000; "));
EXPECT_EQ(format(polygon), "-1000,500;1000,500;0,3000");
}
TEST(LD2450PolygonTest, ParseEmptyDisablesZone) {
Polygon polygon;
ASSERT_TRUE(parse(polygon, "-1000,500;1000,500;0,3000"));
ASSERT_TRUE(parse(polygon, ""));
EXPECT_TRUE(polygon.empty());
ASSERT_TRUE(parse(polygon, "-1000,500;1000,500;0,3000"));
ASSERT_TRUE(parse(polygon, " "));
EXPECT_TRUE(polygon.empty());
EXPECT_EQ(format(polygon), "");
}
TEST(LD2450PolygonTest, ParseLimits) {
Polygon polygon;
EXPECT_TRUE(parse(polygon, "-4860,0;4860,0;4860,7560"));
EXPECT_FALSE(parse(polygon, "-4861,0;4860,0;4860,7560"));
EXPECT_FALSE(parse(polygon, "-4860,0;4861,0;4860,7560"));
EXPECT_FALSE(parse(polygon, "-4860,-1;4860,0;4860,7560"));
EXPECT_FALSE(parse(polygon, "-4860,0;4860,0;4860,7561"));
EXPECT_FALSE(parse(polygon, "-4860,0;4860,0;4860,99999999999"));
}
TEST(LD2450PolygonTest, ParsePointCount) {
Polygon polygon;
EXPECT_FALSE(parse(polygon, "0,0"));
EXPECT_FALSE(parse(polygon, "0,0;100,100"));
std::string text;
for (int i = 0; i < MAX_POLYGON_POINTS; i++) {
text += "-4860,7560;";
}
ASSERT_TRUE(parse(polygon, text.c_str()));
EXPECT_EQ(polygon.count, MAX_POLYGON_POINTS);
// The longest possible polygon fits in the format buffer
EXPECT_EQ(format(polygon) + ";", text);
text += "0,0";
EXPECT_FALSE(parse(polygon, text.c_str()));
}
TEST(LD2450PolygonTest, ParseInvalidKeepsPreviousPolygon) {
Polygon polygon;
ASSERT_TRUE(parse(polygon, "-1000,500;1000,500;0,3000"));
const char *invalid[] = {
"junk",
"1,2;3",
"1,2;3,4;5,",
"1;2;3",
"1,2 3,4 5,6",
"1,2;;3,4;5,6",
"--1,2;3,4;5,6",
"1,2;3,4;5,6x",
// Only spaces are allowed around numbers, as in the text entity's pattern
"1,2;\t3,4;5,6",
};
for (const char *str : invalid) {
EXPECT_FALSE(parse(polygon, str)) << str;
EXPECT_EQ(format(polygon), "-1000,500;1000,500;0,3000") << str;
}
}
TEST(LD2450PolygonTest, IsValid) {
Polygon polygon;
EXPECT_TRUE(polygon.is_valid());
polygon.count = 2;
EXPECT_FALSE(polygon.is_valid());
polygon.count = MAX_POLYGON_POINTS + 1;
EXPECT_FALSE(polygon.is_valid());
polygon.count = 3;
EXPECT_TRUE(polygon.is_valid());
polygon.points[1].y = -5;
EXPECT_FALSE(polygon.is_valid());
}
// --- Point in polygon ---
TEST(LD2450PolygonTest, ContainsEmpty) {
Polygon polygon;
EXPECT_FALSE(polygon.contains(0, 0));
EXPECT_FALSE(polygon.contains(100, 1000));
}
TEST(LD2450PolygonTest, ContainsSquare) {
Polygon polygon;
ASSERT_TRUE(parse(polygon, "-1000,1000;1000,1000;1000,3000;-1000,3000"));
EXPECT_TRUE(polygon.contains(0, 2000));
EXPECT_TRUE(polygon.contains(-999, 1001));
EXPECT_TRUE(polygon.contains(999, 2999));
EXPECT_FALSE(polygon.contains(0, 999));
EXPECT_FALSE(polygon.contains(0, 3001));
EXPECT_FALSE(polygon.contains(-1001, 2000));
EXPECT_FALSE(polygon.contains(1001, 2000));
}
TEST(LD2450PolygonTest, ContainsWithReversedWinding) {
Polygon clockwise;
Polygon counter_clockwise;
ASSERT_TRUE(parse(clockwise, "0,1000;2000,3000;-2000,3000"));
ASSERT_TRUE(parse(counter_clockwise, "-2000,3000;2000,3000;0,1000"));
for (int16_t x = -2500; x <= 2500; x += 250) {
for (int16_t y = 500; y <= 3500; y += 250) {
EXPECT_EQ(clockwise.contains(x, y), counter_clockwise.contains(x, y)) << x << "," << y;
}
}
EXPECT_TRUE(clockwise.contains(0, 2000));
EXPECT_FALSE(clockwise.contains(1500, 2000));
}
TEST(LD2450PolygonTest, ContainsConcave) {
// U shape: the notch between the two arms is outside
Polygon polygon;
ASSERT_TRUE(parse(polygon, "-3000,1000;3000,1000;3000,4000;1000,4000;1000,2000;-1000,2000;-1000,4000;-3000,4000"));
EXPECT_TRUE(polygon.contains(-2000, 3000));
EXPECT_TRUE(polygon.contains(2000, 3000));
EXPECT_TRUE(polygon.contains(0, 1500));
EXPECT_FALSE(polygon.contains(0, 3000));
EXPECT_FALSE(polygon.contains(0, 4500));
}
TEST(LD2450PolygonTest, ContainsFarOutsideRadarArea) {
// The radar reports 15-bit coordinates; the cross products must not overflow
Polygon polygon;
ASSERT_TRUE(parse(polygon, "-4860,0;4860,0;4860,7560;-4860,7560"));
EXPECT_FALSE(polygon.contains(32767, 32767));
EXPECT_FALSE(polygon.contains(-32767, -32767));
EXPECT_FALSE(polygon.contains(-32767, 3000));
EXPECT_TRUE(polygon.contains(0, 3000));
}
} // namespace esphome::ld2450::testing
tests/integration/fixtures/uart_mock_ld2450_has_target_only.yaml
+248
View File
@@ -0,0 +1,248 @@
esphome:
name: uart-mock-ld2450-has-target
host:
api:
batch_delay: 0ms # Disable batching to receive all state updates
logger:
level: VERBOSE
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
# Dummy uart entry to satisfy ld2450's DEPENDENCIES = ["uart"]
# The actual UART bus used is the uart_mock component below
uart:
baud_rate: 115200
port: /dev/null
uart_mock:
id: mock_uart
baud_rate: 256000
auto_start: false
responses:
# Generic ACK for any command, to unblock setup commands
- expect_tx: [0x04, 0x03, 0x02, 0x01]
inject_rx:
[
0xFD, 0xFC, 0xFB, 0xFA,
0x04, 0x00,
0xFF, 0x01,
0x00, 0x00,
0x04, 0x03, 0x02, 0x01,
]
# One frame every 100 ms, as the radar sends them. Coordinates are sign-magnitude, with the high bit set for
# positive values.
injections:
# Target 1 at (-500, 1000)
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0xF4, 0x01, 0xE8, 0x83, 0x00, 0x00, 0x40, 0x01,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
# No targets for 2 s: has_target must hold for the default 5 s presence_timeout
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
ld2450:
id: ld2450_dev
uart_id: mock_uart
# Only has_target: no sensor platform and no presence_timeout number
binary_sensor:
- platform: ld2450
ld2450_id: ld2450_dev
has_target:
name: "Has Target"
button:
- platform: template
name: "Start Scenario"
id: start_scenario_btn
on_press:
- lambda: 'id(mock_uart).start_scenario();'
tests/integration/fixtures/uart_mock_ld2450_polygon_zone_restore.yaml
+55
View File
@@ -0,0 +1,55 @@
esphome:
name: uart-mock-ld2450-zone-restore
host:
api:
batch_delay: 0ms
logger:
level: DEBUG
preferences:
flash_write_interval: 0s
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
# Dummy uart entry to satisfy ld2450's DEPENDENCIES = ["uart"]
# The actual UART bus used is the uart_mock component below
uart:
baud_rate: 115200
port: /dev/null
uart_mock:
id: mock_uart
baud_rate: 256000
responses:
# Generic ACK for any command, to unblock setup commands
- expect_tx: [0x04, 0x03, 0x02, 0x01]
inject_rx:
[
0xFD, 0xFC, 0xFB, 0xFA,
0x04, 0x00,
0xFF, 0x01,
0x00, 0x00,
0x04, 0x03, 0x02, 0x01,
]
ld2450:
id: ld2450_dev
uart_id: mock_uart
text:
- platform: ld2450
ld2450_id: ld2450_dev
polygon_zones:
- id: restore_zone
name: "Restore Zone Polygon"
binary_sensor:
- platform: ld2450
ld2450_id: ld2450_dev
polygon_zones:
- polygon_zone_id: restore_zone
name: "Restore Zone Presence"
tests/integration/fixtures/uart_mock_ld2450_polygon_zone_timeout.yaml
+300
View File
@@ -0,0 +1,300 @@
esphome:
name: uart-mock-ld2450-zone-timeout
host:
api:
batch_delay: 0ms # Disable batching to receive all state updates
logger:
level: VERBOSE
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
# Dummy uart entry to satisfy ld2450's DEPENDENCIES = ["uart"]
# The actual UART bus used is the uart_mock component below
uart:
baud_rate: 115200
port: /dev/null
uart_mock:
id: mock_uart
baud_rate: 256000
auto_start: false
responses:
# Generic ACK for any command, to unblock setup commands
- expect_tx: [0x04, 0x03, 0x02, 0x01]
inject_rx:
[
0xFD, 0xFC, 0xFB, 0xFA,
0x04, 0x00,
0xFF, 0x01,
0x00, 0x00,
0x04, 0x03, 0x02, 0x01,
]
# One frame every 100 ms, as the radar sends them. Target 1 is either at (-500, 1000), inside the zone, or at
# (300, 400), outside it. Coordinates are sign-magnitude, with the high bit set for positive values.
injections:
# Inside the zone
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0xF4, 0x01, 0xE8, 0x83, 0x00, 0x00, 0x40, 0x01,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
# Outside for one frame: radar noise that presence_timeout must hide
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x2C, 0x81, 0x90, 0x81, 0x00, 0x00, 0x64, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
# Inside again: the last inside frame
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0xF4, 0x01, 0xE8, 0x83, 0x00, 0x00, 0x40, 0x01,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
# Outside: presence clears once presence_timeout passes
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x2C, 0x81, 0x90, 0x81, 0x00, 0x00, 0x64, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
# Outside
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x2C, 0x81, 0x90, 0x81, 0x00, 0x00, 0x64, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
# Outside
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x2C, 0x81, 0x90, 0x81, 0x00, 0x00, 0x64, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
# Outside
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x2C, 0x81, 0x90, 0x81, 0x00, 0x00, 0x64, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
# Outside
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x2C, 0x81, 0x90, 0x81, 0x00, 0x00, 0x64, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
# Outside
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x2C, 0x81, 0x90, 0x81, 0x00, 0x00, 0x64, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
# Outside
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x2C, 0x81, 0x90, 0x81, 0x00, 0x00, 0x64, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
# Outside
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x2C, 0x81, 0x90, 0x81, 0x00, 0x00, 0x64, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
# Outside
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x2C, 0x81, 0x90, 0x81, 0x00, 0x00, 0x64, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
# Outside
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x2C, 0x81, 0x90, 0x81, 0x00, 0x00, 0x64, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
# Outside
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x2C, 0x81, 0x90, 0x81, 0x00, 0x00, 0x64, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
# Outside
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x2C, 0x81, 0x90, 0x81, 0x00, 0x00, 0x64, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
# Outside
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x2C, 0x81, 0x90, 0x81, 0x00, 0x00, 0x64, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
# Outside
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x2C, 0x81, 0x90, 0x81, 0x00, 0x00, 0x64, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
# Outside
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x2C, 0x81, 0x90, 0x81, 0x00, 0x00, 0x64, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
# Outside
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x2C, 0x81, 0x90, 0x81, 0x00, 0x00, 0x64, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
# Outside
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x2C, 0x81, 0x90, 0x81, 0x00, 0x00, 0x64, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
# Outside
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x2C, 0x81, 0x90, 0x81, 0x00, 0x00, 0x64, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
# Outside
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x2C, 0x81, 0x90, 0x81, 0x00, 0x00, 0x64, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
# Outside
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x2C, 0x81, 0x90, 0x81, 0x00, 0x00, 0x64, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
ld2450:
id: ld2450_dev
uart_id: mock_uart
text:
- platform: ld2450
ld2450_id: ld2450_dev
polygon_zones:
- id: hold_zone
name: "Hold Zone Polygon"
binary_sensor:
- platform: ld2450
ld2450_id: ld2450_dev
polygon_zones:
- polygon_zone_id: hold_zone
name: "Hold Zone Presence"
number:
- platform: ld2450
ld2450_id: ld2450_dev
presence_timeout:
name: "Presence Timeout"
button:
- platform: template
name: "Start Scenario"
id: start_scenario_btn
on_press:
- lambda: 'id(mock_uart).start_scenario();'
tests/integration/fixtures/uart_mock_ld2450_polygon_zones.yaml
+246
View File
@@ -0,0 +1,246 @@
esphome:
name: uart-mock-ld2450-zones
host:
api:
batch_delay: 0ms # Disable batching to receive all state updates
logger:
level: VERBOSE
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
# Dummy uart entry to satisfy ld2450's DEPENDENCIES = ["uart"]
# The actual UART bus used is the uart_mock component below
uart:
baud_rate: 115200
port: /dev/null
uart_mock:
id: mock_uart
baud_rate: 256000
auto_start: false
responses:
# Generic ACK for any command, to unblock setup commands
- expect_tx: [0x04, 0x03, 0x02, 0x01]
inject_rx:
[
0xFD, 0xFC, 0xFB, 0xFA,
0x04, 0x00,
0xFF, 0x01,
0x00, 0x00,
0x04, 0x03, 0x02, 0x01,
]
injections:
# Frame 1: target 1 at (-500, 1000), target 2 at (200, 500)
# Coordinates are sign-magnitude, with the high bit set for positive values.
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0xF4, 0x01, 0xE8, 0x83, 0x00, 0x00, 0x40, 0x01,
0xC8, 0x80, 0xF4, 0x81, 0x00, 0x00, 0x64, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
# Frame 2: target 1 at (300, 400), no other targets
- delay: 200ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x2C, 0x81, 0x90, 0x81, 0x00, 0x00, 0x64, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
# Frame 3: no targets
- delay: 200ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
# No targets, frame after frame, until presence_timeout (1 s) has cleared every zone
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
ld2450:
id: ld2450_dev
uart_id: mock_uart
text:
- platform: ld2450
ld2450_id: ld2450_dev
polygon_zones:
- id: left_zone
name: "Left Zone Polygon"
- id: near_zone
name: "Near Zone Polygon"
- id: unset_zone
name: "Unset Zone Polygon"
binary_sensor:
- platform: ld2450
ld2450_id: ld2450_dev
polygon_zones:
- polygon_zone_id: left_zone
name: "Left Zone Presence"
- polygon_zone_id: near_zone
name: "Near Zone Presence"
- polygon_zone_id: unset_zone
name: "Unset Zone Presence"
number:
- platform: ld2450
ld2450_id: ld2450_dev
presence_timeout:
name: "Presence Timeout"
button:
- platform: template
name: "Start Scenario"
id: start_scenario_btn
on_press:
- lambda: 'id(mock_uart).start_scenario();'
tests/integration/test_uart_mock_ld2450_has_target_only.py
+64
View File
@@ -0,0 +1,64 @@
"""Integration test for LD2450 has_target without the sensor and number platforms.
Tests:
1. has_target turns on when only binary sensors are configured (target coordinates are still decoded)
2. has_target holds for the default presence_timeout of 5 s, without the presence_timeout number
"""
from __future__ import annotations
import asyncio
from pathlib import Path
from aioesphomeapi import BinarySensorInfo, BinarySensorState, ButtonInfo, EntityState
import pytest
from .state_utils import InitialStateHelper, require_entity
from .types import APIClientConnectedFactory, RunCompiledFunction
# The fixture sends one frame with a target, then 2 s of frames without one
SCENARIO_S = 2.1
@pytest.mark.asyncio
async def test_uart_mock_ld2450_has_target_only(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Test has_target with no sensor platform and no presence_timeout number."""
external_components_path = str(
Path(__file__).parent / "fixtures" / "external_components"
)
yaml_config = yaml_config.replace(
"EXTERNAL_COMPONENT_PATH", external_components_path
)
async with (
run_compiled(yaml_config),
api_client_connected() as client,
):
entities, _ = await client.list_entities_services()
has_target = require_entity(entities, "has_target", BinarySensorInfo)
history: list[bool] = []
turned_on = asyncio.get_running_loop().create_future()
def on_state(state: EntityState) -> None:
if isinstance(state, BinarySensorState) and state.key == has_target.key:
history.append(state.state)
if state.state and not turned_on.done():
turned_on.set_result(None)
helper = InitialStateHelper(entities)
client.subscribe_states(helper.on_state_wrapper(on_state))
await helper.wait_for_initial_states()
start_btn = require_entity(entities, "start_scenario", ButtonInfo)
client.button_command(start_btn.key)
await asyncio.wait_for(turned_on, timeout=5.0)
await asyncio.sleep(SCENARIO_S)
assert history == [True], (
"has_target must turn on and hold through 2 s without targets, "
f"got {history}"
)
tests/integration/test_uart_mock_ld2450_polygon_zone_restore.py
+117
View File
@@ -0,0 +1,117 @@
"""Integration test for restoring LD2450 polygon zones from saved preferences.
Tests:
1. A polygon set from the API is restored after a restart
2. A saved polygon that is no longer valid is reported and leaves the zone disabled
"""
from __future__ import annotations
import asyncio
from pathlib import Path
import socket
import struct
from aioesphomeapi import TextInfo, TextState
import pytest
from .conftest import run_binary_and_wait_for_port, wait_and_connect_api_client
from .host_prefs import clear_host_prefs, read_host_prefs, write_host_prefs
from .state_utils import InitialStateHelper, require_entity
from .types import CompileFunction, ConfigWriter
DEVICE_NAME = "uart-mock-ld2450-zone-restore"
ZONE_POLYGON = "-1000,800;0,800;0,1500;-1000,1500"
# The saved Polygon starts with its points as little-endian int16 x, y pairs
SAVED_POLYGON_PREFIX = struct.pack("<hh", -1000, 800)
# Offset of Polygon::count: after MAX_POLYGON_POINTS (23) points of 4 bytes
COUNT_OFFSET = 23 * 4
@pytest.mark.asyncio
async def test_uart_mock_ld2450_polygon_zone_restore(
yaml_config: str,
write_yaml_config: ConfigWriter,
compile_esphome: CompileFunction,
reserved_tcp_port: tuple[int, socket.socket],
) -> None:
"""Test that polygon zones are restored after a restart, and invalid saved polygons are rejected."""
port, port_socket = reserved_tcp_port
external_components_path = str(
Path(__file__).parent / "fixtures" / "external_components"
)
yaml_config = yaml_config.replace(
"EXTERNAL_COMPONENT_PATH", external_components_path
)
clear_host_prefs(DEVICE_NAME)
binary_path = await compile_esphome(await write_yaml_config(yaml_config))
# Release the reserved port so the binary can bind to it
port_socket.close()
async def initial_polygon(line_callback=None) -> str:
"""Start the binary and return the zone's polygon state as first published."""
async with (
run_binary_and_wait_for_port(
binary_path, "127.0.0.1", port, line_callback=line_callback
),
wait_and_connect_api_client(port=port) as client,
):
entities, _ = await client.list_entities_services()
polygon = require_entity(entities, "restore_zone_polygon", TextInfo)
helper = InitialStateHelper(entities)
client.subscribe_states(helper.on_state_wrapper(lambda state: None))
await helper.wait_for_initial_states()
state = helper.initial_states[polygon.key]
assert isinstance(state, TextState)
return state.state
# First run: set a polygon, which is saved
async with (
run_binary_and_wait_for_port(binary_path, "127.0.0.1", port),
wait_and_connect_api_client(port=port) as client,
):
entities, _ = await client.list_entities_services()
polygon = require_entity(entities, "restore_zone_polygon", TextInfo)
loop = asyncio.get_running_loop()
published = loop.create_future()
def on_state(state: object) -> None:
if (
isinstance(state, TextState)
and state.key == polygon.key
and state.state == ZONE_POLYGON
and not published.done()
):
published.set_result(None)
helper = InitialStateHelper(entities)
client.subscribe_states(helper.on_state_wrapper(on_state))
await helper.wait_for_initial_states()
assert helper.initial_states[polygon.key].state == ""
client.text_command(polygon.key, ZONE_POLYGON)
await asyncio.wait_for(published, timeout=5.0)
# Second run: the polygon is restored
assert await initial_polygon() == ZONE_POLYGON
# Third run: a saved polygon with too few points is rejected
prefs = read_host_prefs(DEVICE_NAME)
keys = [key for key, data in prefs.items() if data.startswith(SAVED_POLYGON_PREFIX)]
assert len(keys) == 1, f"saved polygon not found in {prefs}"
data = bytearray(prefs[keys[0]])
data[COUNT_OFFSET] = 2
prefs[keys[0]] = bytes(data)
write_host_prefs(DEVICE_NAME, prefs)
warnings: list[str] = []
def capture_warning(line: str) -> None:
if "saved polygon is invalid" in line:
warnings.append(line)
assert await initial_polygon(capture_warning) == ""
assert warnings, "expected a warning about the invalid saved polygon"
clear_host_prefs(DEVICE_NAME)
tests/integration/test_uart_mock_ld2450_polygon_zone_timeout.py
+102
View File
@@ -0,0 +1,102 @@
"""Integration test for LD2450 polygon zone presence with presence_timeout.
Tests:
1. A target leaving the zone for a single radar frame does not clear presence
2. Presence clears once presence_timeout has passed since a target was last inside the zone
"""
from __future__ import annotations
import asyncio
from pathlib import Path
from aioesphomeapi import (
BinarySensorInfo,
BinarySensorState,
ButtonInfo,
EntityState,
NumberInfo,
NumberState,
TextInfo,
TextState,
)
import pytest
from .host_prefs import clear_host_prefs
from .state_utils import InitialStateHelper, require_entity
from .types import APIClientConnectedFactory, RunCompiledFunction
DEVICE_NAME = "uart-mock-ld2450-zone-timeout"
ZONE_POLYGON = "-1000,800;0,800;0,1500;-1000,1500"
PRESENCE_TIMEOUT_S = 1
@pytest.mark.asyncio
async def test_uart_mock_ld2450_polygon_zone_timeout(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Test that polygon zone presence follows presence_timeout."""
external_components_path = str(
Path(__file__).parent / "fixtures" / "external_components"
)
yaml_config = yaml_config.replace(
"EXTERNAL_COMPONENT_PATH", external_components_path
)
# The polygon and presence_timeout are saved, so a previous run would otherwise restore them
clear_host_prefs(DEVICE_NAME)
async with (
run_compiled(yaml_config),
api_client_connected() as client,
):
entities, _ = await client.list_entities_services()
polygon = require_entity(entities, "hold_zone_polygon", TextInfo)
presence = require_entity(entities, "hold_zone_presence", BinarySensorInfo)
timeout = require_entity(entities, "presence_timeout", NumberInfo)
loop = asyncio.get_running_loop()
# (time received, state) for each presence change
presence_history: list[tuple[float, bool]] = []
cleared = loop.create_future()
polygon_set = loop.create_future()
timeout_set = loop.create_future()
def on_state(state: EntityState) -> None:
if isinstance(state, BinarySensorState) and state.key == presence.key:
presence_history.append((loop.time(), state.state))
if state.state is False and not cleared.done():
cleared.set_result(None)
elif (
isinstance(state, TextState)
and state.key == polygon.key
and state.state == ZONE_POLYGON
and not polygon_set.done()
):
polygon_set.set_result(None)
elif (
isinstance(state, NumberState)
and state.key == timeout.key
and state.state == PRESENCE_TIMEOUT_S
and not timeout_set.done()
):
timeout_set.set_result(None)
initial_state_helper = InitialStateHelper(entities)
client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
await initial_state_helper.wait_for_initial_states()
client.text_command(polygon.key, ZONE_POLYGON)
client.number_command(timeout.key, PRESENCE_TIMEOUT_S)
await asyncio.wait_for(asyncio.gather(polygon_set, timeout_set), timeout=5.0)
start_btn = require_entity(entities, "start_scenario", ButtonInfo)
client.button_command(start_btn.key)
await asyncio.wait_for(cleared, timeout=5.0)
# The one-frame blip outside never reaches HA
assert [state for _, state in presence_history] == [True, False]
# The last inside frame comes 200 ms after the first, then presence_timeout must pass
(on_time, _), (off_time, _) = presence_history
assert off_time - on_time >= PRESENCE_TIMEOUT_S
tests/integration/test_uart_mock_ld2450_polygon_zones.py
+160
View File
@@ -0,0 +1,160 @@
"""Integration test for LD2450 polygon zones with mock UART.
Tests:
1. Polygons start empty and their presence sensors start unknown
2. A polygon set from the API is published back in canonical form
3. Invalid polygon text is rejected and the previous polygon is published again
4. Presence turns on and off as targets move in and out of each polygon
5. A zone without a polygon keeps presence unknown, and clearing a polygon makes it unknown again
"""
from __future__ import annotations
import asyncio
from pathlib import Path
import re
from aioesphomeapi import (
BinarySensorInfo,
BinarySensorState,
ButtonInfo,
EntityState,
NumberInfo,
NumberState,
TextInfo,
TextState,
)
import pytest
from esphome.components.ld2450.text import POLYGON_MAX_LENGTH, POLYGON_PATTERN
from .host_prefs import clear_host_prefs
from .state_utils import InitialStateHelper, StateWaiter, require_entity
from .types import APIClientConnectedFactory, RunCompiledFunction
DEVICE_NAME = "uart-mock-ld2450-zones"
LEFT_POLYGON = "-1000,800;0,800;0,1500;-1000,1500"
NEAR_POLYGON = "0,0;1000,0;1000,600;0,600"
@pytest.mark.asyncio
async def test_uart_mock_ld2450_polygon_zones(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Test LD2450 polygon zone configuration and presence detection."""
external_components_path = str(
Path(__file__).parent / "fixtures" / "external_components"
)
yaml_config = yaml_config.replace(
"EXTERNAL_COMPONENT_PATH", external_components_path
)
# Polygons are saved, so a previous run would otherwise restore them
clear_host_prefs(DEVICE_NAME)
async with (
run_compiled(yaml_config),
api_client_connected() as client,
):
entities, _ = await client.list_entities_services()
polygons = {
name: require_entity(entities, f"{name}_zone_polygon", TextInfo)
for name in ("left", "near", "unset")
}
presences = {
name: require_entity(entities, f"{name}_zone_presence", BinarySensorInfo)
for name in ("left", "near", "unset")
}
presence_names = {info.key: name for name, info in presences.items()}
# None records presence going back to unknown
presence_history: dict[str, list[bool | None]] = {
name: [] for name in presences
}
waiter = StateWaiter()
expect = waiter.expect
def on_state(state: EntityState) -> None:
if isinstance(state, BinarySensorState) and state.key in presence_names:
presence_history[presence_names[state.key]].append(
None if state.missing_state else state.state
)
waiter.on_state(state)
initial_state_helper = InitialStateHelper(entities)
client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
await initial_state_helper.wait_for_initial_states()
for name, polygon in polygons.items():
# Home Assistant validates input and the zone card finds zones with these
assert polygon.pattern == POLYGON_PATTERN
assert polygon.max_length == POLYGON_MAX_LENGTH
assert initial_state_helper.initial_states[polygon.key].state == ""
assert initial_state_helper.initial_states[
presences[name].key
].missing_state
async def set_polygon(name: str, value: str, expected: str) -> None:
key = polygons[name].key
published = expect(lambda s: isinstance(s, TextState) and s.key == key)
client.text_command(key, value)
state = await published
assert state.state == expected
assert re.fullmatch(POLYGON_PATTERN, state.state)
# Spaces and a trailing separator are accepted and removed
await set_polygon(
"left", f" {LEFT_POLYGON.replace(';', ' ; ')} ;", LEFT_POLYGON
)
await set_polygon("near", NEAR_POLYGON, NEAR_POLYGON)
# Invalid text keeps the polygon in use
await set_polygon("near", "0,0;1000,0", NEAR_POLYGON)
await set_polygon("near", "junk", NEAR_POLYGON)
# Zone presence follows presence_timeout; keep the scenario short
timeout = require_entity(entities, "presence_timeout", NumberInfo)
timeout_set = expect(
lambda s: (
isinstance(s, NumberState) and s.key == timeout.key and s.state == 1
)
)
client.number_command(timeout.key, 1)
await timeout_set
left_cleared = expect(
lambda s: (
isinstance(s, BinarySensorState)
and s.key == presences["left"].key
and s.state is False
)
)
near_cleared = expect(
lambda s: (
isinstance(s, BinarySensorState)
and s.key == presences["near"].key
and s.state is False
)
)
start_btn = require_entity(entities, "start_scenario", ButtonInfo)
client.button_command(start_btn.key)
await asyncio.gather(left_cleared, near_cleared)
# Frame 1: (-500, 1000) is in the left zone and (200, 500) is in the near zone.
# Frame 2: (300, 400) is in the near zone only; the left zone clears 1 s after frame 1.
# Frame 3 onwards: no targets; the near zone clears 1 s after frame 2.
assert presence_history["left"] == [True, False]
assert presence_history["near"] == [True, False]
assert presence_history["unset"] == []
left_unknown = expect(
lambda s: (
isinstance(s, BinarySensorState)
and s.key == presences["left"].key
and s.missing_state
)
)
await set_polygon("left", "", "")
await left_unknown
assert presence_history["left"] == [True, False, None]
tests/unit_tests/test_cpp_generator.py
+43 -2
View File
@@ -1,16 +1,19 @@
from collections.abc import Iterator
import io
import math
from pathlib import Path
import pytest
from esphome import cpp_generator as cg, cpp_types as ct
from esphome import config_validation as cv, cpp_generator as cg, cpp_types as ct
from esphome.const import (
KEY_CORE,
KEY_TARGET_PLATFORM,
PLATFORM_ESP32,
PLATFORM_ESP8266,
)
from esphome.core import CORE, ID, EsphomeError
from esphome.core import CORE, ID, LINE_DIRECTIVE_RESET, DocumentLocation, EsphomeError
from esphome.yaml_util import parse_yaml
class TestExpressions:
@@ -244,6 +247,22 @@ class TestLambdaExpression:
"[captured_var](int32_t x) -> int32_t {\n return captured_var + x;\n}"
)
def test_str__with_source_marks_end_of_body(self) -> None:
"""A lambda from the YAML marks the end of its body for the line reset."""
target = cg.LambdaExpression(
("return 42;",), (), "", int, source=DocumentLocation("test.yaml", 4, 0)
)
actual = str(target)
assert actual == (
"[]() -> int32_t {\n"
' #line 5 "test.yaml"\n'
" return 42;\n"
f" {LINE_DIRECTIVE_RESET}\n"
"}"
)
class TestCallLambda:
"""Tests for the call_lambda() function."""
@@ -735,6 +754,28 @@ class TestProcessLambda:
with pytest.raises(AssertionError):
await cg.process_lambda(lambda_obj, [(int,)])
def test_str__with_line_mark_in_body_marks_end_of_body(self) -> None:
"""A #line written into the body, as lvgl does, gets the reset too."""
target = cg.LambdaExpression(('#line 3 "test.yaml"\nfoo();',), (), "", None)
actual = str(target)
assert actual == (
f'[]() {{\n #line 3 "test.yaml"\n foo();\n {LINE_DIRECTIVE_RESET}\n}}'
)
async def test_process_lambda__twice_keeps_line(self) -> None:
"""Processing one lambda twice, as a YAML alias does, keeps its line."""
config = parse_yaml(Path("test.yaml"), io.StringIO("x: |-\n return 42;\n"))
lambda_obj = cv.lambda_(config["x"])
first = await cg.process_lambda(lambda_obj, [])
second = await cg.process_lambda(lambda_obj, [])
# The body of the block scalar is the second line, 1 counted from 0
assert first.source.line == 1
assert second.source.line == 1
@pytest.mark.asyncio
async def test_templatable__string_with_std_string_returns_flash_literal() -> None:
tests/unit_tests/test_espota2_noise.py
+10 -3
View File
@@ -858,16 +858,23 @@ def test_probe_ota_key_resolution_failure(
) -> None:
"""A name that never resolves is retried until the deadline, then reported."""
pytest.importorskip("aioesphomeapi.noise")
clock = [0.0]
def sleep(seconds: float) -> None:
clock[0] += seconds
with (
patch("esphome.espota2.resolve_ip_address", **kwargs) as resolve,
patch("time.sleep"),
patch("time.sleep", side_effect=sleep),
patch("time.monotonic", side_effect=lambda: clock[0]),
caplog.at_level(logging.WARNING),
):
assert (
espota2.probe_ota_key("nowhere.local", 3232, PSK, timeout=0.2)
espota2.probe_ota_key("nowhere.local", 3232, PSK, timeout=10)
is not ProbeResult.ACCEPTED
)
assert resolve.call_count > 1
# Looked up at 0, 3, 6 and 9 s; the last pause stops at the deadline
assert resolve.call_count == 4
assert any("failed, unreachable" in r.message for r in caplog.records)
tests/unit_tests/test_writer.py
+29 -1
View File
@@ -22,7 +22,9 @@ from esphome.const import (
PLATFORM_RP2,
PLATFORM_RTL87XX,
)
from esphome.core import EsphomeError
from esphome.core import DocumentLocation, EsphomeError, line_directive
from esphome.cpp_generator import LambdaExpression
from esphome.helpers import indent
from esphome.storage_json import StorageJSON
from esphome.writer import (
CPP_AUTO_GENERATE_BEGIN,
@@ -965,6 +967,32 @@ def test_write_cpp_creates_new_file(
assert "App.loop();" in written_content
@pytest.mark.usefixtures("mock_copy_src_tree")
@patch("esphome.writer.write_file_if_changed")
@patch("esphome.writer.CORE")
def test_write_cpp_resets_line_after_lambda(
mock_core: MagicMock,
mock_write_file: MagicMock,
tmp_path: Path,
) -> None:
"""The code after a YAML lambda is reported against main.cpp again."""
main_cpp = tmp_path / "main.cpp"
mock_core.relative_src_path.return_value = main_cpp
mock_core.cpp_global_section = ""
lambda_ = LambdaExpression(
("foo();",), (), source=DocumentLocation("test.yaml", 4, 0)
)
write_cpp(indent(f"bar({lambda_});\nbaz();") + "\n")
lines = mock_write_file.call_args[0][1].split("\n")
closing = lines.index(" });")
assert lines[closing - 3].strip() == '#line 5 "test.yaml"'
# The directive names the line after it, which is closing + 1 (1-based)
assert lines[closing - 1].strip() == line_directive(str(main_cpp), closing + 1)
assert lines[closing + 1] == " baz();"
@pytest.mark.usefixtures("mock_copy_src_tree")
@patch("esphome.writer.CORE")
def test_write_cpp_with_missing_end_marker(
tests/unit_tests/test_yaml_util.py
+18
View File
@@ -2317,3 +2317,21 @@ def test_find_secret_references_in_flow_collections() -> None:
"""Flow mapping and sequence punctuation is not part of the secret name."""
text = "a: {password: !secret wifi_pw}\nb: [!secret one, !secret 'two']\n"
assert yaml_util.find_secret_references(text) == {"wifi_pw", "one", "two"}
@pytest.mark.parametrize(
("yaml_text", "expected_offset"),
[
("value: return 42;\n", 0),
('value: "return 42;"\n', 0),
("value: |-\n return 42;\n", 1),
("value: >-\n return 42;\n", 1),
],
)
def test_content_offset_only_for_block_scalars(
yaml_text: str, expected_offset: int
) -> None:
"""A one-line lambda starts on its own line; only | and > move it to the next."""
doc = yaml_util.parse_yaml(Path("test.yaml"), io.StringIO(yaml_text))
assert doc["value"].content_offset == expected_offset
assert doc["value"].content_start.line == expected_offset